DataOps-platform/tests/integration/test_n8n_retirement_audit.py

from app.core.orchestration.migration.retirement import (
    N8nRetirementChecklist,
    N8nRetirementGate,
)


def checklist(**overrides):
    values = {
        "inventory_disposition_complete": True,
        "all_dataflows_kestra_primary": True,
        "business_cycle_and_recovery_complete": True,
        "standby_observation_no_rollback": True,
        "monitoring_and_alerting_complete": True,
        "dependency_failures_safe": True,
        "secure_archive_complete": True,
        "independent_review_approved": True,
        "final_l4_passed": True,
    }
    values.update(overrides)
    return N8nRetirementChecklist(**values)


def test_retirement_is_blocked_with_explicit_missing_evidence():
    decision = N8nRetirementGate.evaluate(
        checklist(
            inventory_disposition_complete=False,
            independent_review_approved=False,
            final_l4_passed=False,
        )
    )

    assert decision["status"] == "blocked"
    assert decision["retirement_approved"] is False
    assert decision["missing"] == [
        "inventory_disposition_complete",
        "independent_review_approved",
        "final_l4_passed",
    ]


def test_retirement_approval_requires_every_gate():
    decision = N8nRetirementGate.evaluate(checklist())

    assert decision == {
        "status": "approved",
        "retirement_approved": True,
        "missing": [],
    }