348 lines
9.9 KiB
Python
348 lines
9.9 KiB
Python
"""
|
|
FastAPI router for tester tool.
|
|
"""
|
|
|
|
from pathlib import Path
|
|
from typing import Optional
|
|
from pydantic import BaseModel
|
|
from fastapi import APIRouter, HTTPException, Request
|
|
from fastapi.responses import HTMLResponse, PlainTextResponse, FileResponse
|
|
from fastapi.templating import Jinja2Templates
|
|
|
|
from .config import config, environments
|
|
from .core import (
|
|
discover_tests,
|
|
get_tests_tree,
|
|
start_test_run,
|
|
get_run_status,
|
|
list_runs,
|
|
TestStatus,
|
|
)
|
|
from .gherkin.parser import discover_features, extract_tags_from_features, get_feature_names, get_scenario_names
|
|
from .gherkin.sync import sync_features_from_album
|
|
|
|
|
|
router = APIRouter(prefix="/tools/tester", tags=["tester"])
|
|
templates = Jinja2Templates(directory=Path(__file__).parent / "templates")
|
|
|
|
|
|
class RunRequest(BaseModel):
|
|
"""Request to start a test run."""
|
|
test_ids: Optional[list[str]] = None
|
|
|
|
|
|
class RunResponse(BaseModel):
|
|
"""Response after starting a test run."""
|
|
run_id: str
|
|
status: str
|
|
|
|
|
|
class TestResultResponse(BaseModel):
|
|
"""A single test result."""
|
|
test_id: str
|
|
name: str
|
|
status: str
|
|
duration: float
|
|
error_message: Optional[str] = None
|
|
traceback: Optional[str] = None
|
|
artifacts: list[dict] = []
|
|
|
|
|
|
class RunStatusResponse(BaseModel):
|
|
"""Status of a test run."""
|
|
run_id: str
|
|
status: str
|
|
total: int
|
|
completed: int
|
|
passed: int
|
|
failed: int
|
|
errors: int
|
|
skipped: int
|
|
current_test: Optional[str] = None
|
|
results: list[TestResultResponse]
|
|
duration: Optional[float] = None
|
|
|
|
|
|
@router.get("/", response_class=HTMLResponse)
|
|
def index(request: Request):
|
|
"""Render the test runner UI."""
|
|
tests_tree = get_tests_tree()
|
|
tests_list = discover_tests()
|
|
|
|
return templates.TemplateResponse("index.html", {
|
|
"request": request,
|
|
"config": config,
|
|
"tests_tree": tests_tree,
|
|
"total_tests": len(tests_list),
|
|
})
|
|
|
|
|
|
@router.get("/health")
|
|
def health():
|
|
"""Health check endpoint."""
|
|
return {"status": "ok", "tool": "tester"}
|
|
|
|
|
|
@router.get("/filters", response_class=HTMLResponse)
|
|
def test_filters(request: Request):
|
|
"""Show filterable test view with multiple filter options."""
|
|
return templates.TemplateResponse("filters.html", {
|
|
"request": request,
|
|
"config": config,
|
|
})
|
|
|
|
|
|
@router.get("/filters_v2", response_class=HTMLResponse)
|
|
def test_filters_v2(request: Request):
|
|
"""Show Gherkin-driven filter view (v2 with pulse variables)."""
|
|
return templates.TemplateResponse("filters_v2.html", {
|
|
"request": request,
|
|
"config": config,
|
|
})
|
|
|
|
|
|
@router.get("/api/config")
|
|
def get_config():
|
|
"""Get current configuration."""
|
|
api_key = config.get("CONTRACT_TEST_API_KEY", "")
|
|
return {
|
|
"url": config.get("CONTRACT_TEST_URL", ""),
|
|
"has_api_key": bool(api_key),
|
|
"api_key_preview": f"{api_key[:8]}..." if len(api_key) > 8 else "",
|
|
}
|
|
|
|
|
|
@router.get("/api/environments")
|
|
def get_environments():
|
|
"""Get available test environments."""
|
|
# Sanitize API keys - only return preview
|
|
safe_envs = []
|
|
for env in environments:
|
|
safe_env = env.copy()
|
|
api_key = safe_env.get("api_key", "")
|
|
if api_key:
|
|
safe_env["has_api_key"] = True
|
|
safe_env["api_key_preview"] = f"{api_key[:8]}..." if len(api_key) > 8 else "***"
|
|
del safe_env["api_key"] # Don't send full key to frontend
|
|
else:
|
|
safe_env["has_api_key"] = False
|
|
safe_env["api_key_preview"] = ""
|
|
safe_envs.append(safe_env)
|
|
|
|
return {"environments": safe_envs}
|
|
|
|
|
|
@router.post("/api/environment/select")
|
|
def select_environment(env_id: str):
|
|
"""Select a target environment for testing."""
|
|
# Find the environment
|
|
env = next((e for e in environments if e["id"] == env_id), None)
|
|
if not env:
|
|
raise HTTPException(status_code=404, detail=f"Environment {env_id} not found")
|
|
|
|
# Update config (in memory for this session)
|
|
config["CONTRACT_TEST_URL"] = env["url"]
|
|
config["CONTRACT_TEST_API_KEY"] = env.get("api_key", "")
|
|
|
|
return {
|
|
"success": True,
|
|
"environment": {
|
|
"id": env["id"],
|
|
"name": env["name"],
|
|
"url": env["url"],
|
|
"has_api_key": bool(env.get("api_key"))
|
|
}
|
|
}
|
|
|
|
|
|
@router.get("/api/tests")
|
|
def list_tests():
|
|
"""List all discovered tests."""
|
|
tests = discover_tests()
|
|
return {
|
|
"total": len(tests),
|
|
"tests": [
|
|
{
|
|
"id": t.id,
|
|
"name": t.name,
|
|
"module": t.module,
|
|
"class_name": t.class_name,
|
|
"method_name": t.method_name,
|
|
"doc": t.doc,
|
|
}
|
|
for t in tests
|
|
],
|
|
}
|
|
|
|
|
|
@router.get("/api/tests/tree")
|
|
def get_tree():
|
|
"""Get tests as a tree structure."""
|
|
return get_tests_tree()
|
|
|
|
|
|
@router.post("/api/run", response_model=RunResponse)
|
|
def run_tests(request: RunRequest):
|
|
"""Start a test run."""
|
|
run_id = start_test_run(request.test_ids)
|
|
return RunResponse(run_id=run_id, status="running")
|
|
|
|
|
|
@router.get("/api/run/{run_id}", response_model=RunStatusResponse)
|
|
def get_run(run_id: str):
|
|
"""Get status of a test run (for polling)."""
|
|
status = get_run_status(run_id)
|
|
if not status:
|
|
raise HTTPException(status_code=404, detail=f"Run {run_id} not found")
|
|
|
|
duration = None
|
|
if status.started_at:
|
|
end_time = status.finished_at or __import__("time").time()
|
|
duration = round(end_time - status.started_at, 2)
|
|
|
|
return RunStatusResponse(
|
|
run_id=status.run_id,
|
|
status=status.status,
|
|
total=status.total,
|
|
completed=status.completed,
|
|
passed=status.passed,
|
|
failed=status.failed,
|
|
errors=status.errors,
|
|
skipped=status.skipped,
|
|
current_test=status.current_test,
|
|
duration=duration,
|
|
results=[
|
|
TestResultResponse(
|
|
test_id=r.test_id,
|
|
name=r.name,
|
|
status=r.status.value,
|
|
duration=round(r.duration, 3),
|
|
error_message=r.error_message,
|
|
traceback=r.traceback,
|
|
artifacts=r.artifacts,
|
|
)
|
|
for r in status.results
|
|
],
|
|
)
|
|
|
|
|
|
@router.get("/api/runs")
|
|
def list_all_runs():
|
|
"""List all test runs."""
|
|
return {"runs": list_runs()}
|
|
|
|
|
|
@router.get("/api/artifact/{run_id}/{filename}")
|
|
def stream_artifact(run_id: str, filename: str):
|
|
"""
|
|
Stream an artifact file (video, screenshot, trace).
|
|
|
|
Similar to jira vein's attachment streaming endpoint.
|
|
"""
|
|
# Get artifacts directory
|
|
artifacts_dir = Path(__file__).parent / "artifacts"
|
|
|
|
# Search for the artifact in all subdirectories
|
|
for subdir in ["videos", "screenshots", "traces"]:
|
|
artifact_path = artifacts_dir / subdir / run_id / filename
|
|
if artifact_path.exists():
|
|
# Determine media type
|
|
if filename.endswith(".webm"):
|
|
media_type = "video/webm"
|
|
elif filename.endswith(".mp4"):
|
|
media_type = "video/mp4"
|
|
elif filename.endswith(".png"):
|
|
media_type = "image/png"
|
|
elif filename.endswith(".jpg") or filename.endswith(".jpeg"):
|
|
media_type = "image/jpeg"
|
|
elif filename.endswith(".zip"):
|
|
media_type = "application/zip"
|
|
else:
|
|
media_type = "application/octet-stream"
|
|
|
|
return FileResponse(
|
|
path=artifact_path,
|
|
media_type=media_type,
|
|
filename=filename
|
|
)
|
|
|
|
# Not found
|
|
raise HTTPException(status_code=404, detail=f"Artifact not found: {run_id}/{filename}")
|
|
|
|
|
|
@router.get("/api/artifacts/{run_id}")
|
|
def list_artifacts(run_id: str):
|
|
"""List all artifacts for a test run."""
|
|
artifacts_dir = Path(__file__).parent / "artifacts"
|
|
artifacts = []
|
|
|
|
# Search in all artifact directories
|
|
for subdir, artifact_type in [
|
|
("videos", "video"),
|
|
("screenshots", "screenshot"),
|
|
("traces", "trace")
|
|
]:
|
|
run_dir = artifacts_dir / subdir / run_id
|
|
if run_dir.exists():
|
|
for artifact_file in run_dir.iterdir():
|
|
if artifact_file.is_file():
|
|
artifacts.append({
|
|
"type": artifact_type,
|
|
"filename": artifact_file.name,
|
|
"size": artifact_file.stat().st_size,
|
|
"url": f"/tools/tester/api/artifact/{run_id}/{artifact_file.name}"
|
|
})
|
|
|
|
return {"artifacts": artifacts}
|
|
|
|
|
|
@router.get("/api/features")
|
|
def list_features():
|
|
"""List all discovered Gherkin features."""
|
|
features_dir = Path(__file__).parent / "features"
|
|
features = discover_features(features_dir)
|
|
|
|
return {
|
|
"features": [
|
|
{
|
|
"name": f.name,
|
|
"description": f.description,
|
|
"file_path": f.file_path,
|
|
"language": f.language,
|
|
"tags": f.tags,
|
|
"scenario_count": len(f.scenarios),
|
|
"scenarios": [
|
|
{
|
|
"name": s.name,
|
|
"description": s.description,
|
|
"tags": s.tags,
|
|
"type": s.scenario_type,
|
|
}
|
|
for s in f.scenarios
|
|
]
|
|
}
|
|
for f in features
|
|
],
|
|
"total": len(features)
|
|
}
|
|
|
|
|
|
@router.get("/api/features/tags")
|
|
def list_feature_tags():
|
|
"""List all unique tags from Gherkin features."""
|
|
features_dir = Path(__file__).parent / "features"
|
|
features = discover_features(features_dir)
|
|
tags = extract_tags_from_features(features)
|
|
|
|
return {
|
|
"tags": sorted(list(tags)),
|
|
"total": len(tags)
|
|
}
|
|
|
|
|
|
@router.post("/api/features/sync")
|
|
def sync_features():
|
|
"""Sync feature files from album/book/gherkin-samples/."""
|
|
result = sync_features_from_album()
|
|
return result
|