Add manifest-driven fixture selection for degradation curves

Author: ProfRandom92

src/validation/degradation_curve_generator.py

@@ -118,24 +118,43 @@ def _load_fixture_manifest(self, manifest_path: Path = MANIFEST_PATH) -> tuple[d
             raise ValueError(f"invalid fixture manifest format: {manifest_path}")
         return tuple(fixtures)
 
-    def fixtures_for_layered_admissibility_curve(self, manifest_path: Path = MANIFEST_PATH) -> tuple[Path, ...]:
+    def fixtures_for_manifest_family(
+        self,
+        family: str,
+        levels: tuple[str, ...] = LAYERED_CURVE_LEVELS,
+        manifest_path: Path = MANIFEST_PATH,
+    ) -> tuple[Path, ...]:
         level_to_path: dict[str, Path] = {}
 
         for entry in self._load_fixture_manifest(manifest_path):
-            if entry.get("family") != LAYERED_CURVE_FAMILY:
+            if entry.get("family") != family:
                 continue
+
             level = entry.get("degradation_level")
-            if level in LAYERED_CURVE_LEVELS:
-                path_str = entry.get("path")
-                if not path_str:
-                    raise ValueError(f"missing path for fixture in manifest: {entry.get('fixture_id')}")
-                level_to_path[str(level)] = Path(path_str)
+            if level not in levels:
+                continue
+
+            path_str = entry.get("path")
+            if not path_str:
+                raise ValueError(f"missing path for fixture in manifest: {entry.get('fixture_id')}")
 
-        missing_levels = [level for level in LAYERED_CURVE_LEVELS if level not in level_to_path]
+            level_key = str(level)
+            if level_key in level_to_path:
+                raise ValueError(f"duplicate fixture for family '{family}' level '{level_key}'")
+            level_to_path[level_key] = Path(path_str)
+
+        missing_levels = [level for level in levels if level not in level_to_path]
         if missing_levels:
-            raise ValueError(f"missing layered admissibility fixtures for levels: {missing_levels}")
+            raise ValueError(f"missing fixtures for family '{family}' levels: {missing_levels}")
+
+        return tuple(level_to_path[level] for level in levels)
 
-        return tuple(level_to_path[level] for level in LAYERED_CURVE_LEVELS)
+    def fixtures_for_layered_admissibility_curve(self, manifest_path: Path = MANIFEST_PATH) -> tuple[Path, ...]:
+        return self.fixtures_for_manifest_family(
+            family=LAYERED_CURVE_FAMILY,
+            levels=LAYERED_CURVE_LEVELS,
+            manifest_path=manifest_path,
+        )
 
     def generate(self, fixtures: list[Path] | tuple[Path, ...], curve_id: str) -> DegradationCurve:
         points = tuple(self.evaluate_fixture(path) for path in fixtures)

tests/test_degradation_curve_generator.py

@@ -12,6 +12,10 @@
 MILD_FIXTURE = Path("fixtures/coding_workflow_pr_review_mild_v1")
 MODERATE_FIXTURE = Path("fixtures/coding_workflow_pr_review_moderate_v1")
 NEG_FIXTURE = Path("fixtures/coding_workflow_pr_review_degraded_v1")
+INCIDENT_POS_FIXTURE = Path("fixtures/incident_response_page_triage_v1")
+INCIDENT_MILD_FIXTURE = Path("fixtures/incident_response_page_triage_mild_v1")
+INCIDENT_MODERATE_FIXTURE = Path("fixtures/incident_response_page_triage_moderate_v1")
+INCIDENT_NEG_FIXTURE = Path("fixtures/incident_response_page_triage_degraded_v1")
 ARTIFACT_PATH = Path("artifacts/layered_admissibility_results.json")
 CURVE_ID = "coding_workflow_pr_review_curve_v1"
 
@@ -57,6 +61,74 @@ def test_layered_curve_fixtures_are_loaded_from_manifest_order() -> None:
         NEG_FIXTURE.as_posix(),
     ]
 
+
+def test_manifest_family_fixtures_for_coding_workflow_are_loaded_in_level_order() -> None:
+    fixtures = DegradationCurveGenerator().fixtures_for_manifest_family("coding_workflow_pr_review")
+    assert [fixture.as_posix() for fixture in fixtures] == [
+        POS_FIXTURE.as_posix(),
+        MILD_FIXTURE.as_posix(),
+        MODERATE_FIXTURE.as_posix(),
+        NEG_FIXTURE.as_posix(),
+    ]
+
+
+def test_manifest_family_fixtures_for_incident_response_are_loaded_in_level_order() -> None:
+    fixtures = DegradationCurveGenerator().fixtures_for_manifest_family("incident_response_page_triage")
+    assert [fixture.as_posix() for fixture in fixtures] == [
+        INCIDENT_POS_FIXTURE.as_posix(),
+        INCIDENT_MILD_FIXTURE.as_posix(),
+        INCIDENT_MODERATE_FIXTURE.as_posix(),
+        INCIDENT_NEG_FIXTURE.as_posix(),
+    ]
+
+
+def test_layered_curve_wrapper_remains_compatible_with_coding_workflow_family() -> None:
+    generator = DegradationCurveGenerator()
+    assert generator.fixtures_for_layered_admissibility_curve() == generator.fixtures_for_manifest_family(
+        "coding_workflow_pr_review"
+    )
+
+
+def test_manifest_family_selection_missing_family_or_level_raises_value_error() -> None:
+    generator = DegradationCurveGenerator()
+    with pytest.raises(ValueError, match="missing fixtures for family 'nonexistent_family' levels"):
+        generator.fixtures_for_manifest_family("nonexistent_family")
+
+    with pytest.raises(ValueError, match="missing fixtures for family 'coding_workflow_pr_review' levels"):
+        generator.fixtures_for_manifest_family("coding_workflow_pr_review", levels=("baseline", "unknown_level"))
+
+
+def test_manifest_family_selection_duplicate_level_raises_value_error(tmp_path: Path) -> None:
+    manifest_path = tmp_path / "manifest.json"
+    manifest_path.write_text(
+        json.dumps(
+            {
+                "fixtures": [
+                    {
+                        "fixture_id": "fixture_a",
+                        "family": "dup_family",
+                        "degradation_level": "baseline",
+                        "path": "fixtures/a",
+                    },
+                    {
+                        "fixture_id": "fixture_b",
+                        "family": "dup_family",
+                        "degradation_level": "baseline",
+                        "path": "fixtures/b",
+                    },
+                ]
+            }
+        ),
+        encoding="utf-8",
+    )
+
+    with pytest.raises(ValueError, match="duplicate fixture for family 'dup_family' level 'baseline'"):
+        DegradationCurveGenerator().fixtures_for_manifest_family(
+            "dup_family",
+            levels=("baseline",),
+            manifest_path=manifest_path,
+        )
+
 def test_to_dict_is_json_compatible_and_sorted() -> None:
     generator = DegradationCurveGenerator()
     curve = generator.generate(generator.fixtures_for_layered_admissibility_curve(), curve_id=CURVE_ID)