SupplyAgent/tests/supply_infra/pipeline/test_dag_and_gates.py

from __future__ import annotations

from supply_infra.pipeline.dag import PIPELINE_STEPS
from supply_infra.pipeline.gates import evaluate_step_gate


def test_pipeline_has_twelve_strictly_ordered_steps() -> None:
    assert len(PIPELINE_STEPS) == 12
    assert PIPELINE_STEPS[0].key == "global_tree_sync"
    assert PIPELINE_STEPS[-1].key == "aigc_write_record"
    for previous, current in zip(PIPELINE_STEPS, PIPELINE_STEPS[1:]):
        assert current.dependencies == (previous.key,)
        assert current.critical is True


def test_aigc_gate_requires_durable_effect_record() -> None:
    failed = evaluate_step_gate("aigc_write_record", {"success": True})
    assert failed.passed is False
    live_ok = evaluate_step_gate(
        "aigc_write_record",
        {
            "success": True,
            "effect_recorded": True,
            "payload_hash": "abc",
            "external_request_made": True,
        },
    )
    assert live_ok.passed is True


def test_aigc_gate_rejects_failed_batches() -> None:
    decision = evaluate_step_gate(
        "aigc_write_record",
        {
            "success": True,
            "effect_recorded": True,
            "payload_hash": "abc",
            "failed_batch_count": 2,
        },
    )
    assert decision.passed is False
    assert decision.error_code == "aigc_publish_failed"


def test_explicit_step_failure_fails_closed() -> None:
    decision = evaluate_step_gate(
        "global_tree_sync",
        {"success": False, "error": "source missing"},
    )
    assert decision.passed is False
    assert decision.error_code == "step_reported_failure"