diff --git a/src/validation/degradation_curve_generator.py b/src/validation/degradation_curve_generator.py index 2552c7a..4f5c82a 100644 --- a/src/validation/degradation_curve_generator.py +++ b/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) diff --git a/tests/test_degradation_curve_generator.py b/tests/test_degradation_curve_generator.py index 16796a9..43ebb52 100644 --- a/tests/test_degradation_curve_generator.py +++ b/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)