| 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"] == "pi_run" | |
| assert [step["label"] for step in view["pipeline"]][:3] == ["Bootstrap", "Dependencies", "Auth"] | |
| 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" | |