Nexus/scripts/run_nexus_acceptance.py

#!/usr/bin/env python3
"""
Nexus 全功能自动化验收 — 统一入口

自动执行 API 端到端、扩展 API、可选单元测试与 Playwright UI，按功能模块汇总 PASS/FAIL/SKIP。

用法:
  python scripts/run_nexus_acceptance.py
  python scripts/run_nexus_acceptance.py --base http://127.0.0.1:8600
  python scripts/run_nexus_acceptance.py --with-ui
  python scripts/run_nexus_acceptance.py --with-unit
  python scripts/run_nexus_acceptance.py --json reports/acceptance.json

环境变量（与 tests/test_api.py 相同）:
  NEXUS_TEST_BASE              默认 http://127.0.0.1:8600
  NEXUS_TEST_ADMIN_USER        默认 admin
  NEXUS_TEST_ADMIN_PASSWORD    可从仓库根 .env 读取
  NEXUS_E2E_BASE / NEXUS_E2E_USER / NEXUS_E2E_PASSWORD  （--with-ui 时）
"""

from __future__ import annotations

import argparse
import json
import os
import re
import subprocess
import sys
import time
import urllib.error
import urllib.request
from dataclasses import asdict, dataclass, field
from datetime import datetime, timezone
from pathlib import Path

REPO_ROOT = Path(__file__).resolve().parents[1]
TESTS_DIR = REPO_ROOT / "tests"
sys.path.insert(0, str(TESTS_DIR))

from acceptance_catalog import FEATURES, match_feature  # noqa: E402

PASS_RE = re.compile(r"\[PASS\]\s*(.+)")
FAIL_RE = re.compile(r"\[FAIL\]\s*(.+)")


def _safe_console(text: str) -> str:
    """Avoid UnicodeEncodeError on Windows GBK consoles."""
    return text.encode("ascii", errors="replace").decode("ascii")


@dataclass
class CheckRecord:
    name: str
    status: str  # pass | fail | skip
    feature_id: str
    layer: str
    message: str = ""
    duration_ms: float = 0


@dataclass
class FeatureRollup:
    id: str
    name: str
    layer: str
    description: str
    status: str = "skip"  # pass | fail | skip | warn
    passed: int = 0
    failed: int = 0
    skipped: int = 0
    checks: list[CheckRecord] = field(default_factory=list)


@dataclass
class AcceptanceReport:
    started_at: str
    base_url: str
    phases: dict[str, dict]
    features: list[dict]
    summary: dict


def _load_env_from_dotenv() -> None:
    env_path = REPO_ROOT / ".env"
    if not env_path.is_file():
        return
    for line in env_path.read_text(encoding="utf-8", errors="replace").splitlines():
        line = line.strip()
        if not line or line.startswith("#") or "=" not in line:
            continue
        key, _, value = line.partition("=")
        key, value = key.strip(), value.strip().strip("\"'")
        if key and key not in os.environ:
            os.environ[key] = value


def _run_subprocess_script(
    script_rel: str,
    env: dict[str, str],
    timeout: int = 600,
) -> tuple[int, str, str, float]:
    script = REPO_ROOT / script_rel
    t0 = time.perf_counter()
    proc = subprocess.run(
        [sys.executable, str(script)],
        cwd=str(REPO_ROOT),
        env=env,
        capture_output=True,
        text=True,
        timeout=timeout,
        encoding="utf-8",
        errors="replace",
    )
    elapsed = (time.perf_counter() - t0) * 1000
    out = (proc.stdout or "") + (proc.stderr or "")
    return proc.returncode, out, proc.stderr or "", elapsed


def _parse_api_output(output: str, layer: str = "api") -> list[CheckRecord]:
    records: list[CheckRecord] = []
    for line in output.splitlines():
        m_pass = PASS_RE.search(line)
        m_fail = FAIL_RE.search(line)
        if m_pass:
            name = m_pass.group(1).strip()
            fid = match_feature(name)
            records.append(CheckRecord(name, "pass", fid, layer))
        elif m_fail:
            name = m_fail.group(1).strip()
            fid = match_feature(name)
            msg = line.split("—", 1)[-1].strip() if "—" in line else ""
            records.append(CheckRecord(name, "fail", fid, layer, message=msg))
    return records


