model-platform/backend/tests/test_validate_dag.py

"""Unit tests for backend.services.schedules.validate_dag.

Pure function — no DB, no FastAPI, no fixtures beyond SimpleNamespace
stand-ins for the SQLAlchemy rows. The function only reads five
attributes: ``node_id``, ``node_key``, ``edge_id``, ``source_node_id``,
``target_node_id``.
"""

from __future__ import annotations

from types import SimpleNamespace

from backend.services.schedules import validate_dag


def _node(node_id: str, node_key: str) -> SimpleNamespace:
    return SimpleNamespace(node_id=node_id, node_key=node_key)


def _edge(edge_id: str, source: str, target: str) -> SimpleNamespace:
    return SimpleNamespace(
        edge_id=edge_id,
        source_node_id=source,
        target_node_id=target,
    )


def test_empty_nodes_is_rejected_as_dag_empty() -> None:
    result = validate_dag(nodes=[], edges=[])
    assert result["valid"] is False
    assert result["node_count"] == 0
    assert result["edge_count"] == 0
    assert result["topological_order"] == []
    codes = [err["code"] for err in result["errors"]]
    assert "DAG_EMPTY" in codes


def test_linear_chain_orders_by_node_key() -> None:
    nodes = [_node("n1", "A"), _node("n2", "B"), _node("n3", "C")]
    edges = [_edge("e1", "n1", "n2"), _edge("e2", "n2", "n3")]
    result = validate_dag(nodes, edges)
    assert result["valid"] is True
    assert result["root_node_ids"] == ["n1"]
    assert result["leaf_node_ids"] == ["n3"]
    assert result["topological_order"] == ["n1", "n2", "n3"]


def test_diamond_topology_is_valid() -> None:
    #   A -> B -> D
    #   A -> C -> D
    nodes = [
        _node("a", "A"),
        _node("b", "B"),
        _node("c", "C"),
        _node("d", "D"),
    ]
    edges = [
        _edge("e1", "a", "b"),
        _edge("e2", "a", "c"),
        _edge("e3", "b", "d"),
        _edge("e4", "c", "d"),
    ]
    result = validate_dag(nodes, edges)
    assert result["valid"] is True
    assert result["root_node_ids"] == ["a"]
    assert result["leaf_node_ids"] == ["d"]
    # Kahn's algorithm with node_key tie-breaking: starting at A, then B
    # and C both become ready (B alphabetically first), then D.
    assert result["topological_order"] == ["a", "b", "c", "d"]


def test_cycle_is_rejected_with_dag_cycle() -> None:
    # n1 -> n2 -> n3 -> n1
    nodes = [_node("n1", "A"), _node("n2", "B"), _node("n3", "C")]
    edges = [
        _edge("e1", "n1", "n2"),
        _edge("e2", "n2", "n3"),
        _edge("e3", "n3", "n1"),
    ]
    result = validate_dag(nodes, edges)
    assert result["valid"] is False
    codes = [err["code"] for err in result["errors"]]
    assert "DAG_CYCLE" in codes
    cycle_err = next(err for err in result["errors"] if err["code"] == "DAG_CYCLE")
    # The cycle should list every node in the cycle (sorted by node_key).
    assert set(cycle_err["node_ids"]) == {"n1", "n2", "n3"}


def test_self_edge_is_rejected_but_does_not_count_as_cycle() -> None:
    nodes = [_node("n1", "A"), _node("n2", "B")]
    edges = [
        _edge("e_self", "n1", "n1"),
        _edge("e_real", "n1", "n2"),
    ]
    result = validate_dag(nodes, edges)
    codes = [err["code"] for err in result["errors"]]
    assert "DAG_SELF_EDGE" in codes
    # The A->B edge still makes the DAG valid overall except for the self-edge.
    assert "DAG_CYCLE" not in codes
    # One node remains reachable (B), so cycle detection must not fire.
    assert result["topological_order"] == ["n1", "n2"]


def test_duplicate_edge_is_rejected_with_dag_duplicate_edge() -> None:
    nodes = [_node("n1", "A"), _node("n2", "B")]
    edges = [
        _edge("e1", "n1", "n2"),
        _edge("e1_dup", "n1", "n2"),
    ]
    result = validate_dag(nodes, edges)
    codes = [err["code"] for err in result["errors"]]
    assert "DAG_DUPLICATE_EDGE" in codes
    # The first edge still counts toward edge_count, the second is rejected.
    assert result["edge_count"] == 2


def test_edge_to_unknown_node_is_dag_edge_node_missing() -> None:
    nodes = [_node("n1", "A")]
    edges = [
        _edge("e1", "n1", "ghost"),
        _edge("e2", "ghost", "n1"),
    ]
    result = validate_dag(nodes, edges)
    codes = [err["code"] for err in result["errors"]]
    assert codes.count("DAG_EDGE_NODE_MISSING") == 2
    # No cycle should be reported for orphan edges.
    assert "DAG_CYCLE" not in codes


def test_multiple_roots_are_sorted_by_node_key() -> None:
    nodes = [
        _node("z", "Z"),
        _node("a", "A"),
        _node("m", "M"),
    ]
    edges = []
    result = validate_dag(nodes, edges)
    assert result["valid"] is True
    # All three nodes are roots (no indegree) and leaves (no outgoing).
    assert result["root_node_ids"] == ["a", "m", "z"]
    assert result["leaf_node_ids"] == ["a", "m", "z"]
    # Topological order picks the smallest node_key first.
    assert result["topological_order"] == ["a", "m", "z"]