File size: 1,630 Bytes
b67c03f 908e1cb b67c03f 908e1cb 5f8f821 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 | from src.progress import progress_from_events
def test_progress_from_events_tracks_running_pi():
events = [
{"ts": "2026-06-02T10:00:00+00:00", "step": "bootstrap", "status": "started", "message": "started"},
{"ts": "2026-06-02T10:00:02+00:00", "step": "dependencies", "status": "success", "message": "deps"},
{"ts": "2026-06-02T10:00:03+00:00", "step": "auth", "status": "success", "message": "auth"},
{"ts": "2026-06-02T10:00:04+00:00", "step": "pi_run", "status": "started", "message": "Running Pi"},
]
result = progress_from_events(events)
assert result["current_step"] == "pi_run"
assert result["status"] == "running"
assert 20 <= result["progress"] <= 45
assert any(step["step"] == "pi_run" and step["status"] == "running" for step in result["timeline"])
def test_progress_from_events_marks_done():
events = [
{"ts": "2026-06-02T10:00:00+00:00", "step": "bootstrap", "status": "started"},
{"ts": "2026-06-02T10:05:00+00:00", "step": "done", "status": "full_inference_success", "message": "completed"},
]
result = progress_from_events(events)
assert result["progress"] == 100
assert result["status"] == "full_inference_success"
assert any(step["step"] == "done" and step["status"] == "done" for step in result["timeline"])
def test_progress_uses_launch_state_when_events_not_ready():
result = progress_from_events([], state={"status": "running", "created_at": "2026-06-02T10:00:00+00:00"})
assert result["status"] == "running"
assert result["progress"] >= 0
assert result["current_step"] == "bootstrap"
|