DataOps-platform/tests/integration/test_workflow_migration_persistence.py

import os

import pytest
from sqlalchemy import create_engine, text

from app.core.common.identifiers import new_governance_uid
from app.core.orchestration.migration.dual_run import DualRunMode
from app.core.orchestration.migration.repository import PostgresMigrationStore

pytestmark = pytest.mark.skipif(
    os.getenv("RUN_V55_PERSISTENCE_INTEGRATION") != "1",
    reason="set RUN_V55_PERSISTENCE_INTEGRATION=1",
)


def test_cutover_state_and_outbox_commit_atomically_and_idempotently():
    engine = create_engine(
        os.environ["DATAOPS_MCP_TEST_DATABASE_URL"], pool_pre_ping=True
    )
    store = PostgresMigrationStore(engine)
    dataflow_uid = new_governance_uid()
    correlation_id = new_governance_uid()
    idempotency_key = f"cutover:{dataflow_uid}:v55"
    state = None
    operation = None
    try:
        state = store.create_state(
            {
                "dataflow_uid": dataflow_uid,
                "environment": "test",
                "mode": DualRunMode.N8N_PRIMARY_KESTRA_SHADOW.value,
                "n8n_definition_id": f"n8n-{dataflow_uid}",
                "kestra_definition_id": f"kestra-{dataflow_uid}",
                "migration_batch": "low_risk",
            }
        )
        dual_run_id = store.record_dual_run(
            {
                "dataflow_uid": dataflow_uid,
                "environment": "test",
                "input_snapshot_hash": "a" * 64,
                "primary_execution_id": "n8n-v55-persistence",
                "shadow_execution_id": "kestra-v55-persistence",
                "primary_status": "success",
                "shadow_status": "success",
                "isolation": {"strategy": "read_only"},
                "formal_engine_changed": False,
                "correlation_id": correlation_id,
            }
        )
        report_id = store.record_reconciliation(
            {
                "state_id": state["state_id"],
                "dual_run_id": dual_run_id,
                "status": "passed",
                "policy": {},
                "safe_metrics": {"row_count_delta": 0},
                "differences": [],
                "correlation_id": correlation_id,
            }
        )
        assert dual_run_id
        assert report_id
        record = {
            "dataflow_uid": dataflow_uid,
            "environment": "test",
            "source_mode": DualRunMode.N8N_PRIMARY_KESTRA_SHADOW.value,
            "target_mode": DualRunMode.KESTRA_PRIMARY_N8N_STANDBY.value,
            "n8n_definition_id": state["n8n_definition_id"],
            "kestra_definition_id": state["kestra_definition_id"],
            "gates": {"reconciliation_passed": True},
            "idempotency_key": idempotency_key,
            "correlation_id": correlation_id,
        }

        claimed, operation = store.request_transition(record)
        replay_claimed, replay = store.request_transition(record)

        assert claimed is True
        assert replay_claimed is False
        assert replay["operation_id"] == operation["operation_id"]
        with engine.connect() as connection:
            event = connection.execute(
                text(
                    "SELECT event_type, payload "
                    "FROM public.outbox_events "
                    "WHERE aggregate_id = :operation_id"
                ),
                {"operation_id": operation["operation_id"]},
            ).one()
        assert event[0] == "workflow.engine.cutover.requested"
        assert event[1]["idempotency_key"] == idempotency_key
        assert "password" not in repr(event[1]).lower()

        store.complete_transition(
            operation["operation_id"],
            DualRunMode.KESTRA_PRIMARY_N8N_STANDBY.value,
            {"status": "succeeded"},
        )
        current = store.get_state(dataflow_uid, "test")
        assert current["mode"] == "kestra_primary_n8n_standby"
        assert current["status"] == "stable"
    finally:
        if state:
            store.delete_test_state(state["state_id"])
        engine.dispose()