Samir Boulahtit
0e6c9e3eea
Improve the testing dashboard to run pytest in the background: - Add background task execution using FastAPI's BackgroundTasks - Create test_runner_tasks.py following existing background task pattern - API now returns immediately after starting the test run - Frontend polls for status every 2 seconds until completion - Show running indicator with elapsed time counter - Resume polling if user navigates away and returns while tests running - Tests continue running even if user closes the page 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
63 lines
1.9 KiB
Python
63 lines
1.9 KiB
Python
# app/tasks/test_runner_tasks.py
|
|
"""Background tasks for test runner."""
|
|
|
|
import logging
|
|
from datetime import UTC, datetime
|
|
|
|
from app.core.database import SessionLocal
|
|
from app.services.test_runner_service import test_runner_service
|
|
from models.database.test_run import TestRun
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
async def execute_test_run(
|
|
run_id: int,
|
|
test_path: str = "tests",
|
|
extra_args: list[str] | None = None,
|
|
):
|
|
"""Background task to execute pytest tests.
|
|
|
|
Args:
|
|
run_id: ID of the TestRun record
|
|
test_path: Path to tests (relative to project root)
|
|
extra_args: Additional pytest arguments
|
|
"""
|
|
db = SessionLocal()
|
|
test_run = None
|
|
|
|
try:
|
|
# Get the test run record
|
|
test_run = db.query(TestRun).filter(TestRun.id == run_id).first()
|
|
if not test_run:
|
|
logger.error(f"Test run {run_id} not found")
|
|
return
|
|
|
|
logger.info(f"Starting test execution: Run {run_id}, Path: {test_path}")
|
|
|
|
# Execute the tests
|
|
test_runner_service._execute_tests(db, test_run, test_path, extra_args)
|
|
db.commit()
|
|
|
|
logger.info(
|
|
f"Test run {run_id} completed: "
|
|
f"status={test_run.status}, passed={test_run.passed}, "
|
|
f"failed={test_run.failed}, duration={test_run.duration_seconds:.1f}s"
|
|
)
|
|
|
|
except Exception as e:
|
|
logger.error(f"Test run {run_id} failed: {e}", exc_info=True)
|
|
if test_run is not None:
|
|
try:
|
|
test_run.status = "error"
|
|
db.commit()
|
|
except Exception as commit_error:
|
|
logger.error(f"Failed to update test run status: {commit_error}")
|
|
db.rollback()
|
|
finally:
|
|
if hasattr(db, "close") and callable(db.close):
|
|
try:
|
|
db.close()
|
|
except Exception as close_error:
|
|
logger.error(f"Error closing database session: {close_error}")
|