def _http_json(
    method: str,
    url: str,
    body: dict | None = None,
    headers: dict | None = None,
    timeout: int = 30,
) -> tuple[int, str]:
    data = json.dumps(body).encode() if body is not None else None
    req = urllib.request.Request(url, data=data, method=method)
    req.add_header("Content-Type", "application/json")
    if headers:
        for k, v in headers.items():
            req.add_header(k, v)
    try:
        with urllib.request.urlopen(req, timeout=timeout) as resp:
            return resp.getcode(), resp.read().decode("utf-8", errors="replace")
    except urllib.error.HTTPError as e:
        return e.code, e.read().decode("utf-8", errors="replace")


def _extra_api_checks(base: str, token: str) -> list[CheckRecord]:
    """补充 test_full_site_api 未覆盖或需单独断言的检查。"""
    records: list[CheckRecord] = []
    auth = {"Authorization": f"Bearer {token}"} if token else {}

    # 静态 SPA
    t0 = time.perf_counter()
    try:
        req = urllib.request.Request(f"{base.rstrip('/')}/app/")
        with urllib.request.urlopen(req, timeout=15) as resp:
            html = resp.read().decode("utf-8", errors="replace")[:8000]
            ok = resp.getcode() == 200 and ("id=\"app\"" in html or "id='app'" in html or "/assets/" in html)
        records.append(
            CheckRecord(
                "GET /app/ (SPA 入口)",
                "pass" if ok else "fail",
                "infra_app",
                "infra",
                message="" if ok else "缺少 #app 或 assets",
                duration_ms=(time.perf_counter() - t0) * 1000,
            )
        )
    except Exception as ex:
        records.append(CheckRecord("GET /app/", "fail", "infra_app", "infra", message=str(ex)))

    # 登录后探测第一台服务器的 files-capability
    code, raw = _http_json("GET", f"{base}/api/servers/?per_page=5", headers=auth)
    first_id = None
    if code == 200:
        try:
            items = json.loads(raw).get("items") or []
            if items:
                first_id = items[0].get("id")
        except json.JSONDecodeError:
            pass
    if first_id:
        t0 = time.perf_counter()
        code2, raw2 = _http_json(
            "GET",
            f"{base}/api/servers/{first_id}/files-capability",
            headers=auth,
        )
        ok = code2 == 200 and "ssh_user" in raw2
        records.append(
            CheckRecord(
                f"GET /api/servers/{first_id}/files-capability",
                "pass" if ok else "fail",
                "files",
                "api",
                message="" if ok else f"HTTP {code2}",
                duration_ms=(time.perf_counter() - t0) * 1000,
            )
        )
        t0 = time.perf_counter()
        code3, _ = _http_json(
            "POST",
            f"{base}/api/sync/browse",
            body={"server_id": first_id, "path": "/tmp"},
            headers=auth,
        )
        records.append(
            CheckRecord(
                "POST /api/sync/browse /tmp",
                "pass" if code3 == 200 else "fail",
                "files",
                "api",
                message=f"HTTP {code3}" if code3 != 200 else "",
                duration_ms=(time.perf_counter() - t0) * 1000,
            )
        )
    else:
        records.append(
            CheckRecord(
                "files-capability & browse (skip)",
                "skip",
                "files",
                "api",
                message="无可用服务器，跳过 SSH 相关检查",
            )
        )

    return records


