File size: 1,632 Bytes
12c824f 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 | from pathlib import Path
from app import _events_from_job_logs
def test_job_log_event_parser_handles_prefix_suffix_and_ansi():
logs = """
2026-06-02T10:00:00Z stdout: \x1b[32m{"ts":"2026-06-02T10:00:00+00:00","step":"bootstrap","status":"success","message":"ok"}\x1b[0m
INFO prefix {"ts":"2026-06-02T10:01:00+00:00","step":"pi_run","status":"running","message":"Pi"} trailing text
"""
events = _events_from_job_logs(logs)
assert [e["step"] for e in events] == ["bootstrap", "pi_run"]
def test_progress_endpoint_fetches_deeper_job_log_window():
app = Path("app.py").read_text(encoding="utf-8")
assert "max_lines=1000" in app
assert "JSONDecoder().raw_decode" not in app # implementation uses decoder variable
assert "decoder.raw_decode" in app
assert "ANSI_RE" in app
def test_frontend_progress_is_monotonic_per_run():
js = Path("web/static/progress.js").read_text(encoding="utf-8")
assert "const progressStability" in js
assert "function stabilizeProgressPayload" in js
assert "Math.max(rawProgress,progressStability.progress)" in js
assert "timelineFurthestIndex" in js
assert "resetProgressStability(runId)" in Path("web/static/app.js").read_text(encoding="utf-8")
def test_light_poll_cannot_reset_timeline_signature_to_earlier_steps():
js = Path("web/static/progress.js").read_text(encoding="utf-8")
assert "rawIndex>=progressStability.furthestIndex" in js
assert "cloneTimeline(progressStability.timeline)" in js
def test_app_version_v67():
assert '"version": "v70-progress-resume-events-latency"' in Path("app.py").read_text(encoding="utf-8")
|