Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
39 changes: 29 additions & 10 deletions src/validation/degradation_curve_generator.py
Original file line number Diff line number Diff line change
Expand Up @@ -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):
Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

medium

The manifest is re-loaded and parsed from disk on every call to fixtures_for_manifest_family. In a high-efficiency context (as emphasized in the Repository Style Guide), especially when generating multiple curves or running a large test suite, this redundant I/O and JSON parsing can become a performance bottleneck. Consider caching the manifest content to improve efficiency.

References
  1. The project mission emphasizes high-efficiency technical-log compression and industrial-grade performance. (link)

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)
Expand Down
72 changes: 72 additions & 0 deletions tests/test_degradation_curve_generator.py
Original file line number Diff line number Diff line change
Expand Up @@ -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"

Expand Down Expand Up @@ -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)
Expand Down
Loading