def _run_pytest_unit(timeout: int = 300) -> tuple[list[CheckRecord], int]:
    """运行不依赖运行中后端的单元测试子集。"""
    patterns = [
        "test_posix_paths.py",
        "test_remote_path_validation.py",
        "test_unix_ls.py",
        "test_files_elevation.py",
        "test_schema_path_validators.py",
        "test_file_permissions.py",
        "test_security_unit.py",
    ]
    paths = [str(TESTS_DIR / p) for p in patterns if (TESTS_DIR / p).is_file()]
    if not paths:
        return [CheckRecord("pytest unit", "skip", "unit_core", "unit", message="无单元测试文件")], 0

    t0 = time.perf_counter()
    proc = subprocess.run(
        [sys.executable, "-m", "pytest", *paths, "-q", "--tb=no"],
        cwd=str(REPO_ROOT),
        capture_output=True,
        text=True,
        timeout=timeout,
        encoding="utf-8",
        errors="replace",
    )
    elapsed = (time.perf_counter() - t0) * 1000
    out = (proc.stdout or "") + (proc.stderr or "")
    # pytest -q: "12 passed" or "1 failed, 11 passed"
    m = re.search(r"(\d+)\s+passed", out)
    passed_n = int(m.group(1)) if m else 0
    m_fail = re.search(r"(\d+)\s+failed", out)
    failed_n = int(m_fail.group(1)) if m_fail else 0
    status = "pass" if proc.returncode == 0 else "fail"
    return [
        CheckRecord(
            f"pytest unit ({passed_n} passed, {failed_n} failed)",
            status,
            "unit_core",
            "unit",
            message=out.strip()[-500:] if status == "fail" else "",
            duration_ms=elapsed,
        )
    ], proc.returncode


def _run_playwright(env: dict[str, str], timeout: int = 900) -> tuple[list[CheckRecord], int]:
    frontend = REPO_ROOT / "frontend"
    spec = frontend / "e2e" / "full-acceptance.spec.mjs"
    if not spec.is_file():
        return [CheckRecord("playwright", "skip", "ui_dashboard", "ui", message="缺少 e2e spec")], 0
    if not (frontend / "node_modules").is_dir():
        return [
            CheckRecord(
                "playwright",
                "skip",
                "ui_dashboard",
                "ui",
                message="请先在 frontend/ 执行 npm install",
            )
        ], 0
    if not env.get("NEXUS_E2E_PASSWORD"):
        return [
            CheckRecord(
                "playwright UI",
                "skip",
                "ui_dashboard",
                "ui",
                message="未设置 NEXUS_E2E_PASSWORD",
            )
        ], 0

    t0 = time.perf_counter()
    proc = subprocess.run(
        [
            "npx",
            "playwright",
            "test",
            "e2e/full-acceptance.spec.mjs",
            "--reporter=line",
        ],
        cwd=str(frontend),
        env=env,
        capture_output=True,
        text=True,
        timeout=timeout,
        encoding="utf-8",
        errors="replace",
        shell=os.name == "nt",
    )
    elapsed = (time.perf_counter() - t0) * 1000
    out = (proc.stdout or "") + (proc.stderr or "")
    records: list[CheckRecord] = []
    # Playwright line reporter: "  ok 1 [chromium] › ..."
    for line in out.splitlines():
        if " ok " in line and "›" in line:
            title = line.split("›", 1)[-1].strip()
            fid = match_feature(title)
            records.append(CheckRecord(title, "pass", fid, "ui"))
        elif (" x " in line or "✘" in line) and "›" in line:
            title = line.split("›", 1)[-1].strip()
            fid = match_feature(title)
            records.append(CheckRecord(title, "fail", fid, "ui", message=line.strip()))
    if not records:
        records.append(
            CheckRecord(
                "playwright suite",
                "pass" if proc.returncode == 0 else "fail",
                "ui_dashboard",
                "ui",
                message=out[-400:] if proc.returncode != 0 else "",
                duration_ms=elapsed,
            )
        )
    return records, proc.returncode


