DataOps-platform/tests/integration/test_workflow_migration_rollout.py

from datetime import UTC, datetime, timedelta

import pytest

from app.core.orchestration.migration.rollout import (
    MigrationBatch,
    ObservationWindow,
    RolloutEntryGates,
    RolloutExitGates,
    RolloutPolicy,
)


def test_batches_are_fixed_from_low_to_high_risk():
    assert [batch.value for batch in MigrationBatch] == [
        "read_only_low_frequency",
        "idempotent_partition_write",
        "core_multi_node",
        "special_high_risk",
    ]


def test_entry_gates_require_shadow_failure_recovery_and_capacity():
    gates = RolloutEntryGates(
        inventory_complete=True,
        workflow_spec_valid=True,
        policy_complete=True,
        rollback_target_present=True,
        shadow_runs=2,
        required_shadow_runs=3,
        failure_recovery_observed=True,
        reconciliation_passed=True,
        connection_budget_passed=True,
        peak_concurrency_passed=True,
        sla_passed=True,
    )

    with pytest.raises(ValueError, match="shadow_runs"):
        RolloutPolicy.assert_entry(gates)


def test_observation_window_fails_closed_before_deadline_or_on_incident():
    now = datetime(2026, 7, 19, tzinfo=UTC)
    window = ObservationWindow(
        started_at=now - timedelta(hours=2),
        required_duration=timedelta(hours=4),
        unexplained_differences=0,
        duplicate_writes=0,
        critical_sla_breaches=0,
        rollback_requests=0,
    )
    with pytest.raises(ValueError, match="observation_window_complete"):
        RolloutPolicy.assert_observation(window, now=now)

    incident = ObservationWindow(
        started_at=now - timedelta(hours=5),
        required_duration=timedelta(hours=4),
        unexplained_differences=0,
        duplicate_writes=1,
        critical_sla_breaches=0,
        rollback_requests=0,
    )
    with pytest.raises(ValueError, match="duplicate_writes"):
        RolloutPolicy.assert_observation(incident, now=now)


def test_exit_gates_require_recovery_and_unchanged_n8n_standby():
    gates = RolloutExitGates(
        observation_window_complete=True,
        no_unexplained_differences=True,
        no_duplicate_writes=True,
        sla_passed=True,
        pause_retry_rollback_exercised=False,
        n8n_standby=True,
        n8n_unauthorized_changes=0,
    )

    with pytest.raises(ValueError, match="pause_retry_rollback_exercised"):
        RolloutPolicy.assert_exit(gates)