diff --git a/src/openamp_foundry/evidence/family_evidence_trail_view.py b/src/openamp_foundry/evidence/family_evidence_trail_view.py new file mode 100644 index 00000000..0fd7c41e --- /dev/null +++ b/src/openamp_foundry/evidence/family_evidence_trail_view.py @@ -0,0 +1,162 @@ +"""FET- family evidence trail view schema. + +Per-family aggregation of all evidence schema references into a timeline view. +Shows which schemas are present, missing, and when produced. +""" + +from __future__ import annotations + +from dataclasses import dataclass, field + +REQUIRED_TRAIL_SCHEMA_TYPES: frozenset[str] = frozenset({ + "BTI", "BEG", "SAT", "PCC", + "BSP", "SEG", "ECC", "RSR", "PQT", +}) + +VALID_TRAIL_COMPLETENESS_GRADES: frozenset[str] = frozenset({ + "complete", "partial", "empty", +}) + + +@dataclass +class FamilySchemaEvent: + schema_type: str + artifact_id: str + created_at: str + + +@dataclass +class FamilyEvidenceTrailView: + fet_id: str + candidate_family_id: str + batch_id: str + pipeline_version: str + schema_events: list[FamilySchemaEvent] + present_schema_types: list[str] + missing_schema_types: list[str] + n_schemas_present: int + n_schemas_required: int + trail_completeness_grade: str + dry_lab_only: bool + limitations: list[str] + created_at: str + + +def _compute_grade(n_present: int, n_required: int) -> str: + if n_present == 0: + return "empty" + if n_present == n_required: + return "complete" + return "partial" + + +def validate_family_evidence_trail_view(view: FamilyEvidenceTrailView) -> None: + if not view.fet_id.startswith("FET-"): + raise ValueError(f"fet_id must start with 'FET-': {view.fet_id!r}") + if view.candidate_family_id.startswith("TOY-"): + raise ValueError( + f"candidate_family_id must not start with 'TOY-': {view.candidate_family_id!r}" + ) + if not view.candidate_family_id: + raise ValueError("candidate_family_id must be non-empty") + if not view.batch_id: + raise ValueError("batch_id must be non-empty") + if not view.pipeline_version: + raise ValueError("pipeline_version must be non-empty") + for event in view.schema_events: + if event.schema_type not in REQUIRED_TRAIL_SCHEMA_TYPES: + raise ValueError( + f"schema_type {event.schema_type!r} not in REQUIRED_TRAIL_SCHEMA_TYPES" + ) + if not event.artifact_id: + raise ValueError( + f"artifact_id must be non-empty for schema_type {event.schema_type!r}" + ) + if view.trail_completeness_grade not in VALID_TRAIL_COMPLETENESS_GRADES: + raise ValueError( + f"trail_completeness_grade {view.trail_completeness_grade!r} not in " + f"VALID_TRAIL_COMPLETENESS_GRADES" + ) + if not view.dry_lab_only: + raise ValueError("dry_lab_only must be True") + if not view.limitations: + raise ValueError("limitations must be non-empty") + if not view.created_at: + raise ValueError("created_at must be non-empty") + n_req = len(REQUIRED_TRAIL_SCHEMA_TYPES) + if view.n_schemas_required != n_req: + raise ValueError( + f"n_schemas_required must be {n_req}, got {view.n_schemas_required}" + ) + if view.n_schemas_present != len(view.present_schema_types): + raise ValueError("n_schemas_present must equal len(present_schema_types)") + expected_missing = sorted( + REQUIRED_TRAIL_SCHEMA_TYPES - set(view.present_schema_types) + ) + if view.missing_schema_types != expected_missing: + raise ValueError( + f"missing_schema_types mismatch: expected {expected_missing}, " + f"got {view.missing_schema_types}" + ) + + +def build_family_evidence_trail_view( + *, + fet_id: str, + candidate_family_id: str, + batch_id: str, + pipeline_version: str, + schema_event_dicts: list[dict], + limitations: list[str], + created_at: str, +) -> FamilyEvidenceTrailView: + events = [ + FamilySchemaEvent( + schema_type=d["schema_type"], + artifact_id=d["artifact_id"], + created_at=d["created_at"], + ) + for d in schema_event_dicts + ] + present_types = sorted({e.schema_type for e in events}) + missing_types = sorted(REQUIRED_TRAIL_SCHEMA_TYPES - set(present_types)) + n_required = len(REQUIRED_TRAIL_SCHEMA_TYPES) + n_present = len(present_types) + grade = _compute_grade(n_present, n_required) + view = FamilyEvidenceTrailView( + fet_id=fet_id, + candidate_family_id=candidate_family_id, + batch_id=batch_id, + pipeline_version=pipeline_version, + schema_events=events, + present_schema_types=present_types, + missing_schema_types=missing_types, + n_schemas_present=n_present, + n_schemas_required=n_required, + trail_completeness_grade=grade, + dry_lab_only=True, + limitations=limitations, + created_at=created_at, + ) + validate_family_evidence_trail_view(view) + return view + + +def format_family_evidence_trail_view(view: FamilyEvidenceTrailView) -> str: + lines = [ + f"Family Evidence Trail View — {view.fet_id}", + f"Family: {view.candidate_family_id} | Batch: {view.batch_id} | Pipeline: {view.pipeline_version}", + f"Grade: {view.trail_completeness_grade} | Present: {view.n_schemas_present}/{view.n_schemas_required}", + ] + if view.present_schema_types: + lines.append(f"Present: {', '.join(view.present_schema_types)}") + if view.missing_schema_types: + lines.append(f"Missing: {', '.join(view.missing_schema_types)}") + lines.append(f"Created: {view.created_at}") + if view.schema_events: + lines.append("Events:") + for e in view.schema_events: + lines.append(f" {e.schema_type}: {e.artifact_id} @ {e.created_at}") + lines.append(f"Limitations: {'; '.join(view.limitations)}") + lines.append(f"dry_lab_only: {view.dry_lab_only}") + return "\n".join(lines) diff --git a/tests/evidence/test_family_evidence_trail_view.py b/tests/evidence/test_family_evidence_trail_view.py new file mode 100644 index 00000000..378b7fd3 --- /dev/null +++ b/tests/evidence/test_family_evidence_trail_view.py @@ -0,0 +1,321 @@ +"""Tests for FET- family evidence trail view schema.""" + +import pytest +from openamp_foundry.evidence.family_evidence_trail_view import ( + FamilyEvidenceTrailView, + FamilySchemaEvent, + REQUIRED_TRAIL_SCHEMA_TYPES, + VALID_TRAIL_COMPLETENESS_GRADES, + build_family_evidence_trail_view, + format_family_evidence_trail_view, + validate_family_evidence_trail_view, +) + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + +_ALL_EVENTS = [ + {"schema_type": t, "artifact_id": f"ART-{t}", "created_at": "2026-07-10"} + for t in sorted(REQUIRED_TRAIL_SCHEMA_TYPES) +] + +_SOME_EVENTS = [ + {"schema_type": "BSP", "artifact_id": "ART-BSP", "created_at": "2026-07-01"}, + {"schema_type": "SEG", "artifact_id": "ART-SEG", "created_at": "2026-07-02"}, + {"schema_type": "BTI", "artifact_id": "ART-BTI", "created_at": "2026-07-03"}, +] + + +def _build(**kwargs): + defaults = dict( + fet_id="FET-001", + candidate_family_id="FAM-001", + batch_id="BATCH-01", + pipeline_version="v1.0", + schema_event_dicts=_SOME_EVENTS, + limitations=["dry-lab only"], + created_at="2026-07-10", + ) + defaults.update(kwargs) + return build_family_evidence_trail_view(**defaults) + + +# --------------------------------------------------------------------------- +# 1. Constants +# --------------------------------------------------------------------------- + + +def test_required_trail_schema_types_is_frozenset(): + assert isinstance(REQUIRED_TRAIL_SCHEMA_TYPES, frozenset) + + +def test_required_trail_schema_types_count(): + assert len(REQUIRED_TRAIL_SCHEMA_TYPES) == 9 + + +def test_required_trail_schema_types_contains_bti(): + assert "BTI" in REQUIRED_TRAIL_SCHEMA_TYPES + + +def test_required_trail_schema_types_contains_beg(): + assert "BEG" in REQUIRED_TRAIL_SCHEMA_TYPES + + +def test_required_trail_schema_types_contains_sat(): + assert "SAT" in REQUIRED_TRAIL_SCHEMA_TYPES + + +def test_required_trail_schema_types_contains_pcc(): + assert "PCC" in REQUIRED_TRAIL_SCHEMA_TYPES + + +def test_required_trail_schema_types_contains_bsp(): + assert "BSP" in REQUIRED_TRAIL_SCHEMA_TYPES + + +def test_required_trail_schema_types_contains_seg(): + assert "SEG" in REQUIRED_TRAIL_SCHEMA_TYPES + + +def test_required_trail_schema_types_contains_ecc(): + assert "ECC" in REQUIRED_TRAIL_SCHEMA_TYPES + + +def test_required_trail_schema_types_contains_rsr(): + assert "RSR" in REQUIRED_TRAIL_SCHEMA_TYPES + + +def test_required_trail_schema_types_contains_pqt(): + assert "PQT" in REQUIRED_TRAIL_SCHEMA_TYPES + + +def test_valid_grades_is_frozenset(): + assert isinstance(VALID_TRAIL_COMPLETENESS_GRADES, frozenset) + + +def test_valid_grades_contains_complete(): + assert "complete" in VALID_TRAIL_COMPLETENESS_GRADES + + +def test_valid_grades_contains_partial(): + assert "partial" in VALID_TRAIL_COMPLETENESS_GRADES + + +def test_valid_grades_contains_empty(): + assert "empty" in VALID_TRAIL_COMPLETENESS_GRADES + + +def test_valid_grades_count(): + assert len(VALID_TRAIL_COMPLETENESS_GRADES) == 3 + + +# --------------------------------------------------------------------------- +# 2. build – happy paths +# --------------------------------------------------------------------------- + + +def test_build_returns_family_evidence_trail_view(): + assert isinstance(_build(), FamilyEvidenceTrailView) + + +def test_build_fet_id_stored(): + assert _build().fet_id == "FET-001" + + +def test_build_candidate_family_id_stored(): + assert _build().candidate_family_id == "FAM-001" + + +def test_build_batch_id_stored(): + assert _build().batch_id == "BATCH-01" + + +def test_build_pipeline_version_stored(): + assert _build().pipeline_version == "v1.0" + + +def test_build_dry_lab_only_true(): + assert _build().dry_lab_only is True + + +def test_build_n_schemas_required_is_9(): + assert _build().n_schemas_required == 9 + + +def test_build_partial_grade_when_some_present(): + r = _build() + assert r.trail_completeness_grade == "partial" + + +def test_build_complete_grade_when_all_present(): + r = _build(schema_event_dicts=_ALL_EVENTS) + assert r.trail_completeness_grade == "complete" + + +def test_build_empty_grade_when_no_events(): + r = _build(schema_event_dicts=[]) + assert r.trail_completeness_grade == "empty" + + +def test_build_present_schema_types_sorted(): + r = _build() + assert r.present_schema_types == sorted(r.present_schema_types) + + +def test_build_missing_schema_types_sorted(): + r = _build() + assert r.missing_schema_types == sorted(r.missing_schema_types) + + +def test_build_present_plus_missing_equals_required(): + r = _build() + assert set(r.present_schema_types) | set(r.missing_schema_types) == REQUIRED_TRAIL_SCHEMA_TYPES + + +def test_build_n_schemas_present_matches_present_list(): + r = _build() + assert r.n_schemas_present == len(r.present_schema_types) + + +def test_build_schema_events_are_family_schema_event(): + r = _build() + for e in r.schema_events: + assert isinstance(e, FamilySchemaEvent) + + +def test_build_events_count_matches_input(): + r = _build() + assert len(r.schema_events) == len(_SOME_EVENTS) + + +def test_build_limitations_stored(): + r = _build() + assert r.limitations == ["dry-lab only"] + + +def test_build_created_at_stored(): + assert _build().created_at == "2026-07-10" + + +def test_build_all_present_no_missing(): + r = _build(schema_event_dicts=_ALL_EVENTS) + assert r.missing_schema_types == [] + + +def test_build_all_present_n_present_is_9(): + r = _build(schema_event_dicts=_ALL_EVENTS) + assert r.n_schemas_present == 9 + + +def test_build_empty_events_all_missing(): + r = _build(schema_event_dicts=[]) + assert len(r.missing_schema_types) == 9 + + +def test_build_deduplicates_duplicate_event_types(): + events = [ + {"schema_type": "BSP", "artifact_id": "A1", "created_at": "2026-07-01"}, + {"schema_type": "BSP", "artifact_id": "A2", "created_at": "2026-07-02"}, + ] + r = _build(schema_event_dicts=events) + assert r.present_schema_types == ["BSP"] + assert len(r.schema_events) == 2 + + +# --------------------------------------------------------------------------- +# 3. validate – rejection cases +# --------------------------------------------------------------------------- + + +def test_validate_rejects_bad_fet_id_prefix(): + with pytest.raises(ValueError, match="FET-"): + _build(fet_id="BAD-001") + + +def test_validate_rejects_toy_family_id(): + with pytest.raises(ValueError, match="TOY-"): + _build(candidate_family_id="TOY-001") + + +def test_validate_rejects_empty_family_id(): + with pytest.raises(ValueError): + _build(candidate_family_id="") + + +def test_validate_rejects_empty_batch_id(): + with pytest.raises(ValueError): + _build(batch_id="") + + +def test_validate_rejects_empty_pipeline_version(): + with pytest.raises(ValueError): + _build(pipeline_version="") + + +def test_validate_rejects_empty_limitations(): + with pytest.raises(ValueError, match="limitations"): + _build(limitations=[]) + + +def test_validate_rejects_unknown_schema_type_in_event(): + with pytest.raises(ValueError, match="REQUIRED_TRAIL_SCHEMA_TYPES"): + _build(schema_event_dicts=[{"schema_type": "UNKNOWN", "artifact_id": "x", "created_at": "2026-07-01"}]) + + +def test_validate_rejects_empty_artifact_id_in_event(): + with pytest.raises(ValueError, match="artifact_id"): + _build(schema_event_dicts=[{"schema_type": "BSP", "artifact_id": "", "created_at": "2026-07-01"}]) + + +def test_validate_rejects_empty_created_at(): + with pytest.raises(ValueError): + _build(created_at="") + + +# --------------------------------------------------------------------------- +# 4. format +# --------------------------------------------------------------------------- + + +def test_format_contains_fet_id(): + assert "FET-001" in format_family_evidence_trail_view(_build()) + + +def test_format_contains_family_id(): + assert "FAM-001" in format_family_evidence_trail_view(_build()) + + +def test_format_contains_batch_id(): + assert "BATCH-01" in format_family_evidence_trail_view(_build()) + + +def test_format_contains_grade(): + assert "partial" in format_family_evidence_trail_view(_build()) + + +def test_format_contains_present_count(): + r = _build() + assert f"{r.n_schemas_present}/{r.n_schemas_required}" in format_family_evidence_trail_view(r) + + +def test_format_contains_limitations(): + assert "dry-lab only" in format_family_evidence_trail_view(_build()) + + +def test_format_contains_dry_lab_only(): + assert "dry_lab_only: True" in format_family_evidence_trail_view(_build()) + + +def test_format_contains_created_at(): + assert "2026-07-10" in format_family_evidence_trail_view(_build()) + + +def test_format_complete_no_missing_line(): + r = _build(schema_event_dicts=_ALL_EVENTS) + formatted = format_family_evidence_trail_view(r) + assert "Missing:" not in formatted + + +def test_format_is_string(): + assert isinstance(format_family_evidence_trail_view(_build()), str) diff --git a/tests/test_test_count_regression.py b/tests/test_test_count_regression.py index 8e093a0b..c245347e 100644 --- a/tests/test_test_count_regression.py +++ b/tests/test_test_count_regression.py @@ -1,14 +1,15 @@ """Regression test: total test count must stay within 5% of BASELINE.""" import subprocess +import sys import math -BASELINE = 9828 +BASELINE = 9891 def test_test_count_regression(): result = subprocess.run( - ["python", "-m", "pytest", "--collect-only", "-q", "--no-header"], + [sys.executable, "-m", "pytest", "--collect-only", "-q", "--no-header"], capture_output=True, text=True, )