def _rollup_features(all_checks: list[CheckRecord]) -> list[FeatureRollup]:
    by_id: dict[str, FeatureRollup] = {}
    spec_map = {s.id: s for s in FEATURES}
    for spec in FEATURES:
        by_id[spec.id] = FeatureRollup(
            spec.id, spec.name, spec.layer, spec.description,
        )

    for chk in all_checks:
        fid = chk.feature_id if chk.feature_id in by_id else "other"
        if fid not in by_id:
            by_id[fid] = FeatureRollup(fid, fid, chk.layer, "")
        roll = by_id[fid]
        roll.checks.append(chk)
        if chk.status == "pass":
            roll.passed += 1
        elif chk.status == "fail":
            roll.failed += 1
        else:
            roll.skipped += 1

    for roll in by_id.values():
        if roll.failed > 0:
            roll.status = "fail"
        elif roll.passed > 0:
            roll.status = "pass"
        elif roll.skipped > 0:
            roll.status = "skip"
        else:
            roll.status = "skip"

    order = {s.id: i for i, s in enumerate(FEATURES)}
    return sorted(
        by_id.values(),
        key=lambda r: (order.get(r.id, 999), r.id),
    )


def _print_report(rollups: list[FeatureRollup], summary: dict) -> None:
    # ASCII 标记，避免 Windows GBK 控制台 UnicodeEncodeError
    icons = {"pass": "[OK]", "fail": "[FAIL]", "skip": "[SKIP]", "warn": "[WARN]"}
    print("\n" + "=" * 60)
    print("Nexus 功能验收报告")
    print("=" * 60)
    for roll in rollups:
        if roll.status == "skip" and roll.passed == 0 and roll.failed == 0:
            continue
        icon = icons.get(roll.status, "?")
        detail = f" ({roll.passed} 通过"
        if roll.failed:
            detail += f", {roll.failed} 失败"
        if roll.skipped:
            detail += f", {roll.skipped} 跳过"
        detail += ")"
        print(_safe_console(f"  {icon} [{roll.layer:5}] {roll.name}{detail}"))
        for c in roll.checks:
            if c.status == "fail":
                msg = _safe_console((c.message or "")[:120])
                print(_safe_console(f"         - {c.name}: {msg}"))
    print("-" * 60)
    print(
        f"合计: {summary['passed_checks']} 通过, "
        f"{summary['failed_checks']} 失败, "
        f"{summary['skipped_checks']} 跳过 | "
        f"功能模块 {summary['features_pass']}/{summary['features_total']} 正常"
    )
    print("=" * 60)


