beam: deterministic topo sort with cycle detection + tests

wizard-work/beam/test_topo_sort.py

@@ -0,0 +1,60 @@
+"""Tests for topo_sort."""
+import sys
+sys.path.insert(0, "/root/.hermes/wizard-outputs/beam")
+from topo_sort import topo_sort
+
+
+def test_dag_simple():
+    order, cyc = topo_sort(["a", "b", "c"], [("a", "b"), ("b", "c")])
+    assert cyc is None
+    assert order == ["a", "b", "c"]
+
+
+def test_dag_tiebreak_deterministic():
+    order, cyc = topo_sort(["z", "a", "m"], [])
+    assert order == ["a", "m", "z"]
+
+
+def test_cycle():
+    order, cyc = topo_sort(["a", "b", "c"], [("a", "b"), ("b", "c"), ("c", "a")])
+    assert order == []
+    # cycle must be a closed-ish list forming cycle
+    assert cyc is not None
+    assert len(cyc) >= 3
+    # verify edges close
+    edges = [("a", "b"), ("b", "c"), ("c", "a")]
+    eset = set(edges)
+    for i in range(len(cyc)):
+        u, v = cyc[i], cyc[(i + 1) % len(cyc)]
+        assert (u, v) in eset, (u, v)
+
+
+def test_unknown_node_raises():
+    try:
+        topo_sort(["a"], [("a", "b")])
+    except ValueError:
+        return
+    raise AssertionError("expected ValueError")
+
+
+def test_duplicate_raises():
+    try:
+        topo_sort(["a", "a"], [])
+    except ValueError:
+        return
+    raise AssertionError("expected ValueError")
+
+
+def test_partial_order():
+    order, cyc = topo_sort(["a", "b", "c", "d"], [("a", "c"), ("b", "c"), ("c", "d")])
+    assert cyc is None
+    assert order.index("a") < order.index("c")
+    assert order.index("b") < order.index("c")
+    assert order.index("c") < order.index("d")
+
+
+if __name__ == "__main__":
+    for name in [n for n in globals() if n.startswith("test_")]:
+        globals()[name]()
+        print("PASS", name)
+    print("ALL PASS")