diff --git a/scripts/ai_run_tests.py b/scripts/ai_run_tests.py new file mode 100755 index 0000000..1c06060 --- /dev/null +++ b/scripts/ai_run_tests.py @@ -0,0 +1,519 @@ +#!/usr/bin/env python3 +"""AI Test Runner — Unified test execution and reporting for LeoCRM. + +Runs backend (pytest), frontend (vitest), and E2E (playwright) tests, +produces a structured JSON or CSV report, and exits with code 0 (all pass) +or 1 (any failures). + +Usage: + python scripts/ai_run_tests.py [options] + +Options: + --skip-backend Skip backend tests + --skip-frontend Skip frontend tests + --skip-e2e Skip E2E tests (default: skipped unless --run-e2e) + --run-e2e Run E2E tests (opt-in) + --output FORMAT Output format: json (default) or csv + --verbose Show full test output + --report-file PATH Write report to file instead of stdout +""" + +from __future__ import annotations + +import argparse +import csv +import io +import json +import os +import subprocess +import sys +import time +from pathlib import Path +from typing import Any + +# ─── Constants ─── + +PROJECT_ROOT = Path(__file__).resolve().parent.parent +FRONTEND_DIR = PROJECT_ROOT / "frontend" +BACKEND_DIR = PROJECT_ROOT + +# ─── Data Structures ─── + + +class TestSuiteResult: + """Result of a single test suite (backend, frontend, e2e).""" + + def __init__(self, suite: str): + self.suite = suite + self.total = 0 + self.passed = 0 + self.failed = 0 + self.skipped = 0 + self.duration = 0.0 + self.failures: list[dict[str, str]] = [] + self.errors: list[dict[str, str]] = [] + self.raw_output = "" + self.exit_code = 0 + self.command = "" + + def to_dict(self) -> dict[str, Any]: + return { + "suite": self.suite, + "total": self.total, + "passed": self.passed, + "failed": self.failed, + "skipped": self.skipped, + "duration": round(self.duration, 2), + "exit_code": self.exit_code, + "command": self.command, + "failures": self.failures, + "errors": self.errors, + } + + @property + def all_passed(self) -> bool: + return self.failed == 0 and self.exit_code == 0 + + +# ─── Test Runners ─── + + +def run_backend_tests(verbose: bool = False) -> TestSuiteResult: + """Run pytest with --tb=short and JSON report.""" + result = TestSuiteResult("backend") + report_file = BACKEND_DIR / ".test_report_backend.json" + + cmd = [ + sys.executable, + "-m", + "pytest", + "tests/", + "--tb=short", + f"--json-report-file={report_file}", + "-q", + ] + if verbose: + cmd.append("-v") + + result.command = " ".join(cmd) + start = time.time() + + try: + proc = subprocess.run( + cmd, + cwd=str(BACKEND_DIR), + capture_output=True, + text=True, + timeout=600, + ) + result.exit_code = proc.returncode + result.raw_output = proc.stdout + proc.stderr + except subprocess.TimeoutExpired: + result.exit_code = 124 + result.raw_output = "Backend tests timed out after 600s" + result.errors.append({"test": "__timeout__", "error": "Test suite timed out"}) + except FileNotFoundError: + result.exit_code = 127 + result.raw_output = "pytest not found" + result.errors.append({"test": "__setup__", "error": "pytest not installed"}) + + result.duration = time.time() - start + + # Parse JSON report if available + if report_file.exists(): + try: + with open(report_file) as f: + data = json.load(f) + result.total = data.get("summary", {}).get("total", 0) + result.passed = data.get("summary", {}).get("passed", 0) + result.failed = data.get("summary", {}).get("failed", 0) + result.skipped = data.get("summary", {}).get("skipped", 0) + + # Extract failure details + for test in data.get("tests", []): + if test.get("outcome") == "failed": + result.failures.append({ + "test": test.get("nodeid", "unknown"), + "error": test.get("call", {}).get("longrepr", "No error details"), + }) + except (json.JSONDecodeError, KeyError): + pass + finally: + report_file.unlink(missing_ok=True) + else: + # Fallback: parse from output + _parse_pytest_output(result) + + return result + + +def run_frontend_tests(verbose: bool = False) -> TestSuiteResult: + """Run vitest with JSON output.""" + result = TestSuiteResult("frontend") + report_file = FRONTEND_DIR / ".test_report_frontend.json" + + cmd = ["npx", "vitest", "run", "src/__tests__/", "--reporter=json"] + if verbose: + cmd.append("--reporter=verbose") + + result.command = " ".join(cmd) + start = time.time() + + try: + proc = subprocess.run( + cmd, + cwd=str(FRONTEND_DIR), + capture_output=True, + text=True, + timeout=300, + ) + result.exit_code = proc.returncode + result.raw_output = proc.stdout + proc.stderr + except subprocess.TimeoutExpired: + result.exit_code = 124 + result.raw_output = "Frontend tests timed out after 300s" + result.errors.append({"test": "__timeout__", "error": "Test suite timed out"}) + except FileNotFoundError: + result.exit_code = 127 + result.raw_output = "npx/vitest not found" + result.errors.append({"test": "__setup__", "error": "vitest not installed"}) + + result.duration = time.time() - start + + # Try to parse JSON from stdout (vitest --reporter=json outputs to stdout) + _parse_vitest_output(result) + + # Clean up report file if exists + report_file.unlink(missing_ok=True) + + return result + + +def run_e2e_tests(verbose: bool = False) -> TestSuiteResult: + """Run playwright E2E tests.""" + result = TestSuiteResult("e2e") + + cmd = ["npx", "playwright", "test", "--reporter=json"] + if verbose: + cmd.append("--reporter=list") + + result.command = " ".join(cmd) + start = time.time() + + try: + proc = subprocess.run( + cmd, + cwd=str(FRONTEND_DIR), + capture_output=True, + text=True, + timeout=600, + ) + result.exit_code = proc.returncode + result.raw_output = proc.stdout + proc.stderr + except subprocess.TimeoutExpired: + result.exit_code = 124 + result.raw_output = "E2E tests timed out after 600s" + result.errors.append({"test": "__timeout__", "error": "Test suite timed out"}) + except FileNotFoundError: + result.exit_code = 127 + result.raw_output = "playwright not found" + result.errors.append({"test": "__setup__", "error": "playwright not installed"}) + + result.duration = time.time() - start + + # Parse playwright JSON output + _parse_playwright_output(result) + + return result + + +# ─── Output Parsers ─── + + +def _parse_pytest_output(result: TestSuiteResult) -> None: + """Fallback parser for pytest text output.""" + output = result.raw_output + for line in output.splitlines(): + line = line.strip() + # Look for summary line like: "5 passed, 2 failed, 1 skipped in 10.5s" + if "passed" in line or "failed" in line or "skipped" in line: + parts = line.split() + for part in parts: + if part.isdigit(): + if "passed" in line and "passed" in parts[parts.index(part) + 1:]: + result.passed = int(part) + elif "failed" in line and "failed" in parts[parts.index(part) + 1:]: + result.failed = int(part) + elif "skipped" in line and "skipped" in parts[parts.index(part) + 1:]: + result.skipped = int(part) + result.total = result.passed + result.failed + result.skipped + + +def _parse_vitest_output(result: TestSuiteResult) -> None: + """Parse vitest JSON output from stdout.""" + output = result.raw_output + + # Try to find JSON block in output + json_start = output.find("{") + if json_start == -1: + # Fallback: parse from summary line + for line in output.splitlines(): + if "Tests" in line and ("passed" in line or "failed" in line): + # e.g. "Tests 76 passed (76)" or "Tests 2 failed | 74 passed (76)" + parts = line.split() + for i, part in enumerate(parts): + if part.isdigit() and i + 1 < len(parts): + next_word = parts[i + 1] + if next_word == "passed": + result.passed = int(part) + elif next_word == "failed": + result.failed = int(part) + elif next_word == "skipped": + result.skipped = int(part) + result.total = result.passed + result.failed + result.skipped + return + + # Try to extract valid JSON + try: + # Find the last valid JSON object in the output + json_str = output[json_start:] + # Find matching closing brace + brace_count = 0 + end_idx = 0 + for i, char in enumerate(json_str): + if char == "{": + brace_count += 1 + elif char == "}": + brace_count -= 1 + if brace_count == 0: + end_idx = i + 1 + break + + if end_idx > 0: + data = json.loads(json_str[:end_idx]) + result.total = data.get("numTotalTests", 0) + result.passed = data.get("numPassedTests", 0) + result.failed = data.get("numFailedTests", 0) + result.skipped = data.get("numPendingTests", 0) + + # Extract failure details + for test_file in data.get("testResults", []): + for assertion in test_file.get("assertionResults", []): + if assertion.get("status") == "failed": + result.failures.append({ + "test": assertion.get("fullName", "unknown"), + "error": assertion.get("failureMessages", ["No error details"])[0], + }) + except (json.JSONDecodeError, KeyError, IndexError): + # Fallback to text parsing + for line in output.splitlines(): + if "Tests" in line and ("passed" in line or "failed" in line): + parts = line.split() + for i, part in enumerate(parts): + if part.isdigit() and i + 1 < len(parts): + next_word = parts[i + 1] + if next_word == "passed": + result.passed = int(part) + elif next_word == "failed": + result.failed = int(part) + elif next_word == "skipped": + result.skipped = int(part) + result.total = result.passed + result.failed + result.skipped + + +def _parse_playwright_output(result: TestSuiteResult) -> None: + """Parse playwright JSON output.""" + output = result.raw_output + + json_start = output.find("{") + if json_start == -1: + return + + try: + json_str = output[json_start:] + brace_count = 0 + end_idx = 0 + for i, char in enumerate(json_str): + if char == "{": + brace_count += 1 + elif char == "}": + brace_count -= 1 + if brace_count == 0: + end_idx = i + 1 + break + + if end_idx > 0: + data = json.loads(json_str[:end_idx]) + stats = data.get("stats", {}) + result.total = stats.get("total", 0) + result.passed = stats.get("passed", 0) + result.failed = stats.get("failed", 0) + result.skipped = stats.get("skipped", 0) + + # Extract failure details from suites + for suite in data.get("suites", []): + for spec in suite.get("specs", []): + if spec.get("ok") is False: + for test in spec.get("tests", []): + if test.get("status") == "failed": + result.failures.append({ + "test": spec.get("title", "unknown"), + "error": test.get("error", {}).get("message", "No error details"), + }) + except (json.JSONDecodeError, KeyError, IndexError): + pass + + +# ─── Report Generation ─── + + +def generate_json_report(results: list[TestSuiteResult]) -> dict[str, Any]: + """Generate unified JSON report.""" + total_all = sum(r.total for r in results) + passed_all = sum(r.passed for r in results) + failed_all = sum(r.failed for r in results) + skipped_all = sum(r.skipped for r in results) + duration_all = sum(r.duration for r in results) + + all_failures: list[dict[str, str]] = [] + for r in results: + for f in r.failures: + all_failures.append({"suite": r.suite, **f}) + + return { + "summary": { + "total": total_all, + "passed": passed_all, + "failed": failed_all, + "skipped": skipped_all, + "duration": round(duration_all, 2), + "all_passed": failed_all == 0, + }, + "suites": [r.to_dict() for r in results], + "failures": all_failures, + } + + +def generate_csv_report(results: list[TestSuiteResult]) -> str: + """Generate CSV report.""" + output = io.StringIO() + writer = csv.writer(output) + writer.writerow(["suite", "total", "passed", "failed", "skipped", "duration", "exit_code"]) + for r in results: + writer.writerow([ + r.suite, r.total, r.passed, r.failed, r.skipped, + round(r.duration, 2), r.exit_code, + ]) + # Add failures section + writer.writerow([]) + writer.writerow(["FAILURES"]) + writer.writerow(["suite", "test", "error"]) + for r in results: + for f in r.failures: + writer.writerow([r.suite, f.get("test", ""), f.get("error", "")]) + return output.getvalue() + + +def print_console_summary(results: list[TestSuiteResult]) -> None: + """Print a human-readable summary to console.""" + print("\n" + "=" * 60) + print(" AI TEST RUNNER — SUMMARY") + print("=" * 60) + + for r in results: + status = "✓ PASS" if r.all_passed else "✗ FAIL" + print(f"\n [{status}] {r.suite.upper()}") + print(f" Tests: {r.total} total, {r.passed} passed, {r.failed} failed, {r.skipped} skipped") + print(f" Duration: {r.duration:.1f}s") + if r.failures: + print(f" Failures ({len(r.failures)}):") + for f in r.failures[:5]: + test_name = f.get("test", "unknown") + if len(test_name) > 70: + test_name = test_name[:67] + "..." + print(f" - {test_name}") + if len(r.failures) > 5: + print(f" ... and {len(r.failures) - 5} more") + + total_all = sum(r.total for r in results) + passed_all = sum(r.passed for r in results) + failed_all = sum(r.failed for r in results) + duration_all = sum(r.duration for r in results) + + print("\n" + "-" * 60) + overall = "ALL PASSED" if failed_all == 0 else f"{failed_all} FAILURES" + print(f" OVERALL: {overall}") + print(f" Total: {passed_all}/{total_all} tests passed in {duration_all:.1f}s") + print("=" * 60 + "\n") + + +# ─── Main ─── + + +def main() -> int: + parser = argparse.ArgumentParser( + description="AI Test Runner — Unified test execution and reporting for LeoCRM" + ) + parser.add_argument("--skip-backend", action="store_true", help="Skip backend tests") + parser.add_argument("--skip-frontend", action="store_true", help="Skip frontend tests") + parser.add_argument("--skip-e2e", action="store_true", default=True, help="Skip E2E tests (default)") + parser.add_argument("--run-e2e", action="store_true", help="Run E2E tests (opt-in)") + parser.add_argument("--output", choices=["json", "csv"], default="json", help="Output format") + parser.add_argument("--verbose", action="store_true", help="Show full test output") + parser.add_argument("--report-file", type=str, default=None, help="Write report to file") + args = parser.parse_args() + + results: list[TestSuiteResult] = [] + + # Run backend tests + if not args.skip_backend: + print("\n▶ Running backend tests (pytest)...") + backend_result = run_backend_tests(verbose=args.verbose) + results.append(backend_result) + if args.verbose and backend_result.raw_output: + print(backend_result.raw_output[:5000]) + + # Run frontend tests + if not args.skip_frontend: + print("\n▶ Running frontend tests (vitest)...") + frontend_result = run_frontend_tests(verbose=args.verbose) + results.append(frontend_result) + if args.verbose and frontend_result.raw_output: + print(frontend_result.raw_output[:5000]) + + # Run E2E tests (opt-in) + if args.run_e2e and not args.skip_e2e: + print("\n▶ Running E2E tests (playwright)...") + e2e_result = run_e2e_tests(verbose=args.verbose) + results.append(e2e_result) + if args.verbose and e2e_result.raw_output: + print(e2e_result.raw_output[:5000]) + + if not results: + print("\n⚠ No test suites were run. Use --skip-* flags to control which suites run.") + return 0 + + # Print console summary + print_console_summary(results) + + # Generate report + if args.output == "json": + report = generate_json_report(results) + report_str = json.dumps(report, indent=2, ensure_ascii=False) + else: + report_str = generate_csv_report(results) + + # Output report + if args.report_file: + Path(args.report_file).write_text(report_str, encoding="utf-8") + print(f" Report written to: {args.report_file}") + else: + print(report_str) + + # Exit code: 0 if all passed, 1 if any failures + any_failed = any(not r.all_passed for r in results) + return 1 if any_failed else 0 + + +if __name__ == "__main__": + sys.exit(main())