gabriel / musehub public
test_walk_dag_async.py python
198 lines 7.2 KB
Raw
sha256:ef10830ce231e0a20efcb0e2586cb879471247e916616e6fdd0d51df459e2595 fix: typing audit — 0 violations, 0 untyped defs across all… Sonnet 4.6 minor ⚠ breaking 22 days ago
1 """TDD — Phase 1 of issue #40: walk_dag_async utility.
2
3 W1 Structural — walk_dag_async is an async generator
4 W2 Behavioural — BFS order: linear chain A→B→C yields A, B, C
5 W3 Behavioural — exclude set prevents visiting excluded nodes and subtrees
6 W4 Behavioural — max_nodes cap stops the walk after N nodes
7 W5 Behavioural — DFS order: linear chain yields depth-first (same for linear,
8 but diamond graph distinguishes BFS from DFS)
9 W6 Behavioural — multi-root start seeds from all roots simultaneously
10 W7 Behavioural — cycles handled: each node yielded at most once
11 """
12 from __future__ import annotations
13
14 import inspect
15 import pytest
16
17 AdjGraph = dict[str, list[str]]
18
19 # ---------------------------------------------------------------------------
20 # Helpers
21 # ---------------------------------------------------------------------------
22
23 async def _adj(graph: AdjGraph) -> None:
24 """Return an async adjacency function over an in-memory dict."""
25 async def adjacency(node: str) -> list[str]:
26 return graph.get(node, [])
27 return adjacency
28
29
30 # ---------------------------------------------------------------------------
31 # W1 Structural — walk_dag_async is an async generator
32 # ---------------------------------------------------------------------------
33
34 def test_w1_walk_dag_async_is_async_generator() -> None:
35 """walk_dag_async must return an async generator (async iterator)."""
36 from musehub.graph.walk import walk_dag_async
37
38 graph = {"A": ["B"], "B": []}
39 result = walk_dag_async("A", lambda n: _noop_adj(n, graph))
40
41 assert hasattr(result, "__aiter__"), (
42 "walk_dag_async must return an object with __aiter__ (async generator)"
43 )
44 assert hasattr(result, "__anext__"), (
45 "walk_dag_async must return an object with __anext__ (async generator)"
46 )
47
48
49 async def _noop_adj(node: str, graph: AdjGraph) -> list[str]:
50 return graph.get(node, [])
51
52
53 # ---------------------------------------------------------------------------
54 # W2 Behavioural — BFS order: linear chain
55 # ---------------------------------------------------------------------------
56
57 @pytest.mark.asyncio
58 async def test_w2_bfs_linear_chain() -> None:
59 """BFS on A→B→C yields nodes in breadth-first order: A, B, C."""
60 from musehub.graph.walk import walk_dag_async
61
62 graph = {"A": ["B"], "B": ["C"], "C": []}
63
64 async def adj(node: str) -> list[str]:
65 return graph.get(node, [])
66
67 result = [n async for n in walk_dag_async("A", adj)]
68 assert result == ["A", "B", "C"], f"Expected BFS order A→B→C, got {result}"
69
70
71 # ---------------------------------------------------------------------------
72 # W3 Behavioural — exclude set
73 # ---------------------------------------------------------------------------
74
75 @pytest.mark.asyncio
76 async def test_w3_exclude_set() -> None:
77 """Nodes in exclude are not visited and their subtrees are not explored.
78
79 Graph: A→B→C, A→D. exclude={B}.
80 Expected: A, D (B and C skipped).
81 """
82 from musehub.graph.walk import walk_dag_async
83
84 graph = {"A": ["B", "D"], "B": ["C"], "C": [], "D": []}
85
86 async def adj(node: str) -> list[str]:
87 return graph.get(node, [])
88
89 result = [n async for n in walk_dag_async("A", adj, exclude={"B"})]
90 assert "B" not in result, "B must not appear — it is in exclude"
91 assert "C" not in result, "C must not appear — its only path goes through excluded B"
92 assert "A" in result
93 assert "D" in result
94
95
96 # ---------------------------------------------------------------------------
97 # W4 Behavioural — max_nodes cap
98 # ---------------------------------------------------------------------------
99
100 @pytest.mark.asyncio
101 async def test_w4_max_nodes_cap() -> None:
102 """Walk stops after max_nodes nodes have been yielded."""
103 from musehub.graph.walk import walk_dag_async
104
105 # Linear chain of 10 nodes
106 graph = {str(i): [str(i + 1)] for i in range(9)}
107 graph["9"] = []
108
109 async def adj(node: str) -> list[str]:
110 return graph.get(node, [])
111
112 result = [n async for n in walk_dag_async("0", adj, max_nodes=3)]
113 assert len(result) == 3, f"Expected exactly 3 nodes, got {len(result)}"
114
115
116 # ---------------------------------------------------------------------------
117 # W5 Behavioural — DFS order (diamond graph distinguishes from BFS)
118 # ---------------------------------------------------------------------------
119
120 @pytest.mark.asyncio
121 async def test_w5_dfs_order() -> None:
122 """DFS on a diamond graph explores depth-first.
123
124 Diamond: A → B, A → C; B → D; C → D.
125 BFS order: A B C D (level-by-level)
126 DFS order: A B D C (or A C D B — depends on stack direction)
127
128 We assert the DFS result differs from BFS and D appears before C (or B).
129 """
130 from musehub.graph.walk import walk_dag_async
131
132 graph = {"A": ["B", "C"], "B": ["D"], "C": ["D"], "D": []}
133
134 async def adj(node: str) -> list[str]:
135 return graph.get(node, [])
136
137 bfs_result = [n async for n in walk_dag_async("A", adj, order="bfs")]
138 dfs_result = [n async for n in walk_dag_async("A", adj, order="dfs")]
139
140 assert set(bfs_result) == {"A", "B", "C", "D"}, "BFS must visit all nodes"
141 assert set(dfs_result) == {"A", "B", "C", "D"}, "DFS must visit all nodes"
142 assert bfs_result != dfs_result, (
143 "DFS and BFS must produce different orderings on a diamond graph"
144 )
145 # BFS: A B C D (level-by-level).
146 # DFS with deque.pop() (LIFO): neighbours are appended in order [B, C],
147 # so C is popped first → order is A C D B. D must appear before B in DFS.
148 assert dfs_result.index("D") < dfs_result.index("B"), (
149 f"In DFS, D should be reached before B; got {dfs_result}"
150 )
151
152
153 # ---------------------------------------------------------------------------
154 # W6 Behavioural — multi-root
155 # ---------------------------------------------------------------------------
156
157 @pytest.mark.asyncio
158 async def test_w6_multi_root() -> None:
159 """Multi-root seeds the BFS from all start nodes simultaneously.
160
161 Two disjoint chains: A→B and X→Y.
162 Starting from [A, X], all four nodes must be yielded.
163 """
164 from musehub.graph.walk import walk_dag_async
165
166 graph = {"A": ["B"], "B": [], "X": ["Y"], "Y": []}
167
168 async def adj(node: str) -> list[str]:
169 return graph.get(node, [])
170
171 result = [n async for n in walk_dag_async(["A", "X"], adj)]
172 assert set(result) == {"A", "B", "X", "Y"}, (
173 f"Multi-root must reach all nodes; got {result}"
174 )
175
176
177 # ---------------------------------------------------------------------------
178 # W7 Behavioural — cycle safety
179 # ---------------------------------------------------------------------------
180
181 @pytest.mark.asyncio
182 async def test_w7_cycle_safety() -> None:
183 """Each node is yielded at most once even when the graph has cycles.
184
185 Cycle: A→B→C→A. Walk must terminate and yield each node once.
186 """
187 from musehub.graph.walk import walk_dag_async
188
189 graph = {"A": ["B"], "B": ["C"], "C": ["A"]}
190
191 async def adj(node: str) -> list[str]:
192 return graph.get(node, [])
193
194 result = [n async for n in walk_dag_async("A", adj)]
195 assert sorted(result) == ["A", "B", "C"], (
196 f"Each node must appear exactly once; got {result}"
197 )
198 assert len(result) == len(set(result)), "No node may be yielded twice"
File History 1 commit
sha256:ef10830ce231e0a20efcb0e2586cb879471247e916616e6fdd0d51df459e2595 fix: typing audit — 0 violations, 0 untyped defs across all… Sonnet 4.6 minor ⚠ 22 days ago