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")