def main() -> int:
    parser = argparse.ArgumentParser(description="Nexus 全功能自动化验收")
    parser.add_argument("--base", default=None, help="API 基址，默认 NEXUS_TEST_BASE")
    parser.add_argument("--with-ui", action="store_true", help="运行 Playwright UI 验收（需 Node）")
    parser.add_argument("--with-unit", action="store_true", help="运行 pytest 单元测试子集")
    parser.add_argument("--skip-api", action="store_true", help="跳过 test_api / test_full_site_api")
    parser.add_argument("--json", dest="json_path", default=None, help="写入 JSON 报告路径")
    args = parser.parse_args()

    _load_env_from_dotenv()
    base = (args.base or os.environ.get("NEXUS_TEST_BASE", "http://127.0.0.1:8600")).rstrip("/")
    env = os.environ.copy()
    env["NEXUS_TEST_BASE"] = base

    started = datetime.now(timezone.utc).isoformat()
    all_checks: list[CheckRecord] = []
    phases: dict[str, dict] = {}

    print("=" * 60)
    print("Nexus 全功能自动化验收")
    print(f"  目标: {base}")
    print(f"  时间: {started}")
    print("=" * 60)

    exit_codes: list[int] = []

    if not args.skip_api:
        for label, script in (
            ("core_api", "tests/test_api.py"),
            ("extended_api", "tests/test_full_site_api.py"),
        ):
            print(f"\n>>> 阶段: {label} ({script})")
            code, out, err, ms = _run_subprocess_script(script, env)
            checks = _parse_api_output(out)
            all_checks.extend(checks)
            phases[label] = {
                "exit_code": code,
                "checks": len(checks),
                "duration_ms": ms,
                "stdout_tail": out[-1500:] if code != 0 else "",
            }
            exit_codes.append(code)
            if err and code != 0:
                print(err[-500:])

        # 登录拿 token 做补充检查
        import test_api as api_base  # noqa: E402

        api_base._load_credentials_from_env_file()
        api_base.BASE = base
        login = api_base.api_test(
            "acceptance-login",
            "POST",
            "/api/auth/login",
            body={
                "username": api_base.ADMIN_USER,
                "password": api_base.ADMIN_PASSWORD,
            },
            expect_code=200,
        )
        token = (login or {}).get("access_token", "")
        if token:
            print("\n>>> 阶段: extra_api")
            extra = _extra_api_checks(base, token)
            all_checks.extend(extra)
            phases["extra_api"] = {"checks": len(extra)}
        else:
            print("  [WARN] 登录失败，跳过 extra_api（请配置 NEXUS_TEST_ADMIN_PASSWORD）")
            all_checks.append(
                CheckRecord("extra_api", "skip", "auth", "api", message="登录失败"),
            )

    if args.with_unit:
        print("\n>>> 阶段: unit_tests")
        unit_checks, code = _run_pytest_unit()
        all_checks.extend(unit_checks)
        phases["unit_tests"] = {"exit_code": code}
        exit_codes.append(code)

    if args.with_ui:
        print("\n>>> 阶段: playwright_ui")
        ui_env = env.copy()
        ui_env.setdefault("NEXUS_E2E_BASE", base)
        ui_env.setdefault("NEXUS_E2E_USER", os.environ.get("NEXUS_TEST_ADMIN_USER", "admin"))
        if not ui_env.get("NEXUS_E2E_PASSWORD"):
            ui_env["NEXUS_E2E_PASSWORD"] = os.environ.get("NEXUS_TEST_ADMIN_PASSWORD", "")
        ui_checks, code = _run_playwright(ui_env)
        all_checks.extend(ui_checks)
        phases["playwright_ui"] = {"exit_code": code, "checks": len(ui_checks)}
        exit_codes.append(code)

    rollups = _rollup_features(all_checks)
    passed = sum(1 for c in all_checks if c.status == "pass")
    failed = sum(1 for c in all_checks if c.status == "fail")
    skipped = sum(1 for c in all_checks if c.status == "skip")
    features_pass = sum(1 for r in rollups if r.status == "pass")
    features_fail = sum(1 for r in rollups if r.status == "fail")
    features_total = sum(
        1 for r in rollups if r.status in ("pass", "fail") and (r.passed or r.failed)
    )

    summary = {
        "passed_checks": passed,
        "failed_checks": failed,
        "skipped_checks": skipped,
        "features_pass": features_pass,
        "features_fail": features_fail,
        "features_total": features_total,
        "ok": failed == 0 and all(ec == 0 for ec in exit_codes),
    }

    def _rollup_dict(r: FeatureRollup) -> dict:
        d = asdict(r)
        d["checks"] = [asdict(c) for c in r.checks]
        return d

    report = AcceptanceReport(
        started_at=started,
        base_url=base,
        phases=phases,
        features=[_rollup_dict(r) for r in rollups],
        summary=summary,
    )

    _print_report(rollups, summary)

    if args.json_path:
        out_path = Path(args.json_path)
        out_path.parent.mkdir(parents=True, exist_ok=True)
        payload = asdict(report)
        payload["checks"] = [asdict(c) for c in all_checks]
        out_path.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8")
        print(f"\nJSON 报告: {out_path}")

    if not summary["ok"]:
        print("\n验收未通过：请根据上方 [FAIL] 项修复后重试。")
        return 1
    print("\n验收通过：各已测功能表现正常。")
    return 0


if __name__ == "__main__":
    sys.exit(main())