File size: 1,965 Bytes
908e1cb da65740 12c824f 908e1cb | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 | 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
|