| 123456789101112131415161718192021222324252627282930 |
- from app.commands.migrate_n8n_workflow import convert_export
- from app.commands.reconcile_dual_run import reconcile_exports
- from tests.integration.test_n8n_kestra_dual_run import (
- DATAFLOW_UID,
- SOURCE_UID,
- n8n_read_workflow,
- )
- from tests.integration.test_workflow_reconciliation import snapshot
- def test_conversion_command_is_dry_run_and_credential_free():
- report = convert_export(
- n8n_read_workflow(),
- dataflow_uid=DATAFLOW_UID,
- data_source_uids={"Read Orders": SOURCE_UID},
- )
- assert report["status"] == "converted"
- assert report["deployment_performed"] is False
- assert "credential" not in repr(report).lower()
- def test_reconciliation_command_uses_bounded_metrics_only():
- report = reconcile_exports(
- snapshot(engine="n8n", execution_id="n8n-1").__dict__,
- snapshot(engine="kestra", execution_id="kestra-1").__dict__,
- )
- assert report["status"] == "passed"
- assert "rows" not in repr(report).lower()
|
