breakpilot-pwa/backend/api/tests/runners/go_runner.py

"""
Go Test Runner

Fuehrt Go-Tests aus und parsed die Ergebnisse.
"""

import subprocess
import json
from datetime import datetime
from pathlib import Path
from typing import Dict, List, Optional
from dataclasses import dataclass, field


@dataclass
class GoTestResult:
    """Ergebnis eines einzelnen Go-Tests"""
    package: str
    test_name: str
    passed: bool
    duration_seconds: float
    output: str = ""


@dataclass
class GoTestSummary:
    """Zusammenfassung eines Go-Test-Runs"""
    total: int = 0
    passed: int = 0
    failed: int = 0
    skipped: int = 0
    duration_seconds: float = 0.0
    coverage_percent: Optional[float] = None
    results: List[GoTestResult] = field(default_factory=list)
    raw_output: str = ""


class GoTestRunner:
    """
    Runner fuer Go-Tests.

    Verwendet `go test -json` fuer strukturierte Ausgabe.
    """

    def __init__(self, base_path: Path):
        self.base_path = base_path

    async def run(self, with_coverage: bool = True, timeout: int = 300) -> GoTestSummary:
        """
        Fuehrt Go-Tests aus.

        Args:
            with_coverage: Coverage erfassen
            timeout: Timeout in Sekunden

        Returns:
            GoTestSummary mit allen Ergebnissen
        """
        if not self.base_path.exists():
            return GoTestSummary(raw_output="Pfad existiert nicht")

        cmd = ["go", "test", "-v", "-json"]
        if with_coverage:
            cmd.extend(["-cover", "-coverprofile=coverage.out"])
        cmd.append("./...")

        try:
            result = subprocess.run(
                cmd,
                cwd=str(self.base_path),
                capture_output=True,
                text=True,
                timeout=timeout,
            )

            return self._parse_output(result.stdout, result.stderr)

        except subprocess.TimeoutExpired:
            return GoTestSummary(raw_output=f"Timeout nach {timeout} Sekunden")
        except FileNotFoundError:
            return GoTestSummary(raw_output="Go nicht installiert")
        except Exception as e:
            return GoTestSummary(raw_output=str(e))

    def _parse_output(self, stdout: str, stderr: str) -> GoTestSummary:
        """Parsed die JSON-Ausgabe von go test"""
        summary = GoTestSummary(raw_output=stdout[:10000] if stdout else stderr[:10000])

        current_test: Dict[str, str] = {}
        test_outputs: Dict[str, List[str]] = {}

        for line in stdout.split("\n"):
            if not line.strip():
                continue

            try:
                event = json.loads(line)
                action = event.get("Action")
                package = event.get("Package", "")
                test = event.get("Test", "")
                elapsed = event.get("Elapsed", 0)
                output = event.get("Output", "")

                # Test-Output sammeln
                if test and output:
                    key = f"{package}:{test}"
                    if key not in test_outputs:
                        test_outputs[key] = []
                    test_outputs[key].append(output)

                # Test-Ergebnis
                if action == "pass" and test:
                    summary.passed += 1
                    summary.total += 1
                    summary.results.append(GoTestResult(
                        package=package,
                        test_name=test,
                        passed=True,
                        duration_seconds=elapsed,
                        output="".join(test_outputs.get(f"{package}:{test}", [])),
                    ))

                elif action == "fail" and test:
                    summary.failed += 1
                    summary.total += 1
                    summary.results.append(GoTestResult(
                        package=package,
                        test_name=test,
                        passed=False,
                        duration_seconds=elapsed,
                        output="".join(test_outputs.get(f"{package}:{test}", [])),
                    ))

                elif action == "skip" and test:
                    summary.skipped += 1
                    summary.total += 1

                # Package-Ergebnis (Gesamtdauer)
                elif action in ["pass", "fail"] and not test and elapsed:
                    summary.duration_seconds = max(summary.duration_seconds, elapsed)

            except json.JSONDecodeError:
                # Nicht-JSON-Zeilen ignorieren (z.B. Coverage-Output)
                if "coverage:" in line.lower():
                    # z.B. "coverage: 75.2% of statements"
                    try:
                        parts = line.split("coverage:")
                        if len(parts) > 1:
                            percent_str = parts[1].strip().split("%")[0]
                            summary.coverage_percent = float(percent_str)
                    except (ValueError, IndexError):
                        pass

        return summary

    async def run_single_test(self, test_name: str, timeout: int = 60) -> GoTestResult:
        """
        Fuehrt einen einzelnen Test aus.

        Args:
            test_name: Name des Tests (z.B. "TestMyFunction")
            timeout: Timeout in Sekunden

        Returns:
            GoTestResult fuer den spezifischen Test
        """
        cmd = ["go", "test", "-v", "-run", test_name, "./..."]

        try:
            result = subprocess.run(
                cmd,
                cwd=str(self.base_path),
                capture_output=True,
                text=True,
                timeout=timeout,
            )

            passed = "PASS" in result.stdout
            return GoTestResult(
                package=str(self.base_path),
                test_name=test_name,
                passed=passed,
                duration_seconds=0.0,
                output=result.stdout + result.stderr,
            )

        except Exception as e:
            return GoTestResult(
                package=str(self.base_path),
                test_name=test_name,
                passed=False,
                duration_seconds=0.0,
                output=str(e),
            )

    async def get_coverage_report(self) -> Optional[Dict]:
        """
        Liest den Coverage-Bericht.

        Returns:
            Dict mit Coverage-Details oder None
        """
        coverage_file = self.base_path / "coverage.out"
        if not coverage_file.exists():
            return None

        try:
            result = subprocess.run(
                ["go", "tool", "cover", "-func=coverage.out"],
                cwd=str(self.base_path),
                capture_output=True,
                text=True,
            )

            # Parse "total:" Zeile
            for line in result.stdout.split("\n"):
                if "total:" in line:
                    parts = line.split()
                    if len(parts) >= 3:
                        percent_str = parts[-1].replace("%", "")
                        return {
                            "total_coverage": float(percent_str),
                            "raw_output": result.stdout,
                        }

        except Exception:
            pass

        return None