tests/run_tests.py

#!/usr/bin/env python3
"""Test runner + coverage matrix logger for Marionette.

Auto-detects OS, prompts for robot model, runs unit and E2E tests,
logs results to test_results.json, exports static/test_coverage.json,
and prints the coverage matrix with a test catalog summary.

Usage:
    cd marionette
    python tests/run_tests.py              # unit + E2E
    python tests/run_tests.py --hardware   # also run hardware tests (local)
    python tests/run_tests.py --on-robot   # run hardware tests on robot via SSH
"""

import argparse
import json
import platform
import subprocess
import sys
import tempfile
from datetime import datetime, timezone
from pathlib import Path

RESULTS_FILE = Path(__file__).parent / "test_results.json"
TESTS_DIR = Path(__file__).parent
MARIONETTE_DIR = TESTS_DIR.parent
STATIC_DIR = MARIONETTE_DIR / "marionette" / "static"
WEB_COVERAGE_FILE = STATIC_DIR / "test_coverage.json"

ROBOT_MODELS = ["None", "Lite", "Wireless"]
BROWSERS = ["chromium", "firefox", "webkit"]

# Human-readable descriptions for each test class.
# Keep in sync with the actual test classes in test_api.py, test_ui.py,
# and test_hardware.py.
TEST_CLASS_DESCRIPTIONS: dict[str, tuple[str, str]] = {
    # (tier, description)
    "TestSlugify": ("unit", "Move label slug generation"),
    "TestStateEndpoint": ("unit", "GET /api/state shape and initial values"),
    "TestStartingUpMode": ("unit", "Starting-up mode rejects commands"),
    "TestRecordEndpoint": ("unit", "POST /api/record validation and state transitions"),
    "TestPlayEndpoint": ("unit", "POST /api/play validation"),
    "TestStopEndpoints": ("unit", "POST /api/play/stop and /api/record/stop"),
    "TestMoveDelete": ("unit", "DELETE /api/moves/:id and file cleanup"),
    "TestDatasets": ("unit", "Dataset create, select, root change, origin"),
    "TestRegistryPersistence": ("unit", "Registry file survival across restarts"),
    "TestMovesRefresh": ("unit", "Move listing and metadata"),
    "TestExperiments": ("unit", "Feature toggles and experimental settings"),
    "TestHfAutoLogin": ("unit", "HF auto-login detection and caching"),
    "TestSensorData": ("unit", "Sensor data dummy endpoint"),
    "TestUploadAudio": ("unit", "POST /api/upload-audio validation and integration"),
    "TestMotionModelEndpoint": ("unit", "POST /api/motion-model enable, set, persist"),
    "TestCommunityDatasets": ("unit", "Community dataset listing and download validation"),
    "TestCorruptData": ("unit", "Malformed JSON and corrupt registry recovery"),
    "TestConcurrentStateChanges": ("unit", "Concurrent operations rejected when busy"),
    "TestDurationEdgeCases": ("unit", "Duration boundary validation (gt=0.5, le=300)"),
    "TestSyncDatasetExtended": ("unit", "Sync endpoint edge cases and validation"),
    "TestMicAgcConfig": ("unit", "Mic AGC disable/restore with mock USB device"),
    "TestApiContracts": ("unit", "API response shapes (refactoring protection)"),
    # E2E
    "TestPageLoad": ("e2e", "Page loads and main sections visible"),
    "TestIdleState": ("e2e", "Idle state display and controls"),
    "TestSimplifiedForm": ("e2e", "Simplified form layout"),
    "TestFormValidation": ("e2e", "Form fields and HTML5 validation"),
    "TestRecordingSubmission": ("e2e", "Recording submission changes mode"),
    "TestDatasetUI": ("e2e", "Dataset UI elements visible"),
    "TestRecordingLifecycle": ("e2e", "Recording submit, stop, and injected move visibility"),
    "TestPlaybackLifecycle": ("e2e", "Play button, queued playback, and move metadata"),
    "TestDeleteMove": ("e2e", "Delete button, confirm/cancel dialog handling"),
    "TestCreateDataset": ("e2e", "New dataset form show/hide/create/validate"),
    "TestSwitchDataset": ("e2e", "Dataset switching and dropdown population"),
    "TestAudioSourceRadios": ("e2e", "Audio source radio buttons and upload area toggle"),
    "TestSettingsPanel": ("e2e", "Settings expand, experimental toggle, root display"),
    "TestFormEdgeCases": ("e2e", "Empty/long/special labels and localStorage persistence"),
    "TestCommunitySection": ("e2e", "Community datasets section expand and fetch button"),
    "TestHfUploadSection": ("e2e", "HF username field and sync button presence"),
    # Hardware
    "TestHardwareStartup": ("hardware", "Robot reaches idle after startup"),
    "TestHardwareRecording": ("hardware", "Record and verify motion capture"),
    "TestHardwarePlayback": ("hardware", "Playback and delete (silent)"),
    "TestFullPipeline": ("hardware", "Full record → verify files → replay → delete lifecycle"),
    "TestMotionAccuracy": ("hardware", "Synthetic playback accuracy — reference vs observed poses"),
    "TestMultiDuration": ("hardware", "Recording and playback across 1s/3s/5s/10s durations"),
    "TestPerformance": ("hardware", "Startup, recording, and playback latency benchmarks"),
    "TestHardwareAudio": ("hardware", "Audio recording and playback (may skip on mic issues)"),
    "TestExistingRecordingPlayback": ("hardware", "Play back existing audio and silent recordings"),
    "TestRecordingRoundTrip": ("hardware", "Record → playback fidelity and timing verification"),
    "TestAntennaAndBodyYaw": ("hardware", "Verify antenna and body_yaw data in recordings"),
    "TestPlaybackAntennas": ("hardware", "Synthetic antenna oscillation playback"),
    "TestStopDuringPlayback": ("hardware", "Stop mid-playback and during goto-start-pose"),
    "TestStopDuringRecording": ("hardware", "Stop mid-recording, verify partial save"),
    "TestMultipleRecordPlayCycles": ("hardware", "5 back-to-back record/play cycles"),
    "TestPlaybackWithCorruptFile": ("hardware", "Deleted/corrupt JSON during playback"),
}


