diff --git a/src/openamp_foundry/evidence/cross_batch_audit_summary.py b/src/openamp_foundry/evidence/cross_batch_audit_summary.py new file mode 100644 index 00000000..a0bd76e2 --- /dev/null +++ b/src/openamp_foundry/evidence/cross_batch_audit_summary.py @@ -0,0 +1,255 @@ +from __future__ import annotations +from dataclasses import dataclass +from typing import List + + +VALID_BATCH_GRADES = frozenset({"full", "majority", "minority", "none"}) + +LOW_COMPLETENESS_GRADES = frozenset({"minority", "none"}) + +VALID_AUDIT_VERDICTS = frozenset({ + "all_complete", + "majority_complete", + "minority_complete", + "none_complete", + "no_batches", +}) + +COMPLETENESS_AUDIT_THRESHOLD = 0.5 + + +@dataclass +class BatchSummaryEntry: + batch_id: str + bti_id: str + n_families_total: int + n_families_complete: int + n_families_partial: int + n_families_empty: int + batch_completeness_grade: str + flagged_low_completeness: bool + + +@dataclass +class CrossBatchAuditSummary: + cba2_id: str + pipeline_version: str + n_batches_total: int + n_batches_flagged: int + total_families_across_batches: int + total_families_complete: int + total_families_partial: int + total_families_empty: int + batch_entries: list[BatchSummaryEntry] + overall_audit_verdict: str + flagged_batch_ids: list[str] + audit_coverage_rate: float + dry_lab_only: bool + limitations: list[str] + created_at: str + + +def _compute_overall_audit_verdict( + n_batches_total: int, + n_full: int, +) -> str: + if n_batches_total == 0: + return "no_batches" + if n_full == n_batches_total: + return "all_complete" + if n_full >= n_batches_total / 2: + return "majority_complete" + if n_full > 0: + return "minority_complete" + return "none_complete" + + +def validate_cross_batch_audit_summary( + summary: CrossBatchAuditSummary, +) -> list[str]: + violations = [] + + if not summary.cba2_id.startswith("CBA2-"): + violations.append("cba2_id must start with 'CBA2-'") + + if not summary.dry_lab_only: + violations.append("dry_lab_only must be True") + + if summary.n_batches_total != len(summary.batch_entries): + violations.append( + f"n_batches_total ({summary.n_batches_total}) must equal len(batch_entries) ({len(summary.batch_entries)})" + ) + + expected_flagged = sum( + 1 for e in summary.batch_entries if e.flagged_low_completeness + ) + if summary.n_batches_flagged != expected_flagged: + violations.append( + f"n_batches_flagged ({summary.n_batches_flagged}) must equal count of flagged entries ({expected_flagged})" + ) + + expected_total = sum(e.n_families_total for e in summary.batch_entries) + if summary.total_families_across_batches != expected_total: + violations.append( + f"total_families_across_batches ({summary.total_families_across_batches}) must equal sum of n_families_total ({expected_total})" + ) + + expected_complete = sum(e.n_families_complete for e in summary.batch_entries) + if summary.total_families_complete != expected_complete: + violations.append( + f"total_families_complete ({summary.total_families_complete}) must equal sum of n_families_complete ({expected_complete})" + ) + + expected_partial = sum(e.n_families_partial for e in summary.batch_entries) + if summary.total_families_partial != expected_partial: + violations.append( + f"total_families_partial ({summary.total_families_partial}) must equal sum of n_families_partial ({expected_partial})" + ) + + expected_empty = sum(e.n_families_empty for e in summary.batch_entries) + if summary.total_families_empty != expected_empty: + violations.append( + f"total_families_empty ({summary.total_families_empty}) must equal sum of n_families_empty ({expected_empty})" + ) + + for entry in summary.batch_entries: + expected_flag = entry.batch_completeness_grade in LOW_COMPLETENESS_GRADES + if entry.flagged_low_completeness != expected_flag: + violations.append( + f"flagged_low_completeness ({entry.flagged_low_completeness}) for batch {entry.batch_id} must equal {expected_flag} for grade '{entry.batch_completeness_grade}'" + ) + + expected_flagged_ids = sorted( + e.batch_id for e in summary.batch_entries if e.flagged_low_completeness + ) + if summary.flagged_batch_ids != expected_flagged_ids: + violations.append( + f"flagged_batch_ids ({summary.flagged_batch_ids}) must equal {expected_flagged_ids}" + ) + + total_families = summary.total_families_across_batches + expected_coverage = ( + summary.total_families_complete / total_families if total_families > 0 else 0.0 + ) + if abs(summary.audit_coverage_rate - expected_coverage) > 0.001: + violations.append( + f"audit_coverage_rate ({summary.audit_coverage_rate}) must approximately equal {expected_coverage}" + ) + + n_full = sum( + 1 for e in summary.batch_entries if e.batch_completeness_grade == "full" + ) + expected_verdict = _compute_overall_audit_verdict( + summary.n_batches_total, n_full + ) + if summary.overall_audit_verdict != expected_verdict: + violations.append( + f"overall_audit_verdict '{summary.overall_audit_verdict}' must be '{expected_verdict}' " + f"given n_batches_total={summary.n_batches_total} and n_full={n_full}" + ) + + if not summary.limitations: + violations.append("limitations must be non-empty") + + return violations + + +def build_cross_batch_audit_summary( + cba2_id: str, + pipeline_version: str, + batch_entry_dicts: list[dict], + limitations: list[str], + created_at: str, +) -> CrossBatchAuditSummary: + batch_entries: list[BatchSummaryEntry] = [] + for d in batch_entry_dicts: + grade = d["batch_completeness_grade"] + flagged = grade in LOW_COMPLETENESS_GRADES + batch_entries.append( + BatchSummaryEntry( + batch_id=d["batch_id"], + bti_id=d["bti_id"], + n_families_total=d["n_families_total"], + n_families_complete=d["n_families_complete"], + n_families_partial=d["n_families_partial"], + n_families_empty=d["n_families_empty"], + batch_completeness_grade=grade, + flagged_low_completeness=flagged, + ) + ) + + n_batches_total = len(batch_entries) + n_batches_flagged = sum(1 for e in batch_entries if e.flagged_low_completeness) + total_families_across_batches = sum(e.n_families_total for e in batch_entries) + total_families_complete = sum(e.n_families_complete for e in batch_entries) + total_families_partial = sum(e.n_families_partial for e in batch_entries) + total_families_empty = sum(e.n_families_empty for e in batch_entries) + audit_coverage_rate = ( + total_families_complete / total_families_across_batches + if total_families_across_batches > 0 + else 0.0 + ) + flagged_batch_ids = sorted( + e.batch_id for e in batch_entries if e.flagged_low_completeness + ) + n_full = sum( + 1 for e in batch_entries if e.batch_completeness_grade == "full" + ) + overall_audit_verdict = _compute_overall_audit_verdict( + n_batches_total, n_full + ) + + summary = CrossBatchAuditSummary( + cba2_id=cba2_id, + pipeline_version=pipeline_version, + n_batches_total=n_batches_total, + n_batches_flagged=n_batches_flagged, + total_families_across_batches=total_families_across_batches, + total_families_complete=total_families_complete, + total_families_partial=total_families_partial, + total_families_empty=total_families_empty, + batch_entries=batch_entries, + overall_audit_verdict=overall_audit_verdict, + flagged_batch_ids=flagged_batch_ids, + audit_coverage_rate=audit_coverage_rate, + dry_lab_only=True, + limitations=limitations, + created_at=created_at, + ) + + violations = validate_cross_batch_audit_summary(summary) + if violations: + raise ValueError(f"Invalid CrossBatchAuditSummary: {violations}") + return summary + + +def format_cross_batch_audit_summary(summary: CrossBatchAuditSummary) -> str: + flagged_str = ", ".join(summary.flagged_batch_ids) if summary.flagged_batch_ids else "(none)" + limitations_str = "; ".join(summary.limitations) + lines = [ + f"Cross-Batch Audit Summary — {summary.cba2_id}", + f"Pipeline: {summary.pipeline_version} | Created: {summary.created_at}", + f"Batches: {summary.n_batches_total} total, {summary.n_batches_flagged} flagged", + f"Audit Verdict: {summary.overall_audit_verdict} | " + f"Coverage Rate: {summary.audit_coverage_rate:.3f}", + f"Families — Complete: {summary.total_families_complete} | " + f"Partial: {summary.total_families_partial} | " + f"Empty: {summary.total_families_empty} | " + f"Total: {summary.total_families_across_batches}", + f"Flagged Batches: [{flagged_str}]", + f"Limitations: {limitations_str}", + "dry_lab_only: True", + "", + "Batch Entries:", + ] + for entry in summary.batch_entries: + flag_marker = " [FLAGGED]" if entry.flagged_low_completeness else "" + lines.append( + f" {entry.batch_id} ({entry.bti_id}): " + f"grade={entry.batch_completeness_grade}, " + f"complete={entry.n_families_complete}, " + f"partial={entry.n_families_partial}, " + f"empty={entry.n_families_empty}, " + f"total={entry.n_families_total}{flag_marker}" + ) + return "\n".join(lines) diff --git a/tests/evidence/test_cross_batch_audit_summary.py b/tests/evidence/test_cross_batch_audit_summary.py new file mode 100644 index 00000000..51179455 --- /dev/null +++ b/tests/evidence/test_cross_batch_audit_summary.py @@ -0,0 +1,684 @@ +"""Tests for cross_batch_audit_summary module — CBA2- schema.""" + +from __future__ import annotations + +import pytest + +from openamp_foundry.evidence.cross_batch_audit_summary import ( + BatchSummaryEntry, + CrossBatchAuditSummary, + VALID_BATCH_GRADES, + LOW_COMPLETENESS_GRADES, + VALID_AUDIT_VERDICTS, + COMPLETENESS_AUDIT_THRESHOLD, + validate_cross_batch_audit_summary, + build_cross_batch_audit_summary, + format_cross_batch_audit_summary, +) + + +# --------------------------------------------------------------------------- +# helpers +# --------------------------------------------------------------------------- + +def _make_full_entry(batch_id="B001", bti_id="BTI-001", n=3): + return {"batch_id": batch_id, "bti_id": bti_id, "n_families_total": n, + "n_families_complete": n, "n_families_partial": 0, "n_families_empty": 0, + "batch_completeness_grade": "full"} + + +def _make_none_entry(batch_id="B002", bti_id="BTI-002", n=3): + return {"batch_id": batch_id, "bti_id": bti_id, "n_families_total": n, + "n_families_complete": 0, "n_families_partial": n, "n_families_empty": 0, + "batch_completeness_grade": "none"} + + +def _make_minority_entry(batch_id="B003", bti_id="BTI-003", n=3): + return {"batch_id": batch_id, "bti_id": bti_id, "n_families_total": n, + "n_families_complete": 1, "n_families_partial": 2, "n_families_empty": 0, + "batch_completeness_grade": "minority"} + + +def _make_majority_entry(batch_id="B004", bti_id="BTI-004", n=4): + return {"batch_id": batch_id, "bti_id": bti_id, "n_families_total": n, + "n_families_complete": 3, "n_families_partial": 1, "n_families_empty": 0, + "batch_completeness_grade": "majority"} + + +# --------------------------------------------------------------------------- +# Class 1 — BatchSummaryEntry dataclass +# --------------------------------------------------------------------------- + +class TestBatchSummaryEntry: + + def test_full_entry_flagged_false(self): + entry = BatchSummaryEntry( + batch_id="B001", bti_id="BTI-001", + n_families_total=3, n_families_complete=3, + n_families_partial=0, n_families_empty=0, + batch_completeness_grade="full", flagged_low_completeness=False, + ) + assert entry.batch_completeness_grade == "full" + assert entry.flagged_low_completeness is False + + def test_none_entry_flagged_true(self): + entry = BatchSummaryEntry( + batch_id="B002", bti_id="BTI-002", + n_families_total=3, n_families_complete=0, + n_families_partial=3, n_families_empty=0, + batch_completeness_grade="none", flagged_low_completeness=True, + ) + assert entry.batch_completeness_grade == "none" + assert entry.flagged_low_completeness is True + + def test_minority_entry_flagged_true(self): + entry = BatchSummaryEntry( + batch_id="B003", bti_id="BTI-003", + n_families_total=3, n_families_complete=1, + n_families_partial=2, n_families_empty=0, + batch_completeness_grade="minority", flagged_low_completeness=True, + ) + assert entry.batch_completeness_grade == "minority" + assert entry.flagged_low_completeness is True + + def test_majority_entry_flagged_false(self): + entry = BatchSummaryEntry( + batch_id="B004", bti_id="BTI-004", + n_families_total=4, n_families_complete=3, + n_families_partial=1, n_families_empty=0, + batch_completeness_grade="majority", flagged_low_completeness=False, + ) + assert entry.batch_completeness_grade == "majority" + assert entry.flagged_low_completeness is False + + def test_batch_completeness_grade_in_valid_set(self): + for grade in ("full", "majority", "minority", "none"): + entry = BatchSummaryEntry( + batch_id="B001", bti_id="BTI-001", + n_families_total=1, n_families_complete=1, + n_families_partial=0, n_families_empty=0, + batch_completeness_grade=grade, flagged_low_completeness=False, + ) + assert entry.batch_completeness_grade in VALID_BATCH_GRADES + + def test_bti_id_field_present(self): + entry = BatchSummaryEntry( + batch_id="B001", bti_id="BTI-001", + n_families_total=3, n_families_complete=3, + n_families_partial=0, n_families_empty=0, + batch_completeness_grade="full", flagged_low_completeness=False, + ) + assert hasattr(entry, "bti_id") + assert entry.bti_id == "BTI-001" + + +# --------------------------------------------------------------------------- +# Class 2 — validate_cross_batch_audit_summary +# --------------------------------------------------------------------------- + +class TestValidateCrossBatchAuditSummary: + + def _valid_summary(self, **overrides): + d = dict( + cba2_id="CBA2-001", + pipeline_version="v1.0", + n_batches_total=2, + n_batches_flagged=0, + total_families_across_batches=6, + total_families_complete=6, + total_families_partial=0, + total_families_empty=0, + batch_entries=[ + BatchSummaryEntry( + batch_id="B001", bti_id="BTI-001", + n_families_total=3, n_families_complete=3, + n_families_partial=0, n_families_empty=0, + batch_completeness_grade="full", flagged_low_completeness=False, + ), + BatchSummaryEntry( + batch_id="B002", bti_id="BTI-002", + n_families_total=3, n_families_complete=3, + n_families_partial=0, n_families_empty=0, + batch_completeness_grade="full", flagged_low_completeness=False, + ), + ], + overall_audit_verdict="all_complete", + flagged_batch_ids=[], + audit_coverage_rate=1.0, + dry_lab_only=True, + limitations=["toy data only"], + created_at="2025-01-01T00:00:00Z", + ) + d.update(overrides) + return CrossBatchAuditSummary(**d) + + def test_valid_all_complete(self): + violations = validate_cross_batch_audit_summary(self._valid_summary()) + assert violations == [] + + def test_valid_no_batches(self): + s = self._valid_summary( + n_batches_total=0, n_batches_flagged=0, + total_families_across_batches=0, total_families_complete=0, + total_families_partial=0, total_families_empty=0, + batch_entries=[], overall_audit_verdict="no_batches", + flagged_batch_ids=[], audit_coverage_rate=0.0, + ) + violations = validate_cross_batch_audit_summary(s) + assert violations == [] + + def test_invalid_cba2_id_prefix(self): + s = self._valid_summary(cba2_id="INVALID-001") + violations = validate_cross_batch_audit_summary(s) + assert any("cba2_id must start with 'CBA2-'" in v for v in violations) + + def test_dry_lab_only_false(self): + s = self._valid_summary(dry_lab_only=False) + violations = validate_cross_batch_audit_summary(s) + assert any("dry_lab_only must be True" in v for v in violations) + + def test_n_batches_total_mismatch(self): + s = self._valid_summary(n_batches_total=99) + violations = validate_cross_batch_audit_summary(s) + assert any("n_batches_total" in v for v in violations) + + def test_n_batches_flagged_mismatch(self): + s = self._valid_summary(n_batches_flagged=99) + violations = validate_cross_batch_audit_summary(s) + assert any("n_batches_flagged" in v for v in violations) + + def test_total_families_across_batches_mismatch(self): + s = self._valid_summary(total_families_across_batches=99) + violations = validate_cross_batch_audit_summary(s) + assert any("total_families_across_batches" in v for v in violations) + + def test_total_families_complete_mismatch(self): + s = self._valid_summary(total_families_complete=99) + violations = validate_cross_batch_audit_summary(s) + assert any("total_families_complete" in v for v in violations) + + def test_total_families_partial_mismatch(self): + s = self._valid_summary(total_families_partial=99) + violations = validate_cross_batch_audit_summary(s) + assert any("total_families_partial" in v for v in violations) + + def test_total_families_empty_mismatch(self): + s = self._valid_summary(total_families_empty=99) + violations = validate_cross_batch_audit_summary(s) + assert any("total_families_empty" in v for v in violations) + + def test_entry_flagged_inconsistent_with_grade(self): + entry = BatchSummaryEntry( + batch_id="B001", bti_id="BTI-001", + n_families_total=3, n_families_complete=0, + n_families_partial=3, n_families_empty=0, + batch_completeness_grade="none", flagged_low_completeness=False, + ) + s = self._valid_summary( + batch_entries=[entry], + n_batches_total=1, n_batches_flagged=0, + total_families_across_batches=3, total_families_complete=0, + total_families_partial=3, total_families_empty=0, + overall_audit_verdict="none_complete", + flagged_batch_ids=[], audit_coverage_rate=0.0, + ) + violations = validate_cross_batch_audit_summary(s) + assert any("flagged_low_completeness" in v for v in violations) + + def test_flagged_batch_ids_not_sorted(self): + entries = [ + BatchSummaryEntry( + batch_id="B003", bti_id="BTI-003", + n_families_total=3, n_families_complete=0, + n_families_partial=3, n_families_empty=0, + batch_completeness_grade="none", flagged_low_completeness=True, + ), + BatchSummaryEntry( + batch_id="B002", bti_id="BTI-002", + n_families_total=3, n_families_complete=0, + n_families_partial=3, n_families_empty=0, + batch_completeness_grade="none", flagged_low_completeness=True, + ), + ] + s = self._valid_summary( + batch_entries=entries, + n_batches_total=2, n_batches_flagged=2, + total_families_across_batches=6, total_families_complete=0, + total_families_partial=6, total_families_empty=0, + overall_audit_verdict="none_complete", + flagged_batch_ids=["B003", "B002"], + audit_coverage_rate=0.0, + ) + violations = validate_cross_batch_audit_summary(s) + assert any("flagged_batch_ids" in v for v in violations) + + def test_flagged_batch_ids_missing_flagged(self): + entry = BatchSummaryEntry( + batch_id="B002", bti_id="BTI-002", + n_families_total=3, n_families_complete=0, + n_families_partial=3, n_families_empty=0, + batch_completeness_grade="none", flagged_low_completeness=True, + ) + s = self._valid_summary( + batch_entries=[entry], + n_batches_total=1, n_batches_flagged=1, + total_families_across_batches=3, total_families_complete=0, + total_families_partial=3, total_families_empty=0, + overall_audit_verdict="none_complete", + flagged_batch_ids=[], + audit_coverage_rate=0.0, + ) + violations = validate_cross_batch_audit_summary(s) + assert any("flagged_batch_ids" in v for v in violations) + + def test_audit_coverage_rate_mismatch(self): + s = self._valid_summary(audit_coverage_rate=0.5) + violations = validate_cross_batch_audit_summary(s) + assert any("audit_coverage_rate" in v for v in violations) + + def test_verdict_all_complete_with_not_all_full_fails(self): + entries = [ + BatchSummaryEntry( + batch_id="B001", bti_id="BTI-001", + n_families_total=3, n_families_complete=3, + n_families_partial=0, n_families_empty=0, + batch_completeness_grade="full", flagged_low_completeness=False, + ), + BatchSummaryEntry( + batch_id="B002", bti_id="BTI-002", + n_families_total=3, n_families_complete=0, + n_families_partial=3, n_families_empty=0, + batch_completeness_grade="none", flagged_low_completeness=True, + ), + ] + s = self._valid_summary( + batch_entries=entries, + n_batches_total=2, n_batches_flagged=1, + total_families_across_batches=6, total_families_complete=3, + total_families_partial=3, total_families_empty=0, + overall_audit_verdict="all_complete", + flagged_batch_ids=["B002"], + audit_coverage_rate=0.5, + ) + violations = validate_cross_batch_audit_summary(s) + assert any("overall_audit_verdict" in v for v in violations) + + def test_verdict_none_complete_with_some_full_fails(self): + entries = [ + BatchSummaryEntry( + batch_id="B001", bti_id="BTI-001", + n_families_total=3, n_families_complete=3, + n_families_partial=0, n_families_empty=0, + batch_completeness_grade="full", flagged_low_completeness=False, + ), + ] + s = self._valid_summary( + batch_entries=entries, + n_batches_total=1, n_batches_flagged=0, + total_families_across_batches=3, total_families_complete=3, + total_families_partial=0, total_families_empty=0, + overall_audit_verdict="none_complete", + flagged_batch_ids=[], audit_coverage_rate=1.0, + ) + violations = validate_cross_batch_audit_summary(s) + assert any("overall_audit_verdict" in v for v in violations) + + def test_verdict_no_batches_with_n_positive_fails(self): + entry = BatchSummaryEntry( + batch_id="B001", bti_id="BTI-001", + n_families_total=3, n_families_complete=3, + n_families_partial=0, n_families_empty=0, + batch_completeness_grade="full", flagged_low_completeness=False, + ) + s = self._valid_summary( + batch_entries=[entry], + n_batches_total=1, n_batches_flagged=0, + total_families_across_batches=3, total_families_complete=3, + total_families_partial=0, total_families_empty=0, + overall_audit_verdict="no_batches", + flagged_batch_ids=[], audit_coverage_rate=1.0, + ) + violations = validate_cross_batch_audit_summary(s) + assert any("overall_audit_verdict" in v for v in violations) + + def test_limitations_empty_fails(self): + s = self._valid_summary(limitations=[]) + violations = validate_cross_batch_audit_summary(s) + assert any("limitations" in v for v in violations) + + +# --------------------------------------------------------------------------- +# Class 3 — build_cross_batch_audit_summary +# --------------------------------------------------------------------------- + +class TestBuildCrossBatchAuditSummary: + + def test_three_full_batches(self): + d = [ + _make_full_entry("B001", "BTI-001"), + _make_full_entry("B002", "BTI-002"), + _make_full_entry("B003", "BTI-003"), + ] + s = build_cross_batch_audit_summary( + "CBA2-001", "v1.0", d, ["toy data"], "2025-01-01T00:00:00Z", + ) + assert s.overall_audit_verdict == "all_complete" + assert s.n_batches_flagged == 0 + + def test_zero_batches(self): + s = build_cross_batch_audit_summary( + "CBA2-001", "v1.0", [], ["toy data"], "2025-01-01T00:00:00Z", + ) + assert s.overall_audit_verdict == "no_batches" + assert s.n_batches_total == 0 + assert s.total_families_across_batches == 0 + assert s.total_families_complete == 0 + assert s.total_families_partial == 0 + assert s.total_families_empty == 0 + + def test_two_full_one_none(self): + d = [ + _make_full_entry("B001", "BTI-001"), + _make_full_entry("B002", "BTI-002"), + _make_none_entry("B003", "BTI-003"), + ] + s = build_cross_batch_audit_summary( + "CBA2-001", "v1.0", d, ["toy data"], "2025-01-01T00:00:00Z", + ) + assert s.overall_audit_verdict == "majority_complete" + assert s.n_batches_flagged == 1 + + def test_one_full_two_none(self): + d = [ + _make_full_entry("B001", "BTI-001"), + _make_none_entry("B002", "BTI-002"), + _make_none_entry("B003", "BTI-003"), + ] + s = build_cross_batch_audit_summary( + "CBA2-001", "v1.0", d, ["toy data"], "2025-01-01T00:00:00Z", + ) + assert s.overall_audit_verdict == "minority_complete" + assert s.n_batches_flagged == 2 + + def test_three_none(self): + d = [ + _make_none_entry("B001", "BTI-001"), + _make_none_entry("B002", "BTI-002"), + _make_none_entry("B003", "BTI-003"), + ] + s = build_cross_batch_audit_summary( + "CBA2-001", "v1.0", d, ["toy data"], "2025-01-01T00:00:00Z", + ) + assert s.overall_audit_verdict == "none_complete" + assert s.n_batches_flagged == 3 + + def test_dry_lab_only_auto_true(self): + d = [_make_full_entry()] + s = build_cross_batch_audit_summary( + "CBA2-001", "v1.0", d, ["toy"], "2025-01-01T00:00:00Z", + ) + assert s.dry_lab_only is True + + def test_n_batches_total_matches_len(self): + d = [_make_full_entry("B001"), _make_full_entry("B002")] + s = build_cross_batch_audit_summary( + "CBA2-001", "v1.0", d, ["toy"], "2025-01-01T00:00:00Z", + ) + assert s.n_batches_total == len(d) + + def test_total_families_across_batches_sum(self): + d = [ + _make_full_entry("B001", n=3), + _make_full_entry("B002", n=5), + ] + s = build_cross_batch_audit_summary( + "CBA2-001", "v1.0", d, ["toy"], "2025-01-01T00:00:00Z", + ) + assert s.total_families_across_batches == 8 + + def test_total_families_complete_sum(self): + d = [ + _make_full_entry("B001", n=3), + _make_minority_entry("B002", n=3), + ] + s = build_cross_batch_audit_summary( + "CBA2-001", "v1.0", d, ["toy"], "2025-01-01T00:00:00Z", + ) + assert s.total_families_complete == 4 + + def test_total_families_partial_sum(self): + d = [ + _make_full_entry("B001", n=3), + _make_minority_entry("B002", n=3), + ] + s = build_cross_batch_audit_summary( + "CBA2-001", "v1.0", d, ["toy"], "2025-01-01T00:00:00Z", + ) + assert s.total_families_partial == 2 + + def test_total_families_empty_sum(self): + d = [ + _make_full_entry("B001", n=3), + _make_full_entry("B002", n=3), + ] + s = build_cross_batch_audit_summary( + "CBA2-001", "v1.0", d, ["toy"], "2025-01-01T00:00:00Z", + ) + assert s.total_families_empty == 0 + + def test_n_batches_flagged_counted_correctly(self): + d = [ + _make_full_entry("B001"), + _make_minority_entry("B002"), + _make_none_entry("B003"), + ] + s = build_cross_batch_audit_summary( + "CBA2-001", "v1.0", d, ["toy"], "2025-01-01T00:00:00Z", + ) + assert s.n_batches_flagged == 2 + + def test_flagged_batch_ids_sorted(self): + d = [ + _make_none_entry("B003"), + _make_full_entry("B001"), + _make_minority_entry("B002"), + ] + s = build_cross_batch_audit_summary( + "CBA2-001", "v1.0", d, ["toy"], "2025-01-01T00:00:00Z", + ) + assert s.flagged_batch_ids == ["B002", "B003"] + + def test_audit_coverage_rate_computed(self): + d = [ + _make_full_entry("B001", n=4), + _make_minority_entry("B002", n=4), + ] + s = build_cross_batch_audit_summary( + "CBA2-001", "v1.0", d, ["toy"], "2025-01-01T00:00:00Z", + ) + assert s.audit_coverage_rate == pytest.approx(5 / 8) + + def test_one_full_one_full_all_complete(self): + d = [ + _make_full_entry("B001"), + _make_full_entry("B002"), + ] + s = build_cross_batch_audit_summary( + "CBA2-001", "v1.0", d, ["toy"], "2025-01-01T00:00:00Z", + ) + assert s.overall_audit_verdict == "all_complete" + + def test_one_full_one_majority_majority_complete(self): + d = [ + _make_full_entry("B001"), + _make_majority_entry("B002"), + ] + s = build_cross_batch_audit_summary( + "CBA2-001", "v1.0", d, ["toy"], "2025-01-01T00:00:00Z", + ) + assert s.overall_audit_verdict == "majority_complete" + + def test_zero_full_but_n_positive_none_complete(self): + d = [ + _make_minority_entry("B001"), + _make_none_entry("B002"), + ] + s = build_cross_batch_audit_summary( + "CBA2-001", "v1.0", d, ["toy"], "2025-01-01T00:00:00Z", + ) + assert s.overall_audit_verdict == "none_complete" + + def test_audit_coverage_rate_zero_when_zero_families(self): + s = build_cross_batch_audit_summary( + "CBA2-001", "v1.0", [], ["toy"], "2025-01-01T00:00:00Z", + ) + assert s.audit_coverage_rate == 0.0 + + +# --------------------------------------------------------------------------- +# Class 4 — Integration +# --------------------------------------------------------------------------- + +class TestCrossBatchAuditSummaryIntegration: + + def test_full_round_trip_build_validate(self): + d = [ + _make_full_entry("B001"), + _make_none_entry("B002"), + _make_minority_entry("B003"), + ] + s = build_cross_batch_audit_summary( + "CBA2-001", "v1.0", d, ["toy data"], "2025-01-01T00:00:00Z", + ) + violations = validate_cross_batch_audit_summary(s) + assert violations == [] + + def test_flagged_batches_are_minority_or_none(self): + d = [ + _make_full_entry("B001"), + _make_majority_entry("B002"), + _make_minority_entry("B003"), + _make_none_entry("B004"), + ] + s = build_cross_batch_audit_summary( + "CBA2-001", "v1.0", d, ["toy"], "2025-01-01T00:00:00Z", + ) + assert s.n_batches_flagged == 2 + assert all( + e.batch_completeness_grade in LOW_COMPLETENESS_GRADES + for e in s.batch_entries if e.flagged_low_completeness + ) + + def test_flagged_batch_ids_are_sorted(self): + d = [ + _make_none_entry("Z001"), + _make_minority_entry("A001"), + ] + s = build_cross_batch_audit_summary( + "CBA2-001", "v1.0", d, ["toy"], "2025-01-01T00:00:00Z", + ) + assert s.flagged_batch_ids == sorted(s.flagged_batch_ids) + + def test_all_complete_requires_all_full(self): + d = [ + _make_full_entry("B001"), + _make_majority_entry("B002"), + ] + s = build_cross_batch_audit_summary( + "CBA2-001", "v1.0", d, ["toy"], "2025-01-01T00:00:00Z", + ) + assert s.overall_audit_verdict != "all_complete" + + def test_audit_coverage_rate_in_range(self): + d = [ + _make_full_entry("B001"), + _make_none_entry("B002"), + _make_minority_entry("B003"), + _make_majority_entry("B004"), + ] + s = build_cross_batch_audit_summary( + "CBA2-001", "v1.0", d, ["toy"], "2025-01-01T00:00:00Z", + ) + assert 0.0 <= s.audit_coverage_rate <= 1.0 + + def test_multiple_batches_different_grades(self): + d = [ + _make_full_entry("B001"), + _make_majority_entry("B002"), + _make_minority_entry("B003"), + _make_none_entry("B004"), + ] + s = build_cross_batch_audit_summary( + "CBA2-001", "v1.0", d, ["toy"], "2025-01-01T00:00:00Z", + ) + grades = {e.batch_id: e.batch_completeness_grade for e in s.batch_entries} + assert grades["B001"] == "full" + assert grades["B002"] == "majority" + assert grades["B003"] == "minority" + assert grades["B004"] == "none" + + def test_majority_complete_at_exactly_half(self): + d = [ + _make_full_entry("B001"), + _make_full_entry("B002"), + _make_none_entry("B003"), + _make_none_entry("B004"), + ] + s = build_cross_batch_audit_summary( + "CBA2-001", "v1.0", d, ["toy"], "2025-01-01T00:00:00Z", + ) + assert s.overall_audit_verdict == "majority_complete" + + def test_empty_dicts_no_batches(self): + s = build_cross_batch_audit_summary( + "CBA2-001", "v1.0", [], ["toy"], "2025-01-01T00:00:00Z", + ) + assert s.overall_audit_verdict == "no_batches" + assert s.n_batches_total == 0 + + +# --------------------------------------------------------------------------- +# Class 5 — format_cross_batch_audit_summary +# --------------------------------------------------------------------------- + +class TestFormatCrossBatchAuditSummary: + + def test_returns_string(self): + d = [_make_full_entry()] + s = build_cross_batch_audit_summary( + "CBA2-001", "v1.0", d, ["toy"], "2025-01-01T00:00:00Z", + ) + result = format_cross_batch_audit_summary(s) + assert isinstance(result, str) + + def test_contains_cba2_id(self): + d = [_make_full_entry()] + s = build_cross_batch_audit_summary( + "CBA2-001", "v1.0", d, ["toy"], "2025-01-01T00:00:00Z", + ) + result = format_cross_batch_audit_summary(s) + assert "CBA2-001" in result + + def test_contains_overall_audit_verdict(self): + d = [_make_full_entry()] + s = build_cross_batch_audit_summary( + "CBA2-001", "v1.0", d, ["toy"], "2025-01-01T00:00:00Z", + ) + result = format_cross_batch_audit_summary(s) + assert "all_complete" in result + + def test_contains_n_batches_total(self): + d = [_make_full_entry(), _make_none_entry()] + s = build_cross_batch_audit_summary( + "CBA2-001", "v1.0", d, ["toy"], "2025-01-01T00:00:00Z", + ) + result = format_cross_batch_audit_summary(s) + assert "2 total" in result + + def test_contains_n_batches_flagged(self): + d = [_make_full_entry(), _make_none_entry()] + s = build_cross_batch_audit_summary( + "CBA2-001", "v1.0", d, ["toy"], "2025-01-01T00:00:00Z", + ) + result = format_cross_batch_audit_summary(s) + assert "1 flagged" in result diff --git a/tests/test_test_count_regression.py b/tests/test_test_count_regression.py index 724003d4..d0c896f7 100644 --- a/tests/test_test_count_regression.py +++ b/tests/test_test_count_regression.py @@ -3,7 +3,7 @@ import subprocess import math -BASELINE = 9576 +BASELINE = 9639 def test_test_count_regression():