ligbox-ops-platform/projects/ops-desk/api/app/vm123/openpanel_test.py

"""Teste de confirmação — OpenPanel API multidomínio (Spec 028)."""

from __future__ import annotations

import random
import re
import string
import time
from typing import Any

from app.vm123 import openpanel_client

TEST_PASSWORD = "LbOpenTest805353"
DEFAULT_PLAN = "ligbox-site-cms"
USER_RE = re.compile(r"^[a-z][a-z0-9]{2,15}$")


def _step(name: str, ok: bool, detail: str = "", extra: dict | None = None) -> dict[str, Any]:
    return {"name": name, "ok": ok, "detail": detail, **(extra or {})}


def _default_accounts() -> list[tuple[str, str]]:
    suffix = "".join(random.choices(string.digits, k=5))
    user1 = f"apitest{suffix}"
    user2 = f"apitestb{suffix}"
    return [
        (user1, f"apitest{suffix}.ligbox.com.br"),
        (user2, f"apitestb{suffix}.ligbox.com.br"),
    ]


def _normalize_accounts(
    accounts: list[dict[str, str]] | None,
    *,
    auto_names: bool = True,
) -> list[tuple[str, str]]:
    out: list[tuple[str, str]] = []
    if accounts:
        for row in accounts:
            username = (row.get("username") or "").strip().lower()
            domain = (row.get("domain") or "").strip().lower()
            if not username and not domain:
                continue
            if not username and domain:
                username = re.sub(r"[^a-z0-9]", "", domain.split(".")[0])[:15]
            if not domain and username:
                domain = f"{username}.ligbox.com.br" if "." not in username else username
            if not USER_RE.fullmatch(username):
                raise ValueError(f"username inválido: {username!r} (a-z, 3-16 chars)")
            out.append((username, domain))
    if not out and auto_names:
        return _default_accounts()
    if not out:
        raise ValueError("informe pelo menos uma conta/domínio ou active auto_names")
    return out[:5]


def run_confirmation_test(
    *,
    triggered_by: str = "api",
    accounts: list[dict[str, str]] | None = None,
    password: str = TEST_PASSWORD,
    cleanup: bool = True,
    auto_names: bool = True,
    check_reference: bool = True,
) -> dict[str, Any]:
    """Suite E2E: health → list → N contas/domínios → cleanup opcional."""
    started = time.time()
    steps: list[dict[str, Any]] = []
    try:
        pairs = _normalize_accounts(accounts, auto_names=auto_names)
    except ValueError as exc:
        return {
            "ok": False,
            "suite": "openpanel-multidomain-api-confirm",
            "message": str(exc),
            "steps": [_step("validate_input", False, str(exc))],
            "steps_passed": 0,
            "steps_total": 1,
            "triggered_by": triggered_by,
        }

    health = openpanel_client.health()
    steps.append(
        _step(
            "bridge_health",
            health.get("ok") is True,
            f"bridge={health.get('bridge')} url={health.get('bridge_url')}",
            {"response": health},
        )
    )
    if not health.get("ok"):
        return _result(False, steps, pairs[0][0], triggered_by, started, pairs=pairs)

    try:
        listed = openpanel_client.list_users()
        users = (listed.get("users") or {}).get("data") or []
        steps.append(_step("list_users", True, f"{len(users)} contas", {"count": len(users)}))
    except openpanel_client.OpenPanelBridgeError as exc:
        steps.append(_step("list_users", False, str(exc)))
        return _result(False, steps, pairs[0][0], triggered_by, started, pairs=pairs)

    if check_reference:
        try:
            ref = openpanel_client.get_user("diarissima")
            steps.append(
                _step(
                    "get_reference_user",
                    ref.get("success") is True,
                    (ref.get("domains") or "")[:80],
                )
            )
        except openpanel_client.OpenPanelBridgeError as exc:
            steps.append(_step("get_reference_user", False, str(exc)))

    created: list[str] = []
    for idx, (username, domain) in enumerate(pairs, start=1):
        label = f"provision_{username}"
        try:
            prov = openpanel_client.provision_user(
                username=username,
                password=password,
                email=f"hosting@{domain}",
                domain=domain,
                plan_name=DEFAULT_PLAN,
            )
            ok = prov.get("success") is True
            if ok:
                created.append(username)
            steps.append(
                _step(
                    label,
                    ok,
                    prov.get("response", {}).get("message", "")[:120],
                    {"username": username, "domain": domain, "index": idx},
                )
            )
        except openpanel_client.OpenPanelBridgeError as exc:
            steps.append(_step(label, False, str(exc), {"username": username, "domain": domain}))

    try:
        listed2 = openpanel_client.list_users()
        names = {u.get("username") for u in (listed2.get("users") or {}).get("data") or []}
        expected = {u for u, _ in pairs}
        found = expected & names
        steps.append(
            _step(
                "verify_multidomain_platform",
                found == expected,
                f"encontrados {len(found)}/{len(expected)}: {', '.join(sorted(found))}",
                {"users_found": sorted(found), "users_expected": sorted(expected)},
            )
        )
    except openpanel_client.OpenPanelBridgeError as exc:
        steps.append(_step("verify_multidomain_platform", False, str(exc)))

    deleted = 0
    if cleanup:
        for username in created:
            try:
                openpanel_client.delete_user(username)
                deleted += 1
                steps.append(_step(f"cleanup_{username}", True, "removido"))
            except openpanel_client.OpenPanelBridgeError as exc:
                steps.append(_step(f"cleanup_{username}", False, str(exc)))
    else:
        steps.append(_step("cleanup_skipped", True, f"mantidas: {', '.join(created)}"))

    all_ok = all(s["ok"] for s in steps)
    cleanup_ok = (not cleanup) or (deleted == len(created))
    return _result(
        all_ok and cleanup_ok,
        steps,
        pairs[0][0],
        triggered_by,
        started,
        pairs=pairs,
        cleanup=cleanup_ok,
    )


def _result(
    ok: bool,
    steps: list[dict[str, Any]],
    username: str,
    triggered_by: str,
    started: float,
    *,
    pairs: list[tuple[str, str]] | None = None,
    cleanup: bool = True,
) -> dict[str, Any]:
    passed = sum(1 for s in steps if s["ok"])
    return {
        "ok": ok,
        "suite": "openpanel-multidomain-api-confirm",
        "spec": "028-openpanel-ce-ligbox-reengineering",
        "triggered_by": triggered_by,
        "test_user": username,
        "accounts_tested": [{"username": u, "domain": d} for u, d in (pairs or [])],
        "cleanup_done": cleanup,
        "steps_passed": passed,
        "steps_total": len(steps),
        "steps": steps,
        "duration_sec": round(time.time() - started, 2),
        "message": (
            "OpenPanel via API multidomínio Ligbox Re-engenharia — CONFIRMADO"
            if ok
            else "Falha em um ou mais passos — ver steps"
        ),
    }