| 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") |
|
|
