diff --git a/src/openamp_foundry/evidence/cross_schema_linkage_validator.py b/src/openamp_foundry/evidence/cross_schema_linkage_validator.py new file mode 100644 index 00000000..1d39f8f2 --- /dev/null +++ b/src/openamp_foundry/evidence/cross_schema_linkage_validator.py @@ -0,0 +1,171 @@ +"""CSV- cross-schema linkage validator schema. + +Validates that artifact IDs referenced in SAT evidence_artifact_ids, BEG +missing lists, and BTI entries are internally consistent. Catches orphaned +references before external sharing. +""" + +from __future__ import annotations + +from dataclasses import dataclass + +VALID_LINKAGE_VERDICTS: frozenset[str] = frozenset({ + "all_valid", "orphans_found", "no_references", +}) + +VALID_REFERENCE_SOURCES: frozenset[str] = frozenset({ + "SAT", "BEG", "BTI", +}) + + +@dataclass +class OrphanEntry: + artifact_id: str + reference_source: str + details: str + + +@dataclass +class CrossSchemaLinkageValidator: + csv_id: str + batch_id: str + pipeline_version: str + sat_id: str + beg_id: str + bti_id: str + all_known_artifact_ids: list[str] + referenced_artifact_ids: list[str] + orphan_entries: list[OrphanEntry] + n_references_checked: int + n_orphans_found: int + linkage_verdict: str + dry_lab_only: bool + limitations: list[str] + created_at: str + + +def _compute_verdict(n_refs: int, n_orphans: int) -> str: + if n_refs == 0: + return "no_references" + if n_orphans > 0: + return "orphans_found" + return "all_valid" + + +def validate_cross_schema_linkage_validator(csv: CrossSchemaLinkageValidator) -> None: + if not csv.csv_id.startswith("CSV-"): + raise ValueError(f"csv_id must start with 'CSV-': {csv.csv_id!r}") + if not csv.batch_id: + raise ValueError("batch_id must be non-empty") + if not csv.pipeline_version: + raise ValueError("pipeline_version must be non-empty") + if not csv.sat_id.startswith("SAT-"): + raise ValueError(f"sat_id must start with 'SAT-': {csv.sat_id!r}") + if not csv.beg_id.startswith("BEG-"): + raise ValueError(f"beg_id must start with 'BEG-': {csv.beg_id!r}") + if not csv.bti_id.startswith("BTI-"): + raise ValueError(f"bti_id must start with 'BTI-': {csv.bti_id!r}") + if csv.linkage_verdict not in VALID_LINKAGE_VERDICTS: + raise ValueError( + f"linkage_verdict {csv.linkage_verdict!r} not in VALID_LINKAGE_VERDICTS" + ) + for entry in csv.orphan_entries: + if entry.reference_source not in VALID_REFERENCE_SOURCES: + raise ValueError( + f"reference_source {entry.reference_source!r} not in VALID_REFERENCE_SOURCES" + ) + if not csv.dry_lab_only: + raise ValueError("dry_lab_only must be True") + if not csv.limitations: + raise ValueError("limitations must be non-empty") + if not csv.created_at: + raise ValueError("created_at must be non-empty") + if csv.n_references_checked != len(csv.referenced_artifact_ids): + raise ValueError("n_references_checked must equal len(referenced_artifact_ids)") + if csv.n_orphans_found != len(csv.orphan_entries): + raise ValueError("n_orphans_found must equal len(orphan_entries)") + + +def build_cross_schema_linkage_validator( + *, + csv_id: str, + batch_id: str, + pipeline_version: str, + sat_id: str, + beg_id: str, + bti_id: str, + known_artifact_ids: list[str], + reference_dicts: list[dict], + limitations: list[str], + created_at: str, +) -> CrossSchemaLinkageValidator: + """Build a cross-schema linkage validator record. + + Args: + known_artifact_ids: All artifact IDs that actually exist in this batch. + reference_dicts: List of dicts with keys: artifact_id, reference_source. + reference_source must be one of VALID_REFERENCE_SOURCES. + """ + for d in reference_dicts: + if d["reference_source"] not in VALID_REFERENCE_SOURCES: + raise ValueError( + f"reference_source {d['reference_source']!r} not in VALID_REFERENCE_SOURCES" + ) + + known_set = set(known_artifact_ids) + all_known_sorted = sorted(known_artifact_ids) + + referenced = sorted({d["artifact_id"] for d in reference_dicts}) + orphans: list[OrphanEntry] = [] + for d in reference_dicts: + aid = d["artifact_id"] + src = d["reference_source"] + if aid not in known_set: + orphans.append(OrphanEntry( + artifact_id=aid, + reference_source=src, + details=f"{aid!r} referenced in {src} but not found in known artifacts", + )) + + orphans_sorted = sorted(orphans, key=lambda o: (o.artifact_id, o.reference_source)) + verdict = _compute_verdict(len(referenced), len(orphans_sorted)) + + record = CrossSchemaLinkageValidator( + csv_id=csv_id, + batch_id=batch_id, + pipeline_version=pipeline_version, + sat_id=sat_id, + beg_id=beg_id, + bti_id=bti_id, + all_known_artifact_ids=all_known_sorted, + referenced_artifact_ids=referenced, + orphan_entries=orphans_sorted, + n_references_checked=len(referenced), + n_orphans_found=len(orphans_sorted), + linkage_verdict=verdict, + dry_lab_only=True, + limitations=limitations, + created_at=created_at, + ) + validate_cross_schema_linkage_validator(record) + return record + + +def format_cross_schema_linkage_validator(csv: CrossSchemaLinkageValidator) -> str: + lines = [ + f"Cross-Schema Linkage Validator — {csv.csv_id}", + f"Batch: {csv.batch_id} | Pipeline: {csv.pipeline_version}", + f"Verdict: {csv.linkage_verdict}", + f"References checked: {csv.n_references_checked} | Orphans: {csv.n_orphans_found}", + f"SAT: {csv.sat_id} | BEG: {csv.beg_id} | BTI: {csv.bti_id}", + ] + if csv.orphan_entries: + lines.append("Orphaned references:") + for entry in csv.orphan_entries: + lines.append(f" [{entry.reference_source}] {entry.artifact_id}: {entry.details}") + else: + lines.append("No orphaned references found.") + lines.append(f"Created: {csv.created_at}") + lines.append(f"Limitations: {'; '.join(csv.limitations)}") + lines.append(f"dry_lab_only: {csv.dry_lab_only}") + return "\n".join(lines) diff --git a/tests/evidence/test_cross_schema_linkage_validator.py b/tests/evidence/test_cross_schema_linkage_validator.py new file mode 100644 index 00000000..cc2fcf26 --- /dev/null +++ b/tests/evidence/test_cross_schema_linkage_validator.py @@ -0,0 +1,301 @@ +"""Tests for CSV- cross-schema linkage validator schema.""" + +import pytest +from openamp_foundry.evidence.cross_schema_linkage_validator import ( + CrossSchemaLinkageValidator, + OrphanEntry, + VALID_LINKAGE_VERDICTS, + VALID_REFERENCE_SOURCES, + build_cross_schema_linkage_validator, + format_cross_schema_linkage_validator, + validate_cross_schema_linkage_validator, +) + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + +_KNOWN = ["ART-001", "ART-002", "ART-003"] +_REFS_CLEAN = [ + {"artifact_id": "ART-001", "reference_source": "SAT"}, + {"artifact_id": "ART-002", "reference_source": "BEG"}, +] +_REFS_WITH_ORPHAN = [ + {"artifact_id": "ART-001", "reference_source": "SAT"}, + {"artifact_id": "ART-999", "reference_source": "BTI"}, +] + + +def _build(**kwargs): + defaults = dict( + csv_id="CSV-001", + batch_id="BATCH-01", + pipeline_version="v1.0", + sat_id="SAT-001", + beg_id="BEG-001", + bti_id="BTI-001", + known_artifact_ids=_KNOWN, + reference_dicts=_REFS_CLEAN, + limitations=["dry-lab only"], + created_at="2026-07-10", + ) + defaults.update(kwargs) + return build_cross_schema_linkage_validator(**defaults) + + +# --------------------------------------------------------------------------- +# 1. Constants +# --------------------------------------------------------------------------- + + +def test_valid_linkage_verdicts_is_frozenset(): + assert isinstance(VALID_LINKAGE_VERDICTS, frozenset) + + +def test_valid_linkage_verdicts_contains_all_valid(): + assert "all_valid" in VALID_LINKAGE_VERDICTS + + +def test_valid_linkage_verdicts_contains_orphans_found(): + assert "orphans_found" in VALID_LINKAGE_VERDICTS + + +def test_valid_linkage_verdicts_contains_no_references(): + assert "no_references" in VALID_LINKAGE_VERDICTS + + +def test_valid_reference_sources_is_frozenset(): + assert isinstance(VALID_REFERENCE_SOURCES, frozenset) + + +def test_valid_reference_sources_contains_sat(): + assert "SAT" in VALID_REFERENCE_SOURCES + + +def test_valid_reference_sources_contains_beg(): + assert "BEG" in VALID_REFERENCE_SOURCES + + +def test_valid_reference_sources_contains_bti(): + assert "BTI" in VALID_REFERENCE_SOURCES + + +def test_valid_reference_sources_count(): + assert len(VALID_REFERENCE_SOURCES) == 3 + + +# --------------------------------------------------------------------------- +# 2. build – happy paths +# --------------------------------------------------------------------------- + + +def test_build_returns_cross_schema_linkage_validator(): + assert isinstance(_build(), CrossSchemaLinkageValidator) + + +def test_build_csv_id_stored(): + assert _build().csv_id == "CSV-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_all_valid_verdict_when_no_orphans(): + r = _build() + assert r.linkage_verdict == "all_valid" + + +def test_build_orphans_found_verdict_when_unknown_ref(): + r = _build(reference_dicts=_REFS_WITH_ORPHAN) + assert r.linkage_verdict == "orphans_found" + + +def test_build_no_references_verdict_when_empty_refs(): + r = _build(reference_dicts=[]) + assert r.linkage_verdict == "no_references" + + +def test_build_n_orphans_zero_when_clean(): + assert _build().n_orphans_found == 0 + + +def test_build_n_orphans_one_when_unknown_ref(): + r = _build(reference_dicts=_REFS_WITH_ORPHAN) + assert r.n_orphans_found == 1 + + +def test_build_n_references_checked_matches_unique_refs(): + r = _build() + assert r.n_references_checked == len(set(d["artifact_id"] for d in _REFS_CLEAN)) + + +def test_build_referenced_artifact_ids_sorted(): + r = _build() + assert r.referenced_artifact_ids == sorted(r.referenced_artifact_ids) + + +def test_build_all_known_artifact_ids_sorted(): + r = _build() + assert r.all_known_artifact_ids == sorted(r.all_known_artifact_ids) + + +def test_build_orphan_entries_are_orphan_entry(): + r = _build(reference_dicts=_REFS_WITH_ORPHAN) + for entry in r.orphan_entries: + assert isinstance(entry, OrphanEntry) + + +def test_build_orphan_artifact_id_correct(): + r = _build(reference_dicts=_REFS_WITH_ORPHAN) + assert r.orphan_entries[0].artifact_id == "ART-999" + + +def test_build_orphan_reference_source_correct(): + r = _build(reference_dicts=_REFS_WITH_ORPHAN) + assert r.orphan_entries[0].reference_source == "BTI" + + +def test_build_no_orphan_when_all_known(): + refs = [{"artifact_id": aid, "reference_source": "SAT"} for aid in _KNOWN] + r = _build(reference_dicts=refs) + assert r.orphan_entries == [] + + +def test_build_deduplicates_referenced_ids(): + refs = [ + {"artifact_id": "ART-001", "reference_source": "SAT"}, + {"artifact_id": "ART-001", "reference_source": "BEG"}, + ] + r = _build(reference_dicts=refs) + assert r.n_references_checked == 1 + + +def test_build_limitations_stored(): + assert _build().limitations == ["dry-lab only"] + + +def test_build_created_at_stored(): + assert _build().created_at == "2026-07-10" + + +def test_build_sat_id_stored(): + assert _build().sat_id == "SAT-001" + + +def test_build_beg_id_stored(): + assert _build().beg_id == "BEG-001" + + +def test_build_bti_id_stored(): + assert _build().bti_id == "BTI-001" + + +# --------------------------------------------------------------------------- +# 3. validate – rejection cases +# --------------------------------------------------------------------------- + + +def test_validate_rejects_bad_csv_id_prefix(): + with pytest.raises(ValueError, match="CSV-"): + _build(csv_id="BAD-001") + + +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_bad_sat_id_prefix(): + with pytest.raises(ValueError, match="SAT-"): + _build(sat_id="BAD-001") + + +def test_validate_rejects_bad_beg_id_prefix(): + with pytest.raises(ValueError, match="BEG-"): + _build(beg_id="BAD-001") + + +def test_validate_rejects_bad_bti_id_prefix(): + with pytest.raises(ValueError, match="BTI-"): + _build(bti_id="BAD-001") + + +def test_validate_rejects_invalid_reference_source(): + refs = [{"artifact_id": "ART-001", "reference_source": "UNKNOWN"}] + with pytest.raises(ValueError, match="VALID_REFERENCE_SOURCES"): + _build(reference_dicts=refs) + + +def test_validate_rejects_empty_limitations(): + with pytest.raises(ValueError, match="limitations"): + _build(limitations=[]) + + +def test_validate_rejects_empty_created_at(): + with pytest.raises(ValueError): + _build(created_at="") + + +# --------------------------------------------------------------------------- +# 4. format +# --------------------------------------------------------------------------- + + +def test_format_contains_csv_id(): + assert "CSV-001" in format_cross_schema_linkage_validator(_build()) + + +def test_format_contains_batch_id(): + assert "BATCH-01" in format_cross_schema_linkage_validator(_build()) + + +def test_format_contains_verdict(): + assert "all_valid" in format_cross_schema_linkage_validator(_build()) + + +def test_format_contains_sat_id(): + assert "SAT-001" in format_cross_schema_linkage_validator(_build()) + + +def test_format_contains_beg_id(): + assert "BEG-001" in format_cross_schema_linkage_validator(_build()) + + +def test_format_contains_bti_id(): + assert "BTI-001" in format_cross_schema_linkage_validator(_build()) + + +def test_format_contains_orphan_id_when_present(): + r = _build(reference_dicts=_REFS_WITH_ORPHAN) + assert "ART-999" in format_cross_schema_linkage_validator(r) + + +def test_format_no_orphan_message_when_clean(): + formatted = format_cross_schema_linkage_validator(_build()) + assert "No orphaned references found." in formatted + + +def test_format_contains_limitations(): + assert "dry-lab only" in format_cross_schema_linkage_validator(_build()) + + +def test_format_contains_dry_lab_only(): + assert "dry_lab_only: True" in format_cross_schema_linkage_validator(_build()) + + +def test_format_is_string(): + assert isinstance(format_cross_schema_linkage_validator(_build()), str) diff --git a/tests/test_test_count_regression.py b/tests/test_test_count_regression.py index 51a43eb8..525ce3a4 100644 --- a/tests/test_test_count_regression.py +++ b/tests/test_test_count_regression.py @@ -4,7 +4,7 @@ import sys import math -BASELINE = 9954 +BASELINE = 10017 def test_test_count_regression():