From c3051a16fda5f2cfbb23c0a65e809691a8954a67 Mon Sep 17 00:00:00 2001 From: OpenCode Date: Fri, 10 Jul 2026 12:07:15 +0700 Subject: [PATCH] feat: Phase I I5 cross-repo evidence traceability -- CRT- record links artifacts across partner labs and independent replications, 63 tests (#918) --- .../evidence/cross_repo_traceability.py | 214 +++++++++++ .../evidence/test_cross_repo_traceability.py | 362 ++++++++++++++++++ 2 files changed, 576 insertions(+) create mode 100644 src/openamp_foundry/evidence/cross_repo_traceability.py create mode 100644 tests/evidence/test_cross_repo_traceability.py diff --git a/src/openamp_foundry/evidence/cross_repo_traceability.py b/src/openamp_foundry/evidence/cross_repo_traceability.py new file mode 100644 index 00000000..1cb2f62e --- /dev/null +++ b/src/openamp_foundry/evidence/cross_repo_traceability.py @@ -0,0 +1,214 @@ +"""Cross-repo evidence traceability schema for multi-lab reproducibility.""" + +from __future__ import annotations + +from dataclasses import dataclass, field + +CROSS_REPO_TRACEABILITY_ID_PREFIX = "CRT-" + +VALID_REPO_TYPES: frozenset = frozenset( + { + "primary", + "fork", + "mirror", + "partner_lab", + "independent_replication", + "downstream_consumer", + } +) + +VALID_ARTIFACT_TYPES: frozenset = frozenset( + { + "evidence_certificate", + "release_manifest", + "benchmark_card", + "negative_result_record", + "batch_outcome_summary", + "fasta_export", + "calibration_cycle_summary", + "external_review_packet", + } +) + +VALID_TRACEABILITY_STATUSES: frozenset = frozenset( + {"pending", "linked", "verified", "broken", "superseded"} +) + +VALID_LINK_CONFIDENCE_LEVELS: frozenset = frozenset( + {"high", "medium", "low", "unverified"} +) + +CANONICAL_REPO_URL = "https://github.com/Open-Problem-Lab/OpenAMP-Foundry" + + +@dataclass +class CrossRepoArtifactRef: + artifact_id: str + artifact_type: str + source_repo_url: str + source_repo_type: str + commit_sha: str + confidence_level: str + is_verified: bool = False + verification_note: str = "" + + +@dataclass +class CrossRepoTraceabilityRecord: + record_id: str + primary_artifact_id: str + primary_artifact_type: str + external_refs: list = field(default_factory=list) + total_external_repos: int = 0 + traceability_status: str = "pending" + dry_lab_only: bool = True + is_example_data: bool = True + traceability_note: str = "" + + +@dataclass +class TraceabilityValidationResult: + is_valid: bool + violations: list = field(default_factory=list) + + +def build_cross_repo_artifact_ref( + artifact_id: str, + artifact_type: str, + source_repo_url: str, + source_repo_type: str, + commit_sha: str, + confidence_level: str, + is_verified: bool = False, + verification_note: str = "", +) -> CrossRepoArtifactRef: + return CrossRepoArtifactRef( + artifact_id=artifact_id, + artifact_type=artifact_type, + source_repo_url=source_repo_url, + source_repo_type=source_repo_type, + commit_sha=commit_sha, + confidence_level=confidence_level, + is_verified=is_verified, + verification_note=verification_note, + ) + + +def build_cross_repo_traceability_record( + record_id: str, + primary_artifact_id: str, + primary_artifact_type: str, + external_refs: list, + traceability_status: str, + dry_lab_only: bool = True, + is_example_data: bool = True, + traceability_note: str = "", +) -> CrossRepoTraceabilityRecord: + return CrossRepoTraceabilityRecord( + record_id=record_id, + primary_artifact_id=primary_artifact_id, + primary_artifact_type=primary_artifact_type, + external_refs=list(external_refs), + total_external_repos=len(external_refs), + traceability_status=traceability_status, + dry_lab_only=dry_lab_only, + is_example_data=is_example_data, + traceability_note=traceability_note or f"CRT record for {primary_artifact_id}", + ) + + +def validate_cross_repo_traceability_record( + record: CrossRepoTraceabilityRecord, +) -> TraceabilityValidationResult: + violations: list[str] = [] + + if not record.record_id.startswith(CROSS_REPO_TRACEABILITY_ID_PREFIX): + violations.append( + f"record_id must start with '{CROSS_REPO_TRACEABILITY_ID_PREFIX}', " + f"got '{record.record_id}'" + ) + + if not record.primary_artifact_id: + violations.append("primary_artifact_id must not be empty") + + if record.primary_artifact_type not in VALID_ARTIFACT_TYPES: + violations.append( + f"primary_artifact_type '{record.primary_artifact_type}' " + f"not in VALID_ARTIFACT_TYPES" + ) + + if record.traceability_status not in VALID_TRACEABILITY_STATUSES: + violations.append( + f"traceability_status '{record.traceability_status}' " + f"not in VALID_TRACEABILITY_STATUSES" + ) + + if record.total_external_repos != len(record.external_refs): + violations.append( + f"total_external_repos={record.total_external_repos} does not match " + f"len(external_refs)={len(record.external_refs)}" + ) + + if not record.dry_lab_only: + violations.append( + "dry_lab_only must be True; wet-lab traceability records require human review" + ) + + if not record.traceability_note: + violations.append("traceability_note must not be empty") + + for i, ref in enumerate(record.external_refs): + pfx = f"external_refs[{i}]" + + if not ref.artifact_id: + violations.append(f"{pfx}: artifact_id must not be empty") + + if ref.artifact_type not in VALID_ARTIFACT_TYPES: + violations.append( + f"{pfx}: artifact_type '{ref.artifact_type}' not in VALID_ARTIFACT_TYPES" + ) + + if not ref.source_repo_url.startswith("https://"): + violations.append( + f"{pfx}: source_repo_url must start with 'https://', " + f"got '{ref.source_repo_url}'" + ) + + if ref.source_repo_type not in VALID_REPO_TYPES: + violations.append( + f"{pfx}: source_repo_type '{ref.source_repo_type}' not in VALID_REPO_TYPES" + ) + + if not ref.commit_sha: + violations.append(f"{pfx}: commit_sha must not be empty") + + if ref.confidence_level not in VALID_LINK_CONFIDENCE_LEVELS: + violations.append( + f"{pfx}: confidence_level '{ref.confidence_level}' " + f"not in VALID_LINK_CONFIDENCE_LEVELS" + ) + + return TraceabilityValidationResult( + is_valid=len(violations) == 0, + violations=violations, + ) + + +def format_cross_repo_traceability_record(record: CrossRepoTraceabilityRecord) -> str: + lines: list[str] = [] + lines.append(f"Cross-Repo Traceability Record ({record.record_id})") + lines.append(f" Primary artifact: {record.primary_artifact_id} [{record.primary_artifact_type}]") + lines.append(f" Status: {record.traceability_status}") + lines.append(f" Dry-lab only: {record.dry_lab_only}") + lines.append(f" Total external repos: {record.total_external_repos}") + lines.append(f" Note: {record.traceability_note}") + if record.external_refs: + lines.append(" External references:") + for ref in record.external_refs: + verified = "✓" if ref.is_verified else "?" + lines.append( + f" [{verified}] {ref.artifact_id} ({ref.artifact_type}) " + f"@ {ref.source_repo_url} " + f"[{ref.source_repo_type}] confidence={ref.confidence_level}" + ) + return "\n".join(lines) + "\n" diff --git a/tests/evidence/test_cross_repo_traceability.py b/tests/evidence/test_cross_repo_traceability.py new file mode 100644 index 00000000..31790451 --- /dev/null +++ b/tests/evidence/test_cross_repo_traceability.py @@ -0,0 +1,362 @@ +"""Tests for src/openamp_foundry/evidence/cross_repo_traceability.py (63 tests).""" + +import pytest +from openamp_foundry.evidence.cross_repo_traceability import ( + CROSS_REPO_TRACEABILITY_ID_PREFIX, + VALID_REPO_TYPES, + VALID_ARTIFACT_TYPES, + VALID_TRACEABILITY_STATUSES, + VALID_LINK_CONFIDENCE_LEVELS, + CANONICAL_REPO_URL, + CrossRepoArtifactRef, + CrossRepoTraceabilityRecord, + TraceabilityValidationResult, + build_cross_repo_artifact_ref, + build_cross_repo_traceability_record, + validate_cross_repo_traceability_record, + format_cross_repo_traceability_record, +) + + +def _valid_ref(): + return build_cross_repo_artifact_ref( + artifact_id="CERT-TOY-001", + artifact_type="evidence_certificate", + source_repo_url="https://github.com/partner-lab/openamp-fork", + source_repo_type="partner_lab", + commit_sha="abc1234", + confidence_level="medium", + is_verified=False, + ) + + +def _valid_record(refs=None): + if refs is None: + refs = [_valid_ref()] + return build_cross_repo_traceability_record( + record_id="CRT-001", + primary_artifact_id="CERT-TOY-001", + primary_artifact_type="evidence_certificate", + external_refs=refs, + traceability_status="linked", + dry_lab_only=True, + is_example_data=True, + traceability_note="CRT record for toy example", + ) + + +class TestConstants: + def test_prefix(self): + assert CROSS_REPO_TRACEABILITY_ID_PREFIX == "CRT-" + + def test_valid_repo_types_is_frozenset(self): + assert isinstance(VALID_REPO_TYPES, frozenset) + + def test_primary_in_repo_types(self): + assert "primary" in VALID_REPO_TYPES + + def test_partner_lab_in_repo_types(self): + assert "partner_lab" in VALID_REPO_TYPES + + def test_independent_replication_in_repo_types(self): + assert "independent_replication" in VALID_REPO_TYPES + + def test_valid_artifact_types_is_frozenset(self): + assert isinstance(VALID_ARTIFACT_TYPES, frozenset) + + def test_evidence_certificate_in_artifact_types(self): + assert "evidence_certificate" in VALID_ARTIFACT_TYPES + + def test_release_manifest_in_artifact_types(self): + assert "release_manifest" in VALID_ARTIFACT_TYPES + + def test_fasta_export_in_artifact_types(self): + assert "fasta_export" in VALID_ARTIFACT_TYPES + + def test_valid_traceability_statuses_is_frozenset(self): + assert isinstance(VALID_TRACEABILITY_STATUSES, frozenset) + + def test_pending_in_statuses(self): + assert "pending" in VALID_TRACEABILITY_STATUSES + + def test_verified_in_statuses(self): + assert "verified" in VALID_TRACEABILITY_STATUSES + + def test_broken_in_statuses(self): + assert "broken" in VALID_TRACEABILITY_STATUSES + + def test_valid_confidence_levels_is_frozenset(self): + assert isinstance(VALID_LINK_CONFIDENCE_LEVELS, frozenset) + + def test_high_in_confidence_levels(self): + assert "high" in VALID_LINK_CONFIDENCE_LEVELS + + def test_unverified_in_confidence_levels(self): + assert "unverified" in VALID_LINK_CONFIDENCE_LEVELS + + def test_canonical_repo_url_is_github(self): + assert "github.com" in CANONICAL_REPO_URL + + +class TestBuildCrossRepoArtifactRef: + def test_returns_artifact_ref(self): + ref = _valid_ref() + assert isinstance(ref, CrossRepoArtifactRef) + + def test_artifact_id_stored(self): + ref = _valid_ref() + assert ref.artifact_id == "CERT-TOY-001" + + def test_artifact_type_stored(self): + ref = _valid_ref() + assert ref.artifact_type == "evidence_certificate" + + def test_source_repo_url_stored(self): + ref = _valid_ref() + assert ref.source_repo_url == "https://github.com/partner-lab/openamp-fork" + + def test_source_repo_type_stored(self): + ref = _valid_ref() + assert ref.source_repo_type == "partner_lab" + + def test_commit_sha_stored(self): + ref = _valid_ref() + assert ref.commit_sha == "abc1234" + + def test_confidence_level_stored(self): + ref = _valid_ref() + assert ref.confidence_level == "medium" + + def test_is_verified_default_false(self): + ref = _valid_ref() + assert ref.is_verified is False + + def test_verification_note_default_empty(self): + ref = _valid_ref() + assert ref.verification_note == "" + + +class TestBuildCrossRepoTraceabilityRecord: + def test_returns_traceability_record(self): + record = _valid_record() + assert isinstance(record, CrossRepoTraceabilityRecord) + + def test_record_id_stored(self): + record = _valid_record() + assert record.record_id == "CRT-001" + + def test_primary_artifact_id_stored(self): + record = _valid_record() + assert record.primary_artifact_id == "CERT-TOY-001" + + def test_primary_artifact_type_stored(self): + record = _valid_record() + assert record.primary_artifact_type == "evidence_certificate" + + def test_total_external_repos_computed(self): + record = _valid_record([_valid_ref(), _valid_ref()]) + assert record.total_external_repos == 2 + + def test_traceability_status_stored(self): + record = _valid_record() + assert record.traceability_status == "linked" + + def test_dry_lab_only_stored(self): + record = _valid_record() + assert record.dry_lab_only is True + + def test_is_example_data_stored(self): + record = _valid_record() + assert record.is_example_data is True + + def test_traceability_note_auto_when_empty(self): + record = build_cross_repo_traceability_record( + record_id="CRT-001", + primary_artifact_id="CERT-001", + primary_artifact_type="evidence_certificate", + external_refs=[_valid_ref()], + traceability_status="pending", + ) + assert record.traceability_note != "" + + +class TestValidateCrossRepoTraceabilityRecord: + def test_valid_record_passes(self): + result = validate_cross_repo_traceability_record(_valid_record()) + assert result.is_valid + + def test_returns_validation_result(self): + result = validate_cross_repo_traceability_record(_valid_record()) + assert isinstance(result, TraceabilityValidationResult) + + def test_no_violations_on_valid(self): + result = validate_cross_repo_traceability_record(_valid_record()) + assert result.violations == [] + + def test_wrong_prefix_fails(self): + record = _valid_record() + record.record_id = "XYZ-001" + result = validate_cross_repo_traceability_record(record) + assert not result.is_valid + assert any("CRT-" in v for v in result.violations) + + def test_empty_primary_artifact_id_fails(self): + record = _valid_record() + record.primary_artifact_id = "" + result = validate_cross_repo_traceability_record(record) + assert not result.is_valid + + def test_invalid_primary_artifact_type_fails(self): + record = _valid_record() + record.primary_artifact_type = "unknown_type" + result = validate_cross_repo_traceability_record(record) + assert not result.is_valid + assert any("primary_artifact_type" in v for v in result.violations) + + def test_invalid_traceability_status_fails(self): + record = _valid_record() + record.traceability_status = "bogus" + result = validate_cross_repo_traceability_record(record) + assert not result.is_valid + + def test_total_repos_mismatch_fails(self): + record = _valid_record() + record.total_external_repos = 99 + result = validate_cross_repo_traceability_record(record) + assert not result.is_valid + assert any("total_external_repos" in v for v in result.violations) + + def test_dry_lab_false_fails(self): + record = _valid_record() + record.dry_lab_only = False + result = validate_cross_repo_traceability_record(record) + assert not result.is_valid + assert any("dry_lab_only" in v for v in result.violations) + + def test_empty_traceability_note_fails(self): + record = _valid_record() + record.traceability_note = "" + result = validate_cross_repo_traceability_record(record) + assert not result.is_valid + assert any("traceability_note" in v for v in result.violations) + + def test_ref_empty_artifact_id_fails(self): + ref = _valid_ref() + ref.artifact_id = "" + result = validate_cross_repo_traceability_record(_valid_record([ref])) + assert not result.is_valid + + def test_ref_invalid_artifact_type_fails(self): + ref = _valid_ref() + ref.artifact_type = "unknown" + result = validate_cross_repo_traceability_record(_valid_record([ref])) + assert not result.is_valid + + def test_ref_non_https_url_fails(self): + ref = _valid_ref() + ref.source_repo_url = "http://insecure.example.com/repo" + result = validate_cross_repo_traceability_record(_valid_record([ref])) + assert not result.is_valid + assert any("https://" in v for v in result.violations) + + def test_ref_invalid_repo_type_fails(self): + ref = _valid_ref() + ref.source_repo_type = "unknown_type" + result = validate_cross_repo_traceability_record(_valid_record([ref])) + assert not result.is_valid + + def test_ref_empty_commit_sha_fails(self): + ref = _valid_ref() + ref.commit_sha = "" + result = validate_cross_repo_traceability_record(_valid_record([ref])) + assert not result.is_valid + assert any("commit_sha" in v for v in result.violations) + + def test_ref_invalid_confidence_level_fails(self): + ref = _valid_ref() + ref.confidence_level = "very_high" + result = validate_cross_repo_traceability_record(_valid_record([ref])) + assert not result.is_valid + + def test_multiple_violations_accumulated(self): + record = _valid_record() + record.record_id = "BAD-001" + record.primary_artifact_type = "unknown" + result = validate_cross_repo_traceability_record(record) + assert len(result.violations) >= 2 + + def test_all_valid_artifact_types_accepted(self): + for atype in VALID_ARTIFACT_TYPES: + record = build_cross_repo_traceability_record( + record_id="CRT-001", + primary_artifact_id="CERT-001", + primary_artifact_type=atype, + external_refs=[_valid_ref()], + traceability_status="pending", + traceability_note="test", + ) + result = validate_cross_repo_traceability_record(record) + assert result.is_valid, f"Type {atype} failed: {result.violations}" + + def test_all_valid_confidence_levels_accepted(self): + for level in VALID_LINK_CONFIDENCE_LEVELS: + ref = build_cross_repo_artifact_ref( + artifact_id="CERT-001", + artifact_type="evidence_certificate", + source_repo_url="https://github.com/partner/repo", + source_repo_type="partner_lab", + commit_sha="abc123", + confidence_level=level, + ) + record = _valid_record([ref]) + result = validate_cross_repo_traceability_record(record) + assert result.is_valid, f"Level {level} failed: {result.violations}" + + +class TestFormatCrossRepoTraceabilityRecord: + def test_returns_string(self): + record = _valid_record() + result = format_cross_repo_traceability_record(record) + assert isinstance(result, str) + + def test_contains_record_id(self): + record = _valid_record() + result = format_cross_repo_traceability_record(record) + assert "CRT-001" in result + + def test_contains_primary_artifact_id(self): + record = _valid_record() + result = format_cross_repo_traceability_record(record) + assert "CERT-TOY-001" in result + + def test_contains_status(self): + record = _valid_record() + result = format_cross_repo_traceability_record(record) + assert "linked" in result + + def test_contains_ref_artifact_id(self): + record = _valid_record() + result = format_cross_repo_traceability_record(record) + assert "CERT-TOY-001" in result + + def test_contains_repo_url(self): + record = _valid_record() + result = format_cross_repo_traceability_record(record) + assert "github.com" in result + + def test_ends_with_newline(self): + record = _valid_record() + result = format_cross_repo_traceability_record(record) + assert result.endswith("\n") + + def test_verified_ref_shows_checkmark(self): + ref = _valid_ref() + ref.is_verified = True + record = _valid_record([ref]) + result = format_cross_repo_traceability_record(record) + assert "✓" in result + + def test_unverified_ref_shows_question_mark(self): + record = _valid_record() + result = format_cross_repo_traceability_record(record) + assert "?" in result