def detect_os() -> str:
    system = platform.system()
    if system == "Linux":
        return "Linux"
    elif system == "Darwin":
        return "macOS"
    elif system == "Windows":
        return "Windows"
    return system


def prompt_robot_model() -> str:
    print("\nWhich robot model is connected?")
    for i, model in enumerate(ROBOT_MODELS):
        print(f"  {i}) {model}")
    while True:
        choice = input(f"Enter 0-{len(ROBOT_MODELS) - 1} [0]: ").strip()
        if choice == "":
            return ROBOT_MODELS[0]
        try:
            idx = int(choice)
            if 0 <= idx < len(ROBOT_MODELS):
                return ROBOT_MODELS[idx]
        except ValueError:
            pass
        print("Invalid choice, try again.")


def detect_browsers() -> list[str]:
    """Return list of browsers that Playwright has installed."""
    available = []
    for browser in BROWSERS:
        result = subprocess.run(
            [
                sys.executable, "-c",
                f"from playwright.sync_api import sync_playwright; "
                f"p = sync_playwright().start(); "
                f"b = p.{browser}.launch(); b.close(); p.stop()",
            ],
            capture_output=True,
            timeout=30,
        )
        if result.returncode == 0:
            available.append(browser)
    return available


def run_pytest(args: list[str], label: str) -> dict:
    """Run pytest with verbose output and JSON report, return result summary."""
    with tempfile.NamedTemporaryFile(suffix=".json", delete=False) as f:
        json_path = f.name

    cmd = [
        sys.executable, "-m", "pytest",
        "--json-report", f"--json-report-file={json_path}",
        "-v",
        *args,
    ]

    print(f"\n{'─' * 60}")
    print(f"Running: {label}")
    print(f"Command: {' '.join(cmd)}")
    print(f"{'─' * 60}")

    result = subprocess.run(cmd, cwd=str(MARIONETTE_DIR))

    # Parse JSON report
    report_path = Path(json_path)
    report = None
    try:
        report = json.loads(report_path.read_text())
        summary = report.get("summary", {})
        passed = summary.get("passed", 0)
        failed = summary.get("failed", 0)
        errors = summary.get("error", 0)
        total = summary.get("total", 0)
    except (json.JSONDecodeError, FileNotFoundError):
        passed = 0
        failed = 0
        errors = 1
        total = 0
    finally:
        report_path.unlink(missing_ok=True)

    status = "pass" if (failed == 0 and errors == 0 and total > 0) else "fail"
    return {
        "passed": passed,
        "failed": failed,
        "errors": errors,
        "total": total,
        "status": status,
        "returncode": result.returncode,
        "report": report,
    }


