feat(7.9): add AI test runner script with unified JSON/CSV reporting
This commit is contained in:
Agent Zero
Executable
+519
@@ -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())
|
||||
Reference in New Issue
Block a user