diff --git a/src/openamp_foundry/evidence/pipeline_phase_timeline_report.py b/src/openamp_foundry/evidence/pipeline_phase_timeline_report.py new file mode 100644 index 00000000..62f3fcb5 --- /dev/null +++ b/src/openamp_foundry/evidence/pipeline_phase_timeline_report.py @@ -0,0 +1,187 @@ +"""PTR- pipeline phase timeline report schema. + +Ordered sequence of all schema events for a pipeline run, sorted by +created_at; shows progression from nomination to shortlisting with schema +type, artifact ID, and timestamp per event. +""" + +from __future__ import annotations + +from dataclasses import dataclass + +VALID_PIPELINE_SCHEMA_TYPES: frozenset[str] = frozenset({ + "BSP", "PSC", "BOS", "CPS", "CBA", "CRG", + "RSR", "PQT", "BTI", "CBA2", "BEG", "SAT", "PCC", + "FET", "BRC", "CSV", +}) + +VALID_TIMELINE_VERDICTS: frozenset[str] = frozenset({ + "complete", "in_progress", "empty", +}) + +NOMINATION_TYPES: frozenset[str] = frozenset({"BSP", "PSC"}) +SCREENING_TYPES: frozenset[str] = frozenset({"BOS", "CPS", "CBA", "CRG"}) +RANKING_TYPES: frozenset[str] = frozenset({"RSR", "PQT"}) +EVIDENCE_TYPES: frozenset[str] = frozenset({"BTI", "CBA2", "BEG", "SAT", "PCC"}) +INTEGRATION_TYPES: frozenset[str] = frozenset({"FET", "BRC", "CSV"}) + +TERMINAL_SCHEMA_TYPES: frozenset[str] = frozenset({"SAT", "PCC"}) + + +def _phase_for_schema_type(schema_type: str) -> str: + if schema_type in NOMINATION_TYPES: + return "nomination" + if schema_type in SCREENING_TYPES: + return "screening" + if schema_type in RANKING_TYPES: + return "ranking" + if schema_type in EVIDENCE_TYPES: + return "evidence" + if schema_type in INTEGRATION_TYPES: + return "integration" + return "unknown" + + +@dataclass +class TimelineEvent: + schema_type: str + artifact_id: str + phase: str + created_at: str + + +@dataclass +class PipelinePhaseTimelineReport: + ptr_id: str + pipeline_run_id: str + pipeline_version: str + batch_id: str + events: list[TimelineEvent] + n_events: int + schema_types_present: list[str] + terminal_schemas_present: list[str] + timeline_verdict: str + dry_lab_only: bool + limitations: list[str] + created_at: str + + +def _compute_verdict(events: list[TimelineEvent]) -> str: + if not events: + return "empty" + present = {e.schema_type for e in events} + if TERMINAL_SCHEMA_TYPES.issubset(present): + return "complete" + return "in_progress" + + +def validate_pipeline_phase_timeline_report(ptr: PipelinePhaseTimelineReport) -> None: + if not ptr.ptr_id.startswith("PTR-"): + raise ValueError(f"ptr_id must start with 'PTR-': {ptr.ptr_id!r}") + if not ptr.pipeline_run_id: + raise ValueError("pipeline_run_id must be non-empty") + if not ptr.pipeline_version: + raise ValueError("pipeline_version must be non-empty") + if not ptr.batch_id: + raise ValueError("batch_id must be non-empty") + for event in ptr.events: + if event.schema_type not in VALID_PIPELINE_SCHEMA_TYPES: + raise ValueError( + f"schema_type {event.schema_type!r} not in VALID_PIPELINE_SCHEMA_TYPES" + ) + if not event.artifact_id: + raise ValueError( + f"artifact_id must be non-empty for schema_type {event.schema_type!r}" + ) + if ptr.timeline_verdict not in VALID_TIMELINE_VERDICTS: + raise ValueError( + f"timeline_verdict {ptr.timeline_verdict!r} not in VALID_TIMELINE_VERDICTS" + ) + if not ptr.dry_lab_only: + raise ValueError("dry_lab_only must be True") + if not ptr.limitations: + raise ValueError("limitations must be non-empty") + if not ptr.created_at: + raise ValueError("created_at must be non-empty") + if ptr.n_events != len(ptr.events): + raise ValueError("n_events must equal len(events)") + expected_types = sorted({e.schema_type for e in ptr.events}) + if ptr.schema_types_present != expected_types: + raise ValueError("schema_types_present mismatch") + expected_terminal = sorted( + TERMINAL_SCHEMA_TYPES & {e.schema_type for e in ptr.events} + ) + if ptr.terminal_schemas_present != expected_terminal: + raise ValueError("terminal_schemas_present mismatch") + + +def build_pipeline_phase_timeline_report( + *, + ptr_id: str, + pipeline_run_id: str, + pipeline_version: str, + batch_id: str, + event_dicts: list[dict], + limitations: list[str], + created_at: str, +) -> PipelinePhaseTimelineReport: + """Build a pipeline phase timeline report. + + event_dicts: list of dicts with keys: schema_type, artifact_id, created_at. + Events are sorted by created_at, then schema_type. + """ + events_raw = [ + TimelineEvent( + schema_type=d["schema_type"], + artifact_id=d["artifact_id"], + phase=_phase_for_schema_type(d["schema_type"]), + created_at=d["created_at"], + ) + for d in event_dicts + ] + events_sorted = sorted(events_raw, key=lambda e: (e.created_at, e.schema_type)) + schema_types_present = sorted({e.schema_type for e in events_sorted}) + terminal_present = sorted( + TERMINAL_SCHEMA_TYPES & {e.schema_type for e in events_sorted} + ) + verdict = _compute_verdict(events_sorted) + + ptr = PipelinePhaseTimelineReport( + ptr_id=ptr_id, + pipeline_run_id=pipeline_run_id, + pipeline_version=pipeline_version, + batch_id=batch_id, + events=events_sorted, + n_events=len(events_sorted), + schema_types_present=schema_types_present, + terminal_schemas_present=terminal_present, + timeline_verdict=verdict, + dry_lab_only=True, + limitations=limitations, + created_at=created_at, + ) + validate_pipeline_phase_timeline_report(ptr) + return ptr + + +def format_pipeline_phase_timeline_report(ptr: PipelinePhaseTimelineReport) -> str: + lines = [ + f"Pipeline Phase Timeline Report — {ptr.ptr_id}", + f"Run: {ptr.pipeline_run_id} | Pipeline: {ptr.pipeline_version}", + f"Batch: {ptr.batch_id} | Verdict: {ptr.timeline_verdict}", + f"Events: {ptr.n_events} | Schema types: {len(ptr.schema_types_present)}", + ] + if ptr.schema_types_present: + lines.append(f"Present: {', '.join(ptr.schema_types_present)}") + if ptr.terminal_schemas_present: + lines.append(f"Terminal schemas: {', '.join(ptr.terminal_schemas_present)}") + if ptr.events: + lines.append("Timeline:") + for event in ptr.events: + lines.append( + f" [{event.created_at}] {event.schema_type} ({event.phase}): {event.artifact_id}" + ) + lines.append(f"Created: {ptr.created_at}") + lines.append(f"Limitations: {'; '.join(ptr.limitations)}") + lines.append(f"dry_lab_only: {ptr.dry_lab_only}") + return "\n".join(lines) diff --git a/tests/evidence/test_pipeline_phase_timeline_report.py b/tests/evidence/test_pipeline_phase_timeline_report.py new file mode 100644 index 00000000..5af1145c --- /dev/null +++ b/tests/evidence/test_pipeline_phase_timeline_report.py @@ -0,0 +1,294 @@ +"""Tests for PTR- pipeline phase timeline report schema.""" + +import pytest +from openamp_foundry.evidence.pipeline_phase_timeline_report import ( + PipelinePhaseTimelineReport, + TimelineEvent, + VALID_PIPELINE_SCHEMA_TYPES, + VALID_TIMELINE_VERDICTS, + TERMINAL_SCHEMA_TYPES, + build_pipeline_phase_timeline_report, + format_pipeline_phase_timeline_report, + validate_pipeline_phase_timeline_report, +) + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + +_EVENTS_COMPLETE = [ + {"schema_type": "BSP", "artifact_id": "A1", "created_at": "2026-07-01"}, + {"schema_type": "SAT", "artifact_id": "A2", "created_at": "2026-07-05"}, + {"schema_type": "PCC", "artifact_id": "A3", "created_at": "2026-07-10"}, +] + +_EVENTS_PARTIAL = [ + {"schema_type": "BSP", "artifact_id": "A1", "created_at": "2026-07-01"}, + {"schema_type": "RSR", "artifact_id": "A2", "created_at": "2026-07-03"}, +] + + +def _build(**kwargs): + defaults = dict( + ptr_id="PTR-001", + pipeline_run_id="RUN-001", + pipeline_version="v1.0", + batch_id="BATCH-01", + event_dicts=_EVENTS_PARTIAL, + limitations=["dry-lab only"], + created_at="2026-07-10", + ) + defaults.update(kwargs) + return build_pipeline_phase_timeline_report(**defaults) + + +# --------------------------------------------------------------------------- +# 1. Constants +# --------------------------------------------------------------------------- + + +def test_valid_pipeline_schema_types_is_frozenset(): + assert isinstance(VALID_PIPELINE_SCHEMA_TYPES, frozenset) + + +def test_valid_pipeline_schema_types_contains_bsp(): + assert "BSP" in VALID_PIPELINE_SCHEMA_TYPES + + +def test_valid_pipeline_schema_types_contains_sat(): + assert "SAT" in VALID_PIPELINE_SCHEMA_TYPES + + +def test_valid_pipeline_schema_types_contains_pcc(): + assert "PCC" in VALID_PIPELINE_SCHEMA_TYPES + + +def test_valid_pipeline_schema_types_contains_fet(): + assert "FET" in VALID_PIPELINE_SCHEMA_TYPES + + +def test_valid_pipeline_schema_types_count(): + assert len(VALID_PIPELINE_SCHEMA_TYPES) == 16 + + +def test_valid_timeline_verdicts_is_frozenset(): + assert isinstance(VALID_TIMELINE_VERDICTS, frozenset) + + +def test_valid_timeline_verdicts_contains_complete(): + assert "complete" in VALID_TIMELINE_VERDICTS + + +def test_valid_timeline_verdicts_contains_in_progress(): + assert "in_progress" in VALID_TIMELINE_VERDICTS + + +def test_valid_timeline_verdicts_contains_empty(): + assert "empty" in VALID_TIMELINE_VERDICTS + + +def test_terminal_schema_types_contains_sat(): + assert "SAT" in TERMINAL_SCHEMA_TYPES + + +def test_terminal_schema_types_contains_pcc(): + assert "PCC" in TERMINAL_SCHEMA_TYPES + + +# --------------------------------------------------------------------------- +# 2. build – happy paths +# --------------------------------------------------------------------------- + + +def test_build_returns_pipeline_phase_timeline_report(): + assert isinstance(_build(), PipelinePhaseTimelineReport) + + +def test_build_ptr_id_stored(): + assert _build().ptr_id == "PTR-001" + + +def test_build_pipeline_run_id_stored(): + assert _build().pipeline_run_id == "RUN-001" + + +def test_build_pipeline_version_stored(): + assert _build().pipeline_version == "v1.0" + + +def test_build_batch_id_stored(): + assert _build().batch_id == "BATCH-01" + + +def test_build_dry_lab_only_true(): + assert _build().dry_lab_only is True + + +def test_build_complete_verdict_when_both_terminal_present(): + r = _build(event_dicts=_EVENTS_COMPLETE) + assert r.timeline_verdict == "complete" + + +def test_build_in_progress_when_partial(): + r = _build() + assert r.timeline_verdict == "in_progress" + + +def test_build_empty_verdict_when_no_events(): + r = _build(event_dicts=[]) + assert r.timeline_verdict == "empty" + + +def test_build_n_events_matches_input(): + r = _build() + assert r.n_events == len(_EVENTS_PARTIAL) + + +def test_build_events_are_timeline_event(): + r = _build() + for e in r.events: + assert isinstance(e, TimelineEvent) + + +def test_build_events_sorted_by_created_at(): + events = [ + {"schema_type": "SAT", "artifact_id": "A2", "created_at": "2026-07-05"}, + {"schema_type": "BSP", "artifact_id": "A1", "created_at": "2026-07-01"}, + ] + r = _build(event_dicts=events) + assert r.events[0].created_at <= r.events[1].created_at + + +def test_build_schema_types_present_sorted(): + r = _build() + assert r.schema_types_present == sorted(r.schema_types_present) + + +def test_build_terminal_schemas_present_empty_when_partial(): + r = _build() + assert r.terminal_schemas_present == [] + + +def test_build_terminal_schemas_present_when_complete(): + r = _build(event_dicts=_EVENTS_COMPLETE) + assert set(r.terminal_schemas_present) == TERMINAL_SCHEMA_TYPES + + +def test_build_event_phase_nomination_for_bsp(): + events = [{"schema_type": "BSP", "artifact_id": "A1", "created_at": "2026-07-01"}] + r = _build(event_dicts=events) + assert r.events[0].phase == "nomination" + + +def test_build_event_phase_evidence_for_sat(): + events = [{"schema_type": "SAT", "artifact_id": "A1", "created_at": "2026-07-01"}] + r = _build(event_dicts=events) + assert r.events[0].phase == "evidence" + + +def test_build_event_phase_ranking_for_rsr(): + events = [{"schema_type": "RSR", "artifact_id": "A1", "created_at": "2026-07-01"}] + r = _build(event_dicts=events) + assert r.events[0].phase == "ranking" + + +def test_build_event_phase_integration_for_brc(): + events = [{"schema_type": "BRC", "artifact_id": "A1", "created_at": "2026-07-01"}] + r = _build(event_dicts=events) + assert r.events[0].phase == "integration" + + +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_ptr_id_prefix(): + with pytest.raises(ValueError, match="PTR-"): + _build(ptr_id="BAD-001") + + +def test_validate_rejects_empty_pipeline_run_id(): + with pytest.raises(ValueError): + _build(pipeline_run_id="") + + +def test_validate_rejects_empty_pipeline_version(): + with pytest.raises(ValueError): + _build(pipeline_version="") + + +def test_validate_rejects_empty_batch_id(): + with pytest.raises(ValueError): + _build(batch_id="") + + +def test_validate_rejects_unknown_schema_type(): + events = [{"schema_type": "UNKNOWN", "artifact_id": "A1", "created_at": "2026-07-01"}] + with pytest.raises(ValueError, match="VALID_PIPELINE_SCHEMA_TYPES"): + _build(event_dicts=events) + + +def test_validate_rejects_empty_artifact_id(): + events = [{"schema_type": "BSP", "artifact_id": "", "created_at": "2026-07-01"}] + with pytest.raises(ValueError, match="artifact_id"): + _build(event_dicts=events) + + +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_ptr_id(): + assert "PTR-001" in format_pipeline_phase_timeline_report(_build()) + + +def test_format_contains_pipeline_run_id(): + assert "RUN-001" in format_pipeline_phase_timeline_report(_build()) + + +def test_format_contains_verdict(): + assert "in_progress" in format_pipeline_phase_timeline_report(_build()) + + +def test_format_contains_batch_id(): + assert "BATCH-01" in format_pipeline_phase_timeline_report(_build()) + + +def test_format_contains_event_schema_type(): + assert "BSP" in format_pipeline_phase_timeline_report(_build()) + + +def test_format_contains_phase(): + assert "nomination" in format_pipeline_phase_timeline_report(_build()) + + +def test_format_contains_limitations(): + assert "dry-lab only" in format_pipeline_phase_timeline_report(_build()) + + +def test_format_contains_dry_lab_only(): + assert "dry_lab_only: True" in format_pipeline_phase_timeline_report(_build()) + + +def test_format_is_string(): + assert isinstance(format_pipeline_phase_timeline_report(_build()), str) diff --git a/tests/test_test_count_regression.py b/tests/test_test_count_regression.py index 525ce3a4..d159cb82 100644 --- a/tests/test_test_count_regression.py +++ b/tests/test_test_count_regression.py @@ -4,7 +4,7 @@ import sys import math -BASELINE = 10017 +BASELINE = 10080 def test_test_count_regression():