DataOps-platform/tests/integration/test_v55_real_cutover_rollback_l3.py

"""Opt-in V55 PostgreSQL + n8n + Kestra cutover and rollback drill."""

import os
import uuid

import pytest
import requests
from sqlalchemy import create_engine

from app.core.common.identifiers import new_governance_uid
from app.core.data_factory.n8n_client import N8nClient
from app.core.orchestration.engines.kestra import KestraAdapter
from app.core.orchestration.engines.n8n import N8nAdapter
from app.core.orchestration.migration.cutover import (
    CutoverGates,
    EngineRoleController,
    WorkflowCutoverService,
)
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_CUTOVER_L3") != "1",
    reason="set RUN_V55_CUTOVER_L3=1 for the local cutover drill",
)

DATABASE_URL = os.getenv("DATAOPS_MCP_TEST_DATABASE_URL", "")
KESTRA_URL = os.getenv(
    "KESTRA_BASE_URL", "http://127.0.0.1:18080/api/v1"
)
KESTRA_AUTH = (
    os.getenv("KESTRA_USERNAME", ""),
    os.getenv("KESTRA_PASSWORD", ""),
)
KESTRA_NAMESPACE = "dataops.v55"


def passing_gates():
    return CutoverGates(
        inventory_complete=True,
        workflow_spec_valid=True,
        rollback_target_present=True,
        shadow_runs=3,
        required_shadow_runs=3,
        failure_recovery_observed=True,
        reconciliation_passed=True,
        connection_budget_passed=True,
        sla_passed=True,
        n8n_baseline_unchanged=True,
    )


def test_real_n8n_and_kestra_cutover_failure_restore_and_explicit_rollback():
    engine = create_engine(DATABASE_URL, pool_pre_ping=True)
    store = PostgresMigrationStore(engine)
    n8n_client = N8nClient(
        api_url="http://127.0.0.1:15678",
        api_key=os.environ["N8N_API_KEY"],
        timeout=20,
    )
    n8n = N8nAdapter(n8n_client)
    kestra = KestraAdapter(
        base_url=KESTRA_URL,
        username=KESTRA_AUTH[0],
        password=KESTRA_AUTH[1],
    )
    controller = EngineRoleController(
        n8n=n8n,
        kestra=kestra,
        kestra_namespace=KESTRA_NAMESPACE,
    )
    service = WorkflowCutoverService(store=store, engines=controller)
    suffix = uuid.uuid4().hex[:10]
    flow_id = f"v55_cutover_{suffix}"
    flow_url = f"{KESTRA_URL}/main/flows/{KESTRA_NAMESPACE}/{flow_id}"
    workflow_id = None
    state_ids = []

    n8n_definition = {
        "name": f"V55 rollback drill {suffix}",
        "nodes": [
            {
                "parameters": {
                    "rule": {
                        "interval": [
                            {"field": "hours", "hoursInterval": 24}
                        ]
                    }
                },
                "id": f"schedule-{suffix}",
                "name": "Schedule Trigger",
                "type": "n8n-nodes-base.scheduleTrigger",
                "typeVersion": 1.2,
                "position": [0, 0],
            }
        ],
        "connections": {},
        "settings": {"executionOrder": "v1"},
    }
    kestra_definition = f"""
id: {flow_id}
namespace: {KESTRA_NAMESPACE}
description: V55 controlled cutover and rollback drill
disabled: true
tasks:
  - id: evidence
    type: io.kestra.plugin.core.log.Log
    message: V55 cutover evidence
""".strip()

    try:
        workflow = n8n.deploy_disabled(n8n_definition)
        workflow_id = str(workflow["id"])
        n8n.activate("dataops", workflow_id)
        kestra.deploy_disabled(kestra_definition)

        failure_state = store.create_state(
            {
                "dataflow_uid": new_governance_uid(),
                "environment": "test",
                "mode": DualRunMode.N8N_PRIMARY_KESTRA_SHADOW.value,
                "n8n_definition_id": workflow_id,
                "kestra_definition_id": f"missing_{suffix}",
                "migration_batch": "read_only_low_frequency",
            }
        )
        state_ids.append(failure_state["state_id"])
        failed_operation = service.request_cutover(
            dataflow_uid=failure_state["dataflow_uid"],
            environment="test",
            target_mode=DualRunMode.KESTRA_PRIMARY_N8N_STANDBY,
            gates=passing_gates(),
            idempotency_key=f"v55-failure-{suffix}",
            correlation_id=new_governance_uid(),
        )
        failure_result = service.apply_requested_cutover(failed_operation)

        assert failure_result["status"] == "rolled_back"
        assert n8n_client.get_workflow(workflow_id)["active"] is True
        store.delete_test_state(failure_state["state_id"])
        state_ids.remove(failure_state["state_id"])

        success_state = store.create_state(
            {
                "dataflow_uid": new_governance_uid(),
                "environment": "test",
                "mode": DualRunMode.N8N_PRIMARY_KESTRA_SHADOW.value,
                "n8n_definition_id": workflow_id,
                "kestra_definition_id": flow_id,
                "migration_batch": "read_only_low_frequency",
            }
        )
        state_ids.append(success_state["state_id"])
        operation = service.request_cutover(
            dataflow_uid=success_state["dataflow_uid"],
            environment="test",
            target_mode=DualRunMode.KESTRA_PRIMARY_N8N_STANDBY,
            gates=passing_gates(),
            idempotency_key=f"v55-success-{suffix}",
            correlation_id=new_governance_uid(),
        )
        cutover_result = service.apply_requested_cutover(operation)

        assert cutover_result["status"] == "succeeded"
        assert n8n_client.get_workflow(workflow_id)["active"] is False
        response = requests.get(
            flow_url, auth=KESTRA_AUTH, timeout=20
        )
        response.raise_for_status()
        assert response.json()["disabled"] is False

        rollback_result = service.rollback_to_n8n(
            dataflow_uid=success_state["dataflow_uid"],
            environment="test",
            idempotency_key=f"v55-rollback-{suffix}",
            correlation_id=new_governance_uid(),
        )

        assert rollback_result["status"] == "succeeded"
        assert n8n_client.get_workflow(workflow_id)["active"] is True
        response = requests.get(
            flow_url, auth=KESTRA_AUTH, timeout=20
        )
        response.raise_for_status()
        assert response.json()["disabled"] is True
    finally:
        for state_id in state_ids:
            store.delete_test_state(state_id)
        if workflow_id is not None:
            try:
                n8n.deactivate("dataops", workflow_id)
            finally:
                n8n_client.delete_workflow(workflow_id)
        requests.delete(flow_url, auth=KESTRA_AUTH, timeout=20)
        engine.dispose()