def extract_class_counts(report: dict | None) -> dict[str, int]:
    """Extract per-class test counts from a pytest-json-report report."""
    counts: dict[str, int] = {}
    if not report:
        return counts
    for test in report.get("tests", []):
        node_id = test.get("nodeid", "")
        # nodeid looks like "tests/test_api.py::TestSlugify::test_simple_lowercase"
        parts = node_id.split("::")
        if len(parts) >= 2:
            class_name = parts[1]
            counts[class_name] = counts.get(class_name, 0) + 1
    return counts


def check_json_report_plugin() -> bool:
    """Check if pytest-json-report is available."""
    result = subprocess.run(
        [sys.executable, "-m", "pytest", "--co", "--json-report", "-q"],
        capture_output=True,
        cwd=str(MARIONETTE_DIR),
    )
    return result.returncode == 0


def load_results() -> list[dict]:
    if RESULTS_FILE.exists():
        return json.loads(RESULTS_FILE.read_text())
    return []


def save_results(results: list[dict]) -> None:
    RESULTS_FILE.write_text(json.dumps(results, indent=2) + "\n")


def result_fields(result: dict) -> dict:
    """Extract storable fields from a run result (excludes the full report)."""
    return {
        "passed": result["passed"],
        "failed": result["failed"],
        "errors": result["errors"],
        "total": result["total"],
        "status": result["status"],
        "returncode": result["returncode"],
    }


def build_test_catalog(all_class_counts: dict[str, int]) -> list[dict]:
    """Build a test catalog from observed class counts + known descriptions."""
    catalog = []
    for class_name, count in sorted(all_class_counts.items()):
        tier, description = TEST_CLASS_DESCRIPTIONS.get(
            class_name, ("unknown", class_name)
        )
        catalog.append({
            "class": class_name,
            "count": count,
            "tier": tier,
            "description": description,
        })
    return catalog


def print_test_catalog(catalog: list[dict]) -> None:
    """Print a grouped test catalog summary."""
    if not catalog:
        return

    print("Test Catalog\n")

    # Group by tier
    tiers: dict[str, list[dict]] = {}
    for entry in catalog:
        tiers.setdefault(entry["tier"], []).append(entry)

    tier_order = ["unit", "e2e", "hardware", "unknown"]
    tier_labels = {
        "unit": "Unit Tests",
        "e2e": "E2E Browser Tests",
        "hardware": "Hardware Tests",
        "unknown": "Other",
    }

    for tier in tier_order:
        entries = tiers.get(tier)
        if not entries:
            continue
        total = sum(e["count"] for e in entries)
        label = tier_labels.get(tier, tier)
        print(f"  {label} — {total} tests")
        for e in entries:
            print(f"    {e['class']} ({e['count']}){' ' * max(1, 28 - len(e['class']) - len(str(e['count'])))} {e['description']}")
        print()


