| from pathlib import Path |
|
|
| from src.jobs import cancel_job_safe |
| from src.progress import STEP_LABELS, STEP_ORDER, progress_from_events |
|
|
|
|
| def test_progress_timeline_uses_exact_worker_steps(): |
| assert "pi_verification" in STEP_ORDER |
| assert "metadata_sanitize" in STEP_ORDER |
| assert "generation_smoke" in STEP_ORDER |
| events = [ |
| {"ts": "2026-06-02T10:00:00+00:00", "step": "bootstrap", "status": "started"}, |
| {"ts": "2026-06-02T10:00:01+00:00", "step": "metadata_sanitize", "status": "success"}, |
| {"ts": "2026-06-02T10:00:02+00:00", "step": "generation_smoke", "status": "started"}, |
| ] |
| result = progress_from_events(events) |
| assert result["current_step"] == "generation_smoke" |
| assert any(x["step"] == "metadata_sanitize" and x["label"] == STEP_LABELS["metadata_sanitize"] for x in result["timeline"]) |
| assert any(x["step"] == "generation_smoke" and x["status"] == "running" for x in result["timeline"]) |
|
|
|
|
| def test_frontend_progress_polling_is_no_cache_and_faster(): |
| api_js = Path("web/static/api.js").read_text(encoding="utf-8") |
| app_js = Path("web/static/app.js").read_text(encoding="utf-8") |
| progress_js = Path("web/static/progress.js").read_text(encoding="utf-8") |
| assert 'cache:"no-store"' in api_js |
| assert "withCacheBust" in api_js |
| assert "setInterval(refreshProgress,2000)" in app_js |
| assert "stabilizeProgressPayload" in progress_js |
|
|
|
|
| def test_cancel_endpoint_and_button_exist(): |
| app_py = Path("app.py").read_text(encoding="utf-8") |
| html = Path("web/index.html").read_text(encoding="utf-8") |
| js = Path("web/static/app.js").read_text(encoding="utf-8") |
| assert '"/api/runs/{run_id}/cancel"' in app_py |
| assert "cancel_job_safe" in app_py |
| assert 'id="cancelRun"' in html |
| assert "cancelActiveRun" in js |
| assert "/cancel?bucket_name=" in js |
|
|
|
|
| def test_cancel_job_safe_requires_job_id(): |
| assert cancel_job_safe("", namespace="alice", token="fake")["ok"] is False |
|
|
