diff --git a/src/openamp_foundry/evidence/pre_release_audit_gateway.py b/src/openamp_foundry/evidence/pre_release_audit_gateway.py new file mode 100644 index 00000000..cbfb8c23 --- /dev/null +++ b/src/openamp_foundry/evidence/pre_release_audit_gateway.py @@ -0,0 +1,162 @@ +"""PRG- pre-release audit gateway schema. + +Single-gate record asserting BRC (batch release checklist), PCC (pipeline +completeness), and ECI (evidence completeness index) all pass before external +sharing. Blocks release until all three gates are green. +""" + +from __future__ import annotations + +from dataclasses import dataclass + +VALID_GATE_STATUSES: frozenset[str] = frozenset({"pass", "fail", "not_evaluated"}) +VALID_RELEASE_VERDICTS: frozenset[str] = frozenset({"approved", "blocked"}) + +REQUIRED_GATES: tuple[str, ...] = ("BRC", "PCC", "ECI") + + +@dataclass +class GateResult: + gate_type: str + artifact_id: str + status: str + details: str + + +@dataclass +class PreReleaseAuditGateway: + prg_id: str + batch_id: str + pipeline_version: str + gate_results: list[GateResult] + n_gates_total: int + n_gates_passed: int + n_gates_failed: int + release_verdict: str + dry_lab_only: bool + limitations: list[str] + created_at: str + + +def validate_pre_release_audit_gateway(prg: PreReleaseAuditGateway) -> None: + if not prg.prg_id.startswith("PRG-"): + raise ValueError(f"prg_id must start with 'PRG-': {prg.prg_id!r}") + if not prg.batch_id: + raise ValueError("batch_id must be non-empty") + if not prg.pipeline_version: + raise ValueError("pipeline_version must be non-empty") + for gr in prg.gate_results: + if gr.gate_type not in REQUIRED_GATES: + raise ValueError( + f"gate_type {gr.gate_type!r} must be one of {REQUIRED_GATES}" + ) + if gr.status not in VALID_GATE_STATUSES: + raise ValueError( + f"gate status {gr.status!r} not in VALID_GATE_STATUSES" + ) + gate_types = [gr.gate_type for gr in prg.gate_results] + for req in REQUIRED_GATES: + if gate_types.count(req) != 1: + raise ValueError(f"Exactly one gate result required for {req!r}") + if prg.release_verdict not in VALID_RELEASE_VERDICTS: + raise ValueError( + f"release_verdict {prg.release_verdict!r} not in VALID_RELEASE_VERDICTS" + ) + if not prg.dry_lab_only: + raise ValueError("dry_lab_only must be True") + if not prg.limitations: + raise ValueError("limitations must be non-empty") + if not prg.created_at: + raise ValueError("created_at must be non-empty") + if prg.n_gates_total != len(prg.gate_results): + raise ValueError("n_gates_total must equal len(gate_results)") + n_passed = sum(1 for gr in prg.gate_results if gr.status == "pass") + if prg.n_gates_passed != n_passed: + raise ValueError("n_gates_passed mismatch") + if prg.n_gates_failed != sum( + 1 for gr in prg.gate_results if gr.status == "fail" + ): + raise ValueError("n_gates_failed mismatch") + + +def build_pre_release_audit_gateway( + *, + prg_id: str, + batch_id: str, + pipeline_version: str, + brc_artifact_id: str, + brc_verdict: str, + pcc_artifact_id: str, + pcc_grade: str, + eci_artifact_id: str, + eci_grade: str, + limitations: list[str], + created_at: str, +) -> PreReleaseAuditGateway: + """Build a pre-release audit gateway. + + brc_verdict: e.g. "approved", "blocked", "conditional" + pcc_grade: "A"/"B"/"C"/"D" — pass if A or B + eci_grade: "A"/"B"/"C"/"D" — pass if A or B + """ + brc_pass = brc_verdict == "approved" + pcc_pass = pcc_grade in ("A", "B") + eci_pass = eci_grade in ("A", "B") + + gate_results = [ + GateResult( + gate_type="BRC", + artifact_id=brc_artifact_id, + status="pass" if brc_pass else "fail", + details=f"BRC verdict={brc_verdict!r} — {'pass' if brc_pass else 'fail'}", + ), + GateResult( + gate_type="PCC", + artifact_id=pcc_artifact_id, + status="pass" if pcc_pass else "fail", + details=f"PCC grade={pcc_grade!r} — {'pass' if pcc_pass else 'fail (need A or B)'}", + ), + GateResult( + gate_type="ECI", + artifact_id=eci_artifact_id, + status="pass" if eci_pass else "fail", + details=f"ECI grade={eci_grade!r} — {'pass' if eci_pass else 'fail (need A or B)'}", + ), + ] + n_total = len(gate_results) + n_passed = sum(1 for gr in gate_results if gr.status == "pass") + n_failed = sum(1 for gr in gate_results if gr.status == "fail") + verdict = "approved" if n_failed == 0 else "blocked" + + prg = PreReleaseAuditGateway( + prg_id=prg_id, + batch_id=batch_id, + pipeline_version=pipeline_version, + gate_results=gate_results, + n_gates_total=n_total, + n_gates_passed=n_passed, + n_gates_failed=n_failed, + release_verdict=verdict, + dry_lab_only=True, + limitations=limitations, + created_at=created_at, + ) + validate_pre_release_audit_gateway(prg) + return prg + + +def format_pre_release_audit_gateway(prg: PreReleaseAuditGateway) -> str: + lines = [ + f"Pre-Release Audit Gateway — {prg.prg_id}", + f"Batch: {prg.batch_id} | Pipeline: {prg.pipeline_version}", + f"Verdict: {prg.release_verdict} | Gates: {prg.n_gates_total}", + f"Passed: {prg.n_gates_passed} | Failed: {prg.n_gates_failed}", + "Gate results:", + ] + for gr in prg.gate_results: + status_label = "PASS" if gr.status == "pass" else "FAIL" + lines.append(f" [{status_label}] {gr.gate_type} ({gr.artifact_id}): {gr.details}") + lines.append(f"Created: {prg.created_at}") + lines.append(f"Limitations: {'; '.join(prg.limitations)}") + lines.append(f"dry_lab_only: {prg.dry_lab_only}") + return "\n".join(lines) diff --git a/tests/evidence/test_pre_release_audit_gateway.py b/tests/evidence/test_pre_release_audit_gateway.py new file mode 100644 index 00000000..404e24a3 --- /dev/null +++ b/tests/evidence/test_pre_release_audit_gateway.py @@ -0,0 +1,300 @@ +"""Tests for PRG- pre-release audit gateway schema.""" + +import pytest +from openamp_foundry.evidence.pre_release_audit_gateway import ( + PreReleaseAuditGateway, + GateResult, + VALID_GATE_STATUSES, + VALID_RELEASE_VERDICTS, + REQUIRED_GATES, + build_pre_release_audit_gateway, + format_pre_release_audit_gateway, + validate_pre_release_audit_gateway, +) + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + + +def _build(**kwargs): + defaults = dict( + prg_id="PRG-001", + batch_id="BATCH-01", + pipeline_version="v1.0", + brc_artifact_id="BRC-001", + brc_verdict="approved", + pcc_artifact_id="PCC-001", + pcc_grade="A", + eci_artifact_id="ECI-001", + eci_grade="B", + limitations=["dry-lab only"], + created_at="2026-07-10", + ) + defaults.update(kwargs) + return build_pre_release_audit_gateway(**defaults) + + +# --------------------------------------------------------------------------- +# 1. Constants +# --------------------------------------------------------------------------- + + +def test_valid_gate_statuses_is_frozenset(): + assert isinstance(VALID_GATE_STATUSES, frozenset) + + +def test_valid_gate_statuses_contains_pass(): + assert "pass" in VALID_GATE_STATUSES + + +def test_valid_gate_statuses_contains_fail(): + assert "fail" in VALID_GATE_STATUSES + + +def test_valid_gate_statuses_contains_not_evaluated(): + assert "not_evaluated" in VALID_GATE_STATUSES + + +def test_valid_release_verdicts_is_frozenset(): + assert isinstance(VALID_RELEASE_VERDICTS, frozenset) + + +def test_valid_release_verdicts_contains_approved(): + assert "approved" in VALID_RELEASE_VERDICTS + + +def test_valid_release_verdicts_contains_blocked(): + assert "blocked" in VALID_RELEASE_VERDICTS + + +def test_required_gates_is_tuple(): + assert isinstance(REQUIRED_GATES, tuple) + + +def test_required_gates_contains_brc(): + assert "BRC" in REQUIRED_GATES + + +def test_required_gates_contains_pcc(): + assert "PCC" in REQUIRED_GATES + + +def test_required_gates_contains_eci(): + assert "ECI" in REQUIRED_GATES + + +def test_required_gates_count(): + assert len(REQUIRED_GATES) == 3 + + +# --------------------------------------------------------------------------- +# 2. build – happy paths +# --------------------------------------------------------------------------- + + +def test_build_returns_pre_release_audit_gateway(): + assert isinstance(_build(), PreReleaseAuditGateway) + + +def test_build_prg_id_stored(): + assert _build().prg_id == "PRG-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_approved_when_all_pass(): + r = _build() + assert r.release_verdict == "approved" + + +def test_build_blocked_when_brc_not_approved(): + r = _build(brc_verdict="blocked") + assert r.release_verdict == "blocked" + + +def test_build_blocked_when_pcc_grade_c(): + r = _build(pcc_grade="C") + assert r.release_verdict == "blocked" + + +def test_build_blocked_when_eci_grade_d(): + r = _build(eci_grade="D") + assert r.release_verdict == "blocked" + + +def test_build_approved_with_pcc_grade_b(): + r = _build(pcc_grade="B") + assert r.release_verdict == "approved" + + +def test_build_approved_with_eci_grade_a(): + r = _build(eci_grade="A") + assert r.release_verdict == "approved" + + +def test_build_n_gates_total_is_3(): + assert _build().n_gates_total == 3 + + +def test_build_n_gates_passed_3_when_all_pass(): + r = _build() + assert r.n_gates_passed == 3 + + +def test_build_n_gates_failed_0_when_all_pass(): + r = _build() + assert r.n_gates_failed == 0 + + +def test_build_n_gates_failed_1_on_bad_pcc(): + r = _build(pcc_grade="C") + assert r.n_gates_failed == 1 + + +def test_build_gate_results_are_gate_result(): + for gr in _build().gate_results: + assert isinstance(gr, GateResult) + + +def test_build_gate_results_contain_brc(): + types = [gr.gate_type for gr in _build().gate_results] + assert "BRC" in types + + +def test_build_gate_results_contain_pcc(): + types = [gr.gate_type for gr in _build().gate_results] + assert "PCC" in types + + +def test_build_gate_results_contain_eci(): + types = [gr.gate_type for gr in _build().gate_results] + assert "ECI" in types + + +def test_build_brc_artifact_id_stored(): + r = _build() + brc_gate = next(gr for gr in r.gate_results if gr.gate_type == "BRC") + assert brc_gate.artifact_id == "BRC-001" + + +def test_build_pcc_artifact_id_stored(): + r = _build() + pcc_gate = next(gr for gr in r.gate_results if gr.gate_type == "PCC") + assert pcc_gate.artifact_id == "PCC-001" + + +def test_build_eci_artifact_id_stored(): + r = _build() + eci_gate = next(gr for gr in r.gate_results if gr.gate_type == "ECI") + assert eci_gate.artifact_id == "ECI-001" + + +def test_build_brc_status_pass_when_approved(): + r = _build() + brc_gate = next(gr for gr in r.gate_results if gr.gate_type == "BRC") + assert brc_gate.status == "pass" + + +def test_build_brc_status_fail_when_blocked(): + r = _build(brc_verdict="blocked") + brc_gate = next(gr for gr in r.gate_results if gr.gate_type == "BRC") + assert brc_gate.status == "fail" + + +def test_build_pcc_status_fail_when_grade_d(): + r = _build(pcc_grade="D") + pcc_gate = next(gr for gr in r.gate_results if gr.gate_type == "PCC") + assert pcc_gate.status == "fail" + + +def test_build_limitations_stored(): + assert _build().limitations == ["dry-lab only"] + + +def test_build_created_at_stored(): + assert _build().created_at == "2026-07-10" + + +# --------------------------------------------------------------------------- +# 3. validate – rejection cases +# --------------------------------------------------------------------------- + + +def test_validate_rejects_bad_prg_id_prefix(): + with pytest.raises(ValueError, match="PRG-"): + _build(prg_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_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_prg_id(): + assert "PRG-001" in format_pre_release_audit_gateway(_build()) + + +def test_format_contains_batch_id(): + assert "BATCH-01" in format_pre_release_audit_gateway(_build()) + + +def test_format_contains_verdict(): + assert "approved" in format_pre_release_audit_gateway(_build()) + + +def test_format_contains_pass_markers(): + assert "[PASS]" in format_pre_release_audit_gateway(_build()) + + +def test_format_contains_fail_markers_when_blocked(): + r = _build(pcc_grade="D") + assert "[FAIL]" in format_pre_release_audit_gateway(r) + + +def test_format_contains_gate_types(): + formatted = format_pre_release_audit_gateway(_build()) + for gt in REQUIRED_GATES: + assert gt in formatted + + +def test_format_contains_limitations(): + assert "dry-lab only" in format_pre_release_audit_gateway(_build()) + + +def test_format_contains_dry_lab_only(): + assert "dry_lab_only: True" in format_pre_release_audit_gateway(_build()) + + +def test_format_is_string(): + assert isinstance(format_pre_release_audit_gateway(_build()), str) diff --git a/tests/test_test_count_regression.py b/tests/test_test_count_regression.py index 24310a5b..33fb2e69 100644 --- a/tests/test_test_count_regression.py +++ b/tests/test_test_count_regression.py @@ -4,7 +4,7 @@ import sys import math -BASELINE = 10143 +BASELINE = 10206 def test_test_count_regression():