soleprint/station/tools/tester/playwright/runner.py

"""
Execute Playwright tests and capture artifacts.
"""

import subprocess
import json
import time
from pathlib import Path
from typing import Optional
from dataclasses import dataclass, field


@dataclass
class PlaywrightResult:
    """Result of a Playwright test execution."""
    test_id: str
    name: str
    status: str  # "passed", "failed", "skipped"
    duration: float
    error_message: Optional[str] = None
    traceback: Optional[str] = None
    artifacts: list[dict] = field(default_factory=list)


class PlaywrightRunner:
    """Run Playwright tests and collect artifacts."""

    def __init__(self, tests_dir: Path, artifacts_dir: Path):
        self.tests_dir = tests_dir
        self.artifacts_dir = artifacts_dir
        self.videos_dir = artifacts_dir / "videos"
        self.screenshots_dir = artifacts_dir / "screenshots"
        self.traces_dir = artifacts_dir / "traces"

        # Ensure artifact directories exist
        self.videos_dir.mkdir(parents=True, exist_ok=True)
        self.screenshots_dir.mkdir(parents=True, exist_ok=True)
        self.traces_dir.mkdir(parents=True, exist_ok=True)

    def run_tests(
        self,
        test_files: Optional[list[str]] = None,
        run_id: Optional[str] = None
    ) -> list[PlaywrightResult]:
        """
        Run Playwright tests and collect results.

        Args:
            test_files: List of test file paths to run (relative to tests_dir).
                       If None, runs all tests.
            run_id: Optional run ID to namespace artifacts.

        Returns:
            List of PlaywrightResult objects.
        """
        if not self.tests_dir.exists():
            return []

        # Build playwright command
        cmd = ["npx", "playwright", "test"]

        # Add specific test files if provided
        if test_files:
            cmd.extend(test_files)

        # Add reporter for JSON output
        results_file = self.artifacts_dir / f"results_{run_id or 'latest'}.json"
        cmd.extend([
            "--reporter=json",
            f"--output={results_file}"
        ])

        # Configure artifact collection
        # Videos and screenshots are configured in playwright.config.ts
        # We'll assume config is set to capture on failure

        # Run tests
        start_time = time.time()

        try:
            result = subprocess.run(
                cmd,
                cwd=self.tests_dir,
                capture_output=True,
                text=True,
                timeout=600  # 10 minute timeout
            )

            # Parse results
            if results_file.exists():
                with open(results_file) as f:
                    results_data = json.load(f)
                    return self._parse_results(results_data, run_id)
            else:
                # No results file - likely error
                return self._create_error_result(result.stderr)

        except subprocess.TimeoutExpired:
            return self._create_error_result("Tests timed out after 10 minutes")
        except Exception as e:
            return self._create_error_result(str(e))

    def _parse_results(
        self,
        results_data: dict,
        run_id: Optional[str]
    ) -> list[PlaywrightResult]:
        """Parse Playwright JSON results."""
        parsed_results = []

        # Playwright JSON reporter structure:
        # {
        #   "suites": [...],
        #   "tests": [...],
        # }

        tests = results_data.get("tests", [])

        for test in tests:
            test_id = test.get("testId", "unknown")
            title = test.get("title", "Unknown test")
            status = test.get("status", "unknown")  # passed, failed, skipped
            duration = test.get("duration", 0) / 1000.0  # Convert ms to seconds

            error_message = None
            traceback = None

            # Extract error if failed
            if status == "failed":
                error = test.get("error", {})
                error_message = error.get("message", "Test failed")
                traceback = error.get("stack", "")

            # Collect artifacts
            artifacts = []
            for attachment in test.get("attachments", []):
                artifact_type = attachment.get("contentType", "")
                artifact_path = attachment.get("path", "")

                if artifact_path:
                    artifact_file = Path(artifact_path)
                    if artifact_file.exists():
                        # Determine type
                        if "video" in artifact_type:
                            type_label = "video"
                        elif "image" in artifact_type:
                            type_label = "screenshot"
                        elif "trace" in artifact_type:
                            type_label = "trace"
                        else:
                            type_label = "attachment"

                        artifacts.append({
                            "type": type_label,
                            "filename": artifact_file.name,
                            "path": str(artifact_file),
                            "size": artifact_file.stat().st_size,
                            "mimetype": artifact_type,
                        })

            parsed_results.append(PlaywrightResult(
                test_id=test_id,
                name=title,
                status=status,
                duration=duration,
                error_message=error_message,
                traceback=traceback,
                artifacts=artifacts,
            ))

        return parsed_results

    def _create_error_result(self, error_msg: str) -> list[PlaywrightResult]:
        """Create an error result when test execution fails."""
        return [
            PlaywrightResult(
                test_id="playwright_error",
                name="Playwright Execution Error",
                status="failed",
                duration=0.0,
                error_message=error_msg,
                traceback="",
                artifacts=[],
            )
        ]

    def get_artifact_url(self, run_id: str, artifact_filename: str) -> str:
        """Generate URL for streaming an artifact."""
        return f"/tools/tester/api/artifact/{run_id}/{artifact_filename}"