def export_web_coverage(
    results: list[dict],
    catalog: list[dict],
) -> None:
    """Write static/test_coverage.json for the web UI."""
    # Build matrix entries (latest result per combination)
    latest: dict[tuple, dict] = {}
    for r in results:
        key = (r["os"], r["robot"], r["suite"], r.get("browser"))
        existing = latest.get(key)
        if existing is None or r["timestamp"] >= existing["timestamp"]:
            latest[key] = r

    matrix = []
    for key, r in sorted(latest.items()):
        dt = datetime.fromisoformat(r["timestamp"])
        matrix.append({
            "os": r["os"],
            "robot": r["robot"],
            "suite": r["suite"],
            "browser": r.get("browser"),
            "status": r["status"],
            "passed": r["passed"],
            "failed": r.get("failed", 0),
            "total": r["total"],
            "date": dt.strftime("%b %d"),
        })

    payload = {
        "generated_at": datetime.now(timezone.utc).isoformat(),
        "matrix": matrix,
        "test_catalog": catalog,
    }

    STATIC_DIR.mkdir(parents=True, exist_ok=True)
    WEB_COVERAGE_FILE.write_text(json.dumps(payload, indent=2) + "\n")
    print(f"Web coverage exported to {WEB_COVERAGE_FILE}")


def _print_matrix(results: list[dict]) -> None:
    """Print coverage matrix from results."""
    os_list = ["Linux", "macOS", "Windows"]
    robot_list = ["None", "Lite", "Wireless"]
    suites = [
        ("unit", None, "Unit Tests"),
        ("e2e", "chromium", "Chromium"),
        ("e2e", "firefox", "Firefox"),
        ("e2e", "webkit", "WebKit"),
        ("hardware", None, "Hardware"),
    ]

    # Build lookup: (os, robot, suite, browser) -> latest result
    latest: dict[tuple, dict] = {}
    for r in results:
        key = (r["os"], r["robot"], r["suite"], r.get("browser"))
        existing = latest.get(key)
        if existing is None or r["timestamp"] >= existing["timestamp"]:
            latest[key] = r

    # Only show columns that have at least one result
    active_suites = [
        s for s in suites
        if any(
            latest.get((os_name, robot, s[0], s[1]))
            for os_name in os_list
            for robot in robot_list
        )
    ]
    if not active_suites:
        active_suites = suites[:4]  # Show defaults even if empty

    col_w = 11
    label_w = 13

    print("Coverage Matrix (latest result per combination)\n")
    header = f"{'':>{label_w}} \u2502 " + " \u2502 ".join(
        f"{s[2]:^{col_w}}" for s in active_suites
    )
    print(header)
    sep = "\u2500" * label_w + "\u2500\u253c\u2500" + "\u2500\u253c\u2500".join(
        "\u2500" * (col_w + 1) for _ in active_suites
    )
    print(sep)

    for os_name in os_list:
        for robot in robot_list:
            combo = f"{os_name}/{robot[:4] + '.' if len(robot) > 4 else robot}"
            cells = []
            for suite, browser, _ in active_suites:
                key = (os_name, robot, suite, browser)
                r = latest.get(key)
                if r is None:
                    cells.append(f"{'—':^{col_w}}")
                else:
                    dt = datetime.fromisoformat(r["timestamp"])
                    date_str = dt.strftime("%b %d")
                    icon = "\u2705" if r["status"] == "pass" else "\u274c"
                    cells.append(f"{icon + ' ' + date_str:^{col_w}}")
            row = f"{combo:>{label_w}} \u2502 " + " \u2502 ".join(cells)
            print(row)

    print()


