AirCoding/AirPlan/docs/spec/AirPlanV2/lib/air_runtime/test_runtime.py

"""
AirTst 测试运行器运行时 — V2 新增组件。
统一测试执行接口，支持多框架，产出结构化结果。
"""

from __future__ import annotations

import json
import subprocess
from dataclasses import dataclass, field
from pathlib import Path

from air_runtime.io import atomic_json_write
from air_runtime.paths import tst_state_path


@dataclass
class TestCase:
    name: str
    status: str  # passed | failed | skipped
    duration: str = ""
    message: str = ""


@dataclass
class TestSuite:
    name: str
    total: int = 0
    passed: int = 0
    failed: int = 0
    skipped: int = 0
    cases: list[TestCase] = field(default_factory=list)


@dataclass
class TestRunResult:
    framework: str
    total: int = 0
    passed: int = 0
    failed: int = 0
    disabled: int = 0
    duration: str = ""
    suites: list[TestSuite] = field(default_factory=list)
    failures: list[dict] = field(default_factory=list)


class TestRunner:
    """统一测试执行器。"""

    FRAMEWORKS = {
        "pytest": ["python", "-m", "pytest", "--json-report", "-q"],
        "googletest": ["ctest", "--output-on-failure"],
        "jest": ["npx", "jest", "--json"],
        "vitest": ["npx", "vitest", "run", "--reporter=json"],
        "go": ["go", "test", "-json", "./..."],
        "cargo": ["cargo", "test", "--", "--format=json"],
    }

    def run(self, task_id: str, project_root: Path, framework: str,
            target_path: Path | None = None,
            extra_args: list[str] | None = None) -> TestRunResult:
        if framework not in self.FRAMEWORKS:
            return TestRunResult(framework=framework, failures=[{"error": f"unsupported framework: {framework}"}])

        cmd = list(self.FRAMEWORKS[framework])
        if target_path:
            cmd.append(str(target_path))
        if extra_args:
            cmd.extend(extra_args)

        try:
            result = subprocess.run(cmd, capture_output=True, text=True,
                                   timeout=600, cwd=str(project_root))
        except subprocess.TimeoutExpired:
            return TestRunResult(framework=framework, failures=[{"error": "timeout"}])

        run_result = self._parse_result(framework, result.stdout)
        self._save_report(task_id, project_root, run_result)
        return run_result

    def _parse_result(self, framework: str, stdout: str) -> TestRunResult:
        if framework == "pytest":
            return self._parse_pytest(stdout)
        if framework in ("jest", "vitest"):
            return self._parse_jest(stdout)
        if framework == "googletest":
            return self._parse_googletest(stdout)
        if framework == "go":
            return self._parse_go(stdout)
        if framework == "cargo":
            return self._parse_cargo(stdout)
        return TestRunResult(framework=framework, total=0)

    def _parse_pytest(self, stdout: str) -> TestRunResult:
        try:
            data = json.loads(stdout)
        except json.JSONDecodeError:
            return TestRunResult(framework="pytest", failures=[{"error": "json parse failed"}])
        return TestRunResult(
            framework="pytest",
            total=data.get("summary", {}).get("total", 0),
            passed=data.get("summary", {}).get("passed", 0),
            failed=data.get("summary", {}).get("failed", 0),
            duration=str(data.get("duration", "")),
        )

    def _parse_jest(self, stdout: str) -> TestRunResult:
        try:
            data = json.loads(stdout)
        except json.JSONDecodeError:
            return TestRunResult(framework="jest", failures=[{"error": "json parse failed"}])
        return TestRunResult(
            framework="jest",
            total=data.get("numTotalTests", 0),
            passed=data.get("numPassedTests", 0),
            failed=data.get("numFailedTests", 0),
        )

    def _parse_googletest(self, stdout: str) -> TestRunResult:
        """解析 ctest 输出。ctest 不输出 JSON，从文本提取统计。"""
        import re
        total = passed = failed = disabled = 0
        for line in stdout.splitlines():
            m = re.match(r"(\d+)% tests passed, (\d+) tests failed out of (\d+)", line)
            if m:
                failed = int(m.group(2))
                total = int(m.group(3))
                passed = total - failed
        # GoogleTest 也支持 --gtest_output=json
        try:
            data = json.loads(stdout)
            if isinstance(data, dict):
                total = sum(s.get("tests", 0) for s in data.get("testsuites", []))
                failed = sum(s.get("failures", 0) for s in data.get("testsuites", []))
                passed = total - failed
                disabled = sum(s.get("disabled", 0) for s in data.get("testsuites", []))
        except json.JSONDecodeError:
            pass
        return TestRunResult(
            framework="googletest", total=total, passed=passed, failed=failed, disabled=disabled,
        )

    def _parse_go(self, stdout: str) -> TestRunResult:
        """解析 go test -json 输出（JSONL 格式，每行一个事件）。"""
        total = passed = failed = 0
        for line in stdout.splitlines():
            try:
                ev = json.loads(line)
            except json.JSONDecodeError:
                continue
            action = ev.get("Action", "")
            if action == "pass":
                passed += 1
                total += 1
            elif action == "fail":
                failed += 1
                total += 1
            elif action == "skip":
                total += 1
        return TestRunResult(framework="go", total=total, passed=passed, failed=failed)

    def _parse_cargo(self, stdout: str) -> TestRunResult:
        """解析 cargo test --format=json 输出（JSONL 格式）。"""
        total = passed = failed = 0
        for line in stdout.splitlines():
            try:
                ev = json.loads(line)
            except json.JSONDecodeError:
                continue
            if ev.get("type") == "test":
                total += 1
                if ev.get("event") == "ok":
                    passed += 1
                elif ev.get("event") == "failed":
                    failed += 1
        return TestRunResult(framework="cargo", total=total, passed=passed, failed=failed)

    def _save_report(self, task_id: str, project_root: Path, result: TestRunResult) -> None:
        report_dir = tst_state_path(project_root).parent / "reports"
        report_dir.mkdir(parents=True, exist_ok=True)
        from air_runtime.utils import session_stamp
        report_path = report_dir / f"{task_id}-{session_stamp()}.json"
        atomic_json_write(report_path, result.__dict__)