File size: 2,352 Bytes
bac155e 6c2d9a7 bac155e da65740 | 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 46 47 48 49 50 51 52 53 54 55 | from pathlib import Path
from app import _events_from_job_logs, _merge_events, _normalize_job_stage
def test_progress_endpoint_polls_hf_job_logs_and_extracts_worker_events():
app_py = Path("app.py").read_text(encoding="utf-8")
assert "inspect_job_safe(job_id" in app_py
assert "fetch_recent_logs_safe(job_id" in app_py
assert "_events_from_job_logs(job_logs)" in app_py
assert "_merge_events(bundle.get" in app_py
assert '"job_logs_polled": bool(include_job_logs)' in app_py
assert '"job_log_events_count": len(job_log_events)' in app_py
def test_job_log_event_parser_extracts_json_events_from_logs():
logs = """noise
{"ts":"2026-06-02T10:00:00+00:00","step":"pi_run","status":"running","message":"Pi is still working"}
prefix {"ts":"2026-06-02T10:00:01+00:00","step":"upload_files","status":"success","message":"Uploaded"}
not json
"""
events = _events_from_job_logs(logs)
assert [event["step"] for event in events] == ["pi_run", "upload_files"]
def test_merge_events_deduplicates_bucket_and_job_log_events():
event = {"ts": "t", "step": "pi_run", "status": "running", "message": "same"}
merged = _merge_events([event], [dict(event), {"ts": "t2", "step": "done", "status": "success"}])
assert len(merged) == 2
def test_job_stage_normalization():
assert _normalize_job_stage("JobStage.RUNNING") == "running"
assert _normalize_job_stage("COMPLETED") == "success"
assert _normalize_job_stage("CANCELLED") == "cancelled"
def test_active_run_only_keeps_cancel_button():
html = Path("web/index.html").read_text(encoding="utf-8")
active = html.split('id="activeRunPanel"', 1)[1].split('id="runsExplorerPanel"', 1)[0]
assert 'id="cancelRun"' in active
for removed in ['id="openArtifacts"', 'id="openRunState"', 'id="openRunReport"', 'id="openRunLogs"', 'id="openRunTraces"', 'id="openJob"', 'id="openSpace"', 'id="openSettings"']:
assert removed not in active
def test_progress_renderer_no_longer_sets_active_run_link_buttons():
js = Path("web/static/progress.js").read_text(encoding="utf-8")
assert "renderActiveBucketLinks" not in js
assert "openRunState" not in js
assert "setLink('openJob'" not in js
def test_app_version_v59():
assert '"version": "v70-progress-resume-events-latency"' in Path("app.py").read_text(encoding="utf-8")
|