| 12345678910111213141516171819202122232425 |
- from app.core.orchestration.migration.cutover import WorkflowCutoverService
- from app.core.orchestration.migration.dual_run import DualRunMode
- from tests.integration.test_workflow_engine_cutover import Engines, Store
- def test_explicit_rollback_disables_kestra_before_reactivating_n8n():
- store = Store()
- store.mode = DualRunMode.KESTRA_PRIMARY_N8N_STANDBY
- engines = Engines()
- service = WorkflowCutoverService(store=store, engines=engines)
- result = service.rollback_to_n8n(
- dataflow_uid="01900000-0000-7000-8000-000000000094",
- environment="test",
- idempotency_key="rollback:orders:v55",
- correlation_id="01900000-0000-7000-8000-000000000095",
- )
- assert result["status"] == "succeeded"
- assert result["mode"] == "n8n_primary_kestra_shadow"
- assert engines.calls == [
- ("deactivate_kestra", "kestra-orders"),
- ("activate_n8n", "n8n-orders"),
- ]
- assert store.mode is DualRunMode.N8N_PRIMARY_KESTRA_SHADOW
|
