DataOps-platform/tests/integration/test_n8n_kestra_dual_run.py

import copy

import pytest

from app.core.orchestration.migration.dual_run import (
    DualRunCoordinator,
    DualRunMode,
    MigrationBlocked,
    N8nWorkflowConverter,
    ShadowIsolation,
    validate_mode_transition,
)

DATAFLOW_UID = "01900000-0000-7000-8000-000000000091"
SOURCE_UID = "01900000-0000-7000-8000-000000000092"


def n8n_read_workflow():
    return {
        "id": "n8n-read-orders",
        "name": "Read orders",
        "active": True,
        "nodes": [
            {
                "id": "trigger-1",
                "name": "Manual",
                "type": "n8n-nodes-base.manualTrigger",
                "parameters": {},
            },
            {
                "id": "query-1",
                "name": "Read Orders",
                "type": "n8n-nodes-base.postgres",
                "parameters": {
                    "operation": "executeQuery",
                    "query": "SELECT id, amount FROM orders ORDER BY id",
                    "purpose": "read",
                },
            },
        ],
        "connections": {
            "Manual": {
                "main": [[{"node": "Read Orders", "type": "main", "index": 0}]]
            }
        },
    }


def test_dual_run_modes_allow_only_guarded_forward_and_rollback_paths():
    assert validate_mode_transition(
        DualRunMode.N8N_PRIMARY,
        DualRunMode.N8N_PRIMARY_KESTRA_SHADOW,
    )
    assert validate_mode_transition(
        DualRunMode.N8N_PRIMARY_KESTRA_SHADOW,
        DualRunMode.KESTRA_PRIMARY_N8N_STANDBY,
    )
    assert validate_mode_transition(
        DualRunMode.KESTRA_PRIMARY_N8N_STANDBY,
        DualRunMode.KESTRA_PRIMARY,
    )
    assert validate_mode_transition(
        DualRunMode.KESTRA_PRIMARY_N8N_STANDBY,
        DualRunMode.N8N_PRIMARY_KESTRA_SHADOW,
    )

    with pytest.raises(ValueError, match="unsupported migration transition"):
        validate_mode_transition(
            DualRunMode.N8N_PRIMARY,
            DualRunMode.KESTRA_PRIMARY,
        )


def test_converter_builds_read_only_workflow_spec_without_credentials():
    converted = N8nWorkflowConverter().convert(
        n8n_read_workflow(),
        dataflow_uid=DATAFLOW_UID,
        data_source_uids={"Read Orders": SOURCE_UID},
    )

    assert converted.workflow_spec["nodes"] == [
        {
            "id": "read_orders",
            "type": "sql.query",
            "data_source_uid": SOURCE_UID,
            "purpose": "read",
            "config": {
                "statement": "SELECT id, amount FROM orders ORDER BY id",
                "parameters": {},
            },
        }
    ]
    assert converted.workflow_spec["edges"] == []
    assert converted.schedule_plan["triggers"] == [{"type": "manual"}]
    assert converted.definition_hash
    assert "credential" not in repr(converted).lower()


def test_converter_blocks_community_nodes_with_actionable_report():
    workflow = n8n_read_workflow()
    workflow["nodes"].append(
        {
            "id": "community-1",
            "name": "Community magic",
            "type": "@vendor/n8n-nodes-secret-magic",
            "parameters": {"operation": "run"},
        }
    )

    with pytest.raises(MigrationBlocked) as error:
        N8nWorkflowConverter().convert(
            workflow,
            dataflow_uid=DATAFLOW_UID,
            data_source_uids={"Read Orders": SOURCE_UID},
        )

    assert error.value.report["status"] == "blocked"
    assert error.value.report["workflow_id"] == "n8n-read-orders"
    assert error.value.report["blockers"] == [
        {
            "node_name": "Community magic",
            "node_type": "@vendor/n8n-nodes-secret-magic",
            "reason_code": "unsupported_node_type",
        }
    ]
    assert "script" not in repr(error.value.report).lower()


class Engine:
    def __init__(self, name):
        self.name = name
        self.calls = []

    def execute(self, definition_id, inputs):
        self.calls.append((definition_id, copy.deepcopy(inputs)))
        return {
            "execution_id": f"{self.name}-execution-1",
            "status": "success",
        }


class Store:
    def __init__(self):
        self.records = []

    def record_dual_run(self, record):
        self.records.append(copy.deepcopy(record))
        return copy.deepcopy(record)


def test_shadow_execution_reuses_input_snapshot_and_never_changes_primary():
    converted = N8nWorkflowConverter().convert(
        n8n_read_workflow(),
        dataflow_uid=DATAFLOW_UID,
        data_source_uids={"Read Orders": SOURCE_UID},
    )
    n8n = Engine("n8n")
    kestra = Engine("kestra")
    store = Store()
    coordinator = DualRunCoordinator(
        n8n_engine=n8n,
        kestra_engine=kestra,
        store=store,
    )

    result = coordinator.execute_shadow(
        dataflow_uid=DATAFLOW_UID,
        environment="test",
        mode=DualRunMode.N8N_PRIMARY_KESTRA_SHADOW,
        n8n_definition_id="n8n-read-orders",
        kestra_definition_id="kestra-read-orders",
        workflow_spec=converted.workflow_spec,
        inputs={"day": "2026-07-19"},
        correlation_id="01900000-0000-7000-8000-000000000093",
        isolation=ShadowIsolation.read_only(),
    )

    assert n8n.calls[0][1] == kestra.calls[0][1]
    assert result["input_snapshot_hash"] == store.records[0]["input_snapshot_hash"]
    assert result["primary_engine"] == "n8n"
    assert result["shadow_engine"] == "kestra"
    assert result["formal_engine_changed"] is False


def test_shadow_write_requires_explicit_isolated_target():
    write_spec = N8nWorkflowConverter().convert(
        n8n_read_workflow(),
        dataflow_uid=DATAFLOW_UID,
        data_source_uids={"Read Orders": SOURCE_UID},
    ).workflow_spec
    write_spec["nodes"][0].update(
        {
            "type": "sql.execute",
            "purpose": "write",
            "idempotency": {
                "strategy": "partition_replace",
                "key": "orders:${parameters.day}",
            },
        }
    )

    with pytest.raises(ValueError, match="isolated shadow target"):
        ShadowIsolation.read_only().validate(write_spec)

    write_spec["nodes"][0]["config"]["shadow_target"] = (
        "dataops_shadow.orders_20260719"
    )
    isolated = ShadowIsolation.isolated_target("dataops_shadow.")
    assert isolated.validate(write_spec)["strategy"] == "isolated_target"