File size: 3,597 Bytes
b23082b fab8b21 b23082b | 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 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 | from datetime import datetime, timezone
from src.view_models import build_run_view_model, find_latest_resumable_run, normalize_run_status
def _bundle(status="running", events=None, **extra):
bundle = {
"summary": {
"run_id": "run-1",
"status": status,
"model_id": "org/model",
"target_space": "alice/demo",
"target_space_url": "https://huggingface.co/spaces/alice/demo",
"created_at": "2026-06-03T10:00:00+00:00",
"updated_at": "2026-06-03T10:01:00+00:00",
"selected_hardware": "zero-a10g",
},
"state": {"status": status},
"launch": {"status": status, "job_id": "job-1", "created_by": "alice"},
"events": events or [],
"inference_gate": {},
"generation_smoke": {},
"hardware_strategy": {},
"technical_blockers": {},
}
bundle.update(extra)
return bundle
def test_view_model_maps_running_run_to_product_pipeline():
bundle = _bundle(
events=[
{"ts": "2026-06-03T10:00:10+00:00", "step": "model_analysis", "status": "success", "message": "Analyzed model"},
{"ts": "2026-06-03T10:00:20+00:00", "step": "pi_run", "status": "running", "message": "Generating files"},
]
)
view = build_run_view_model("run-1", bundle, bucket_source="alice/space-factory-runs", now=datetime(2026, 6, 3, 10, 2, tzinfo=timezone.utc))
assert view["header"]["status"] == "running"
assert view["header"]["current_phase"] == "generate_app"
assert [step["label"] for step in view["pipeline"]][:3] == ["Bucket ready", "Job launched", "Model analysis"]
assert any(item["agent"] == "Coder Agent" for item in view["activity"])
def test_view_model_manual_hardware_is_waiting_manual_action():
bundle = _bundle(status="manual_hardware_required", hardware_strategy={"manual_action_required": True})
view = build_run_view_model("run-1", bundle, bucket_source="alice/space-factory-runs", now=datetime(2026, 6, 3, 10, 2, tzinfo=timezone.utc))
assert view["header"]["status"] == "waiting_manual_action"
assert view["actions"]["requires_manual_action"] is True
assert view["diagnostics"]["issue"]["status"] == "action_required"
def test_view_model_success_marks_pipeline_completed_and_live_test_passed():
bundle = _bundle(status="full_inference_success", generation_smoke={"ok": True, "latency_seconds": 12.4})
view = build_run_view_model("run-1", bundle, bucket_source="alice/space-factory-runs", now=datetime(2026, 6, 3, 10, 2, tzinfo=timezone.utc))
assert view["header"]["status"] == "succeeded"
assert all(step["status"] == "completed" for step in view["pipeline"])
assert view["space_test"]["status"] == "passed"
def test_view_model_marks_old_non_terminal_run_as_stale():
bundle = _bundle(status="running")
status = normalize_run_status(bundle, now=datetime(2026, 6, 4, 10, 2, tzinfo=timezone.utc))
assert status["global_status"] == "stale"
assert status["is_stale"] is True
def test_find_latest_resumable_run_ignores_terminal_and_manual():
runs = [
{"run_id": "old-success", "status": "full_inference_success", "updated_at": "2026-06-03T10:00:00+00:00"},
{"run_id": "manual", "status": "manual_hardware_required", "updated_at": "2026-06-03T10:03:00+00:00"},
{"run_id": "active", "status": "running", "updated_at": "2026-06-03T10:02:00+00:00"},
]
picked = find_latest_resumable_run(runs, now=datetime(2026, 6, 3, 10, 4, tzinfo=timezone.utc))
assert picked["run_id"] == "active"
|