DataOps-platform/tests/test_workflow_repository.py

def test_workflow_hash_is_canonical_and_credentials_are_redacted():
    from app.core.data_flow.workflow_repository import safe_workflow_snapshot, workflow_hash

    first = {"nodes": [{"id": "1", "credentials": {"postgres": {"id": "secret"}}}], "name": "flow", "updatedAt": "today"}
    second = {"name": "flow", "nodes": [{"credentials": {"postgres": {"id": "changed"}}, "id": "1"}], "updatedAt": "tomorrow"}

    assert workflow_hash(first) == workflow_hash(second)
    snapshot = safe_workflow_snapshot(first)
    assert snapshot["nodes"][0]["credentials"] == "[redacted]"
    assert "updatedAt" not in snapshot


def test_numeric_dataflow_ids_are_rejected():
    import pytest
    from app.core.data_flow.workflow_repository import validate_dataflow_uid

    with pytest.raises(ValueError):
        validate_dataflow_uid("123")


def test_legacy_n8n_repository_populates_engine_neutral_columns():
    from pathlib import Path

    repository = (
        Path(__file__).resolve().parents[1]
        / "app"
        / "core"
        / "data_flow"
        / "workflow_repository.py"
    ).read_text(encoding="utf-8")

    assert "engine_type" in repository
    assert "engine_definition_id" in repository
    assert '"n8n"' in repository