def main() -> None:
    parser = argparse.ArgumentParser(description="Run Marionette tests and log results.")
    parser.add_argument(
        "--hardware", action="store_true",
        help="Also run hardware integration tests (requires connected robot)",
    )
    parser.add_argument(
        "--on-robot", action="store_true",
        help="Run hardware tests on the robot via SSH (delegates to run_on_robot.py)",
    )
    parser.add_argument(
        "--host", default="reachy-mini.local",
        help="Robot hostname for --on-robot (default: reachy-mini.local)",
    )
    parser.add_argument(
        "--user", default="pollen",
        help="SSH user for --on-robot (default: pollen)",
    )
    args = parser.parse_args()

    # Delegate to run_on_robot.py if --on-robot
    if args.on_robot:
        run_on_robot = TESTS_DIR / "run_on_robot.py"
        cmd = [sys.executable, str(run_on_robot), "--host", args.host, "--user", args.user]
        sys.exit(subprocess.run(cmd).returncode)

    os_name = detect_os()
    print(f"Detected OS: {os_name}")

    robot = prompt_robot_model()
    print(f"Robot model: {robot}")

    # Check for pytest-json-report
    if not check_json_report_plugin():
        print("\npytest-json-report is required but not installed.")
        print("Install it with: pip install pytest-json-report")
        sys.exit(1)

    now = datetime.now(timezone.utc).isoformat()
    combo = f"{os_name}/{robot}"
    all_results = load_results()
    runs: list[dict] = []
    all_class_counts: dict[str, int] = {}

    # 1. Unit tests
    print(f"\n{'═' * 60}")
    print("UNIT TESTS")
    print(f"{'═' * 60}")
    unit = run_pytest(["tests/test_api.py"], "Unit tests")
    entry = {
        "timestamp": now,
        "os": os_name,
        "robot": robot,
        "suite": "unit",
        "browser": None,
        **result_fields(unit),
    }
    all_results.append(entry)
    runs.append(entry)
    all_class_counts.update(extract_class_counts(unit.get("report")))

    # 2. E2E tests per browser
    print(f"\n{'═' * 60}")
    print("E2E BROWSER TESTS")
    print(f"{'═' * 60}")

    available = detect_browsers()
    if not available:
        print("\nNo Playwright browsers found. Skipping E2E tests.")
        print("Install browsers with: playwright install --with-deps chromium firefox")
    else:
        print(f"Available browsers: {', '.join(available)}")

    for browser in available:
        e2e = run_pytest(
            ["tests/e2e", "--browser", browser],
            f"E2E — {browser}",
        )
        entry = {
            "timestamp": now,
            "os": os_name,
            "robot": robot,
            "suite": "e2e",
            "browser": browser,
            **result_fields(e2e),
        }
        all_results.append(entry)
        runs.append(entry)
        # E2E class counts only need to be captured once (same tests across browsers)
        if not any(k.startswith("TestPage") for k in all_class_counts):
            all_class_counts.update(extract_class_counts(e2e.get("report")))

    # 3. Hardware tests (optional)
    if args.hardware:
        hw_test_file = TESTS_DIR / "test_hardware.py"
        if hw_test_file.exists():
            print(f"\n{'═' * 60}")
            print("HARDWARE TESTS")
            print(f"{'═' * 60}")
            hw = run_pytest(
                ["tests/test_hardware.py", "-m", "hardware"],
                "Hardware integration tests",
            )
            entry = {
                "timestamp": now,
                "os": os_name,
                "robot": robot,
                "suite": "hardware",
                "browser": None,
                **result_fields(hw),
            }
            all_results.append(entry)
            runs.append(entry)
            all_class_counts.update(extract_class_counts(hw.get("report")))
        else:
            print("\nNo hardware test file found. Skipping.")

    # Save results
    save_results(all_results)
    print(f"\nResults saved to {RESULTS_FILE}")

    # Print run summary
    print(f"\n{'═' * 60}")
    print(f"SUMMARY — {combo}")
    print(f"{'═' * 60}")
    for r in runs:
        suite_label = r["suite"]
        if r.get("browser"):
            suite_label += f" ({r['browser']})"
        icon = "\u2705" if r["status"] == "pass" else "\u274c"
        print(f"  {icon} {suite_label}: {r['passed']}/{r['total']} passed")

    # Print test catalog
    catalog = build_test_catalog(all_class_counts)
    print()
    print_test_catalog(catalog)

    # Print matrix
    _print_matrix(all_results)

    # Export web coverage
    export_web_coverage(all_results, catalog)


if __name__ == "__main__":
    main()