| from spi.evaluation.arbiter import MasterArbiter |
| from spi.evaluation.evidence_fusion import fuse |
| from spi.evaluation.level1 import evaluate_artifact, select_evaluators |
| from spi.schemas.evidence import MetricResult |
| from spi.schemas.production import Shot |
|
|
|
|
| def _shot(**over): |
| return Shot(shot_id="SHOT_0001", scene_id="SCENE_001", |
| duration_s=4.0, fps=12, width=256, height=144, |
| entities=["CHAR_0001"], **over) |
|
|
|
|
| def _artifact_ok(): |
| return {"frames": 48, "width": 256, "height": 144, "entity_ids": ["CHAR_0001"]} |
|
|
|
|
| def test_all_pass(): |
| results = evaluate_artifact(_artifact_ok(), _shot(), "JOB_SHOT_0001") |
| assert all(r.status == "PASS" for r in results) |
| assert len(results) == 3 |
|
|
|
|
| def test_wrong_frames_early_exit(): |
| artifact = _artifact_ok() | {"frames": 10} |
| results = evaluate_artifact(artifact, _shot(), "JOB_SHOT_0001") |
| assert results[0].status == "REPAIR" |
| assert len(results) == 1 |
|
|
|
|
| def test_missing_entity_flagged_with_frames(): |
| artifact = _artifact_ok() | {"entity_ids": []} |
| results = evaluate_artifact(artifact, _shot(), "JOB_SHOT_0001") |
| ep = [r for r in results if r.metric == "entity_presence"][0] |
| assert ep.status == "REPAIR" and ep.evidence[0].start_frame is not None |
|
|
|
|
| def test_evaluator_routing_skips_irrelevant(): |
| assert "physics" not in select_evaluators({"frames": 48}, has_interaction=False) |
| assert "anatomy" in select_evaluators({"frames": 48}, has_humans=True) |
| assert "detail_4k" not in select_evaluators({"frames": 48}, resolution="draft") |
|
|
|
|
| def test_fusion_reports_disagreement(): |
| a = MetricResult(metric="m", target="S1", confidence=0.9, status="PASS", score=1.0, |
| evidence=[{"kind": "x", "finding": "ok", "source": "t", "confidence": 1.0}]) |
| b = MetricResult(metric="m", target="S2", confidence=0.5, status="REPAIR", score=0.0, |
| evidence=[{"kind": "x", "finding": "bad", "source": "t", "confidence": 0.5}]) |
| fused = fuse([a, b]) |
| assert fused["fused"]["m"]["status"] == "REPAIR" |
| assert fused["fused"]["m"]["disagreements"] == ["S2"] |
|
|
|
|
| def test_arbiter_unresolved_not_fake_pass(): |
| m = MetricResult(metric="identity", target="S1", confidence=0.3, status="UNRESOLVED", |
| uncertainties=["tracker below threshold"]) |
| d = MasterArbiter().decide([m]) |
| assert d.status == "UNRESOLVED" |
|
|
|
|
| def test_arbiter_repair_then_replan_after_two_attempts(): |
| bad = MetricResult(metric="entity_presence", target="S1", confidence=0.9, status="REPAIR", |
| score=0.0, |
| evidence=[{"kind": "identity_report", "finding": "CHAR_0001 missing", |
| "start_frame": 0, "end_frame": 48, "source": "t"}]) |
| d1 = MasterArbiter().decide([bad], repair_attempts=0) |
| assert d1.status == "REPAIR" and d1.repair_targets[0]["frames"] == (0, 48) |
| d2 = MasterArbiter().decide([bad], repair_attempts=2) |
| assert d2.status == "REPLAN" |
|
|
|
|
| def test_arbiter_pass_when_all_green(): |
| m = MetricResult(metric="frame_count", target="S1", confidence=1.0, status="PASS", |
| score=1.0, |
| evidence=[{"kind": "metadata", "finding": "ok", "source": "level1"}]) |
| assert MasterArbiter().decide([m]).status == "PASS" |
