Coverage for tests / unit / compiler / test_chains.py: 0%
311 statements
« prev ^ index » next coverage.py v7.13.4, created at 2026-03-20 10:09 -0400
1"""Unit tests for the compiler chains module."""
3import networkx
5from graphqler.chains.chain import Chain
6from graphqler.chains.chain_generator import ChainGenerator
7from graphqler.chains.strategies.dfs_strategy import DFSChainStrategy
8from graphqler.chains.strategies.all_dependencies_strategy import AllDependenciesChainStrategy
9from graphqler.graph.node import Node
10from graphqler import config
13# ---------------------------------------------------------------------------
14# Helpers
15# ---------------------------------------------------------------------------
17def _make_node(name: str, graphql_type: str = "Query", mutation_type: str | None = None) -> Node:
18 node = Node(graphql_type, name, {})
19 if mutation_type:
20 node.mutation_type = mutation_type
21 return node
24def _linear_graph(*node_names: str) -> tuple[networkx.DiGraph, list[Node]]:
25 """Build a simple linear graph A -> B -> C -> ... and return (graph, [nodes])."""
26 graph = networkx.DiGraph()
27 nodes = [_make_node(name) for name in node_names]
28 graph.add_nodes_from(nodes)
29 for i in range(len(nodes) - 1):
30 graph.add_edge(nodes[i], nodes[i + 1])
31 return graph, nodes
34# ---------------------------------------------------------------------------
35# Chain dataclass
36# ---------------------------------------------------------------------------
38class TestChain:
39 def test_repr_empty(self):
40 c = Chain()
41 assert repr(c) == "Chain([])"
43 def test_repr_single(self):
44 n = _make_node("A")
45 c = Chain(nodes=[n])
46 assert "A" in repr(c)
48 def test_repr_multiple(self):
49 nodes = [_make_node(name) for name in ["A", "B", "C"]]
50 c = Chain(nodes=nodes)
51 assert repr(c) == "Chain([A -> B -> C])"
53 def test_len(self):
54 nodes = [_make_node(name) for name in ["A", "B"]]
55 assert len(Chain(nodes=nodes)) == 2
57 def test_len_empty(self):
58 assert len(Chain()) == 0
60 def test_last_node_returns_last(self):
61 nodes = [_make_node(name) for name in ["X", "Y", "Z"]]
62 c = Chain(nodes=nodes)
63 assert c.last_node() is nodes[-1]
65 def test_last_node_empty_returns_none(self):
66 assert Chain().last_node() is None
68 def test_has_mutation_type_true(self):
69 n = _make_node("m", graphql_type="Mutation", mutation_type="DELETE")
70 c = Chain(nodes=[n])
71 assert c.has_mutation_type(["DELETE"]) is True
73 def test_has_mutation_type_false(self):
74 n = _make_node("q", graphql_type="Query")
75 c = Chain(nodes=[n])
76 assert c.has_mutation_type(["DELETE", "UPDATE"]) is False
78 def test_has_mutation_type_mixed(self):
79 nodes = [
80 _make_node("create", graphql_type="Mutation", mutation_type="CREATE"),
81 _make_node("update", graphql_type="Mutation", mutation_type="UPDATE"),
82 ]
83 c = Chain(nodes=nodes)
84 assert c.has_mutation_type(["UPDATE"]) is True
85 assert c.has_mutation_type(["DELETE"]) is False
88# ---------------------------------------------------------------------------
89# DFSChainStrategy
90# ---------------------------------------------------------------------------
92class TestDFSChainStrategy:
93 def test_single_node(self):
94 graph = networkx.DiGraph()
95 n = _make_node("A")
96 graph.add_node(n)
97 chains = DFSChainStrategy().generate(graph, [n])
98 assert len(chains) == 1
99 assert chains[0].nodes == [n]
101 def test_linear_two_nodes(self):
102 graph, nodes = _linear_graph("A", "B")
103 chains = DFSChainStrategy().generate(graph, [nodes[0]])
104 node_lists = [c.nodes for c in chains]
105 assert nodes[:1] in node_lists
106 assert nodes[:2] in node_lists
107 assert len(chains) == 2
109 def test_linear_three_nodes_produces_prefix_chains(self):
110 graph, nodes = _linear_graph("A", "B", "C")
111 chains = DFSChainStrategy().generate(graph, [nodes[0]])
112 node_lists = [c.nodes for c in chains]
113 assert nodes[:1] in node_lists
114 assert nodes[:2] in node_lists
115 assert nodes[:3] in node_lists
116 assert len(chains) == 3
118 def test_branching_graph(self):
119 graph = networkx.DiGraph()
120 a, b, c = _make_node("A"), _make_node("B"), _make_node("C")
121 graph.add_edges_from([(a, b), (a, c)])
122 chains = DFSChainStrategy().generate(graph, [a])
123 node_lists = [c.nodes for c in chains]
124 assert [a] in node_lists
125 assert [a, b] in node_lists
126 assert [a, c] in node_lists
127 assert len(chains) == 3
129 def test_cycle_avoidance(self):
130 graph = networkx.DiGraph()
131 a, b = _make_node("A"), _make_node("B")
132 graph.add_edges_from([(a, b), (b, a)])
133 chains = DFSChainStrategy().generate(graph, [a])
134 assert len(chains) == 2
136 def test_multiple_starters(self):
137 graph = networkx.DiGraph()
138 x, y = _make_node("X"), _make_node("Y")
139 graph.add_nodes_from([x, y])
140 chains = DFSChainStrategy().generate(graph, [x, y])
141 node_lists = [c.nodes for c in chains]
142 assert [x] in node_lists
143 assert [y] in node_lists
145 def test_empty_graph(self):
146 graph = networkx.DiGraph()
147 chains = DFSChainStrategy().generate(graph, [])
148 assert chains == []
150 def test_filter_stops_at_filtered_node(self):
151 """Chains should not include DELETE node when DELETE is filtered."""
152 graph = networkx.DiGraph()
153 create = _make_node("create", graphql_type="Mutation", mutation_type="CREATE")
154 user = _make_node("User", graphql_type="Object")
155 delete = _make_node("delete", graphql_type="Mutation", mutation_type="DELETE")
156 graph.add_edges_from([(create, user), (user, delete)])
158 chains = DFSChainStrategy().generate(graph, [create], filter_mutation_type=["DELETE"])
159 for chain in chains:
160 assert delete not in chain.nodes
161 node_lists = [c.nodes for c in chains]
162 assert [create] in node_lists
163 assert [create, user] in node_lists
164 assert len(chains) == 2
166 def test_filter_none_includes_all(self):
167 graph = networkx.DiGraph()
168 create = _make_node("create", graphql_type="Mutation", mutation_type="CREATE")
169 delete = _make_node("delete", graphql_type="Mutation", mutation_type="DELETE")
170 graph.add_edge(create, delete)
171 chains = DFSChainStrategy().generate(graph, [create], filter_mutation_type=None)
172 assert len(chains) == 2
174 def test_filter_all_mutations_keeps_only_queries(self):
175 graph = networkx.DiGraph()
176 query = _make_node("getUser", graphql_type="Query")
177 create = _make_node("createUser", graphql_type="Mutation", mutation_type="CREATE")
178 graph.add_nodes_from([query, create])
179 chains = DFSChainStrategy().generate(
180 graph, [query, create],
181 filter_mutation_type=["CREATE", "UPDATE", "DELETE", "UNKNOWN"],
182 )
183 for chain in chains:
184 assert create not in chain.nodes
185 assert any(query in c.nodes for c in chains)
188# ---------------------------------------------------------------------------
189# ChainGenerator
190# ---------------------------------------------------------------------------
192class TestChainGenerator:
193 def test_default_strategy_is_all_dependencies(self):
194 gen = ChainGenerator()
195 assert isinstance(gen._strategy, AllDependenciesChainStrategy)
197 def test_custom_strategy_accepted(self):
198 custom = DFSChainStrategy()
199 gen = ChainGenerator(strategy=custom)
200 assert gen._strategy is custom
202 def test_chains_empty_before_generate(self):
203 gen = ChainGenerator()
204 assert gen.chains == []
206 def test_generate_returns_and_stores_chains(self):
207 graph, nodes = _linear_graph("A", "B")
208 gen = ChainGenerator()
209 result = gen.generate(graph, [nodes[0]])
210 assert len(result) > 0
211 assert gen.chains is result
213 def test_generate_overwrites_previous_chains(self):
214 graph, nodes = _linear_graph("A", "B")
215 gen = ChainGenerator()
216 gen.generate(graph, [nodes[0]])
217 first = gen.chains
219 graph2, nodes2 = _linear_graph("X")
220 gen.generate(graph2, [nodes2[0]])
221 assert gen.chains is not first
223 def test_chains_inspectable_after_generate(self):
224 graph, nodes = _linear_graph("A", "B", "C")
225 gen = ChainGenerator()
226 gen.generate(graph, [nodes[0]])
227 assert len(gen.chains) > 0
228 for chain in gen.chains:
229 assert isinstance(chain, Chain)
231 def test_three_pass_produces_triplicated_query_chains(self):
232 """Pure query nodes are never filtered → each prefix chain appears in all 3 passes."""
233 graph, nodes = _linear_graph("A") # single Query node
234 gen = ChainGenerator()
235 chains = gen.generate(graph, [nodes[0]])
236 # pass1 + pass2 + pass3, each with 1 chain → 3 total
237 assert len(chains) == 3
239 def test_delete_chain_excluded_from_pass1_and_pass2(self):
240 """A chain containing a DELETE node should not appear in pass1 or pass2."""
241 graph = networkx.DiGraph()
242 create = _make_node("create", graphql_type="Mutation", mutation_type="CREATE")
243 delete = _make_node("delete", graphql_type="Mutation", mutation_type="DELETE")
244 graph.add_edge(create, delete)
246 gen = ChainGenerator()
247 chains = gen.generate(graph, [create])
249 chains_with_delete = [c for c in chains if delete in c.nodes]
250 # Only pass3 generates [create, delete]
251 assert len(chains_with_delete) >= 1
253 def test_disable_mutations_excludes_all_mutations(self):
254 original = config.DISABLE_MUTATIONS
255 try:
256 config.DISABLE_MUTATIONS = True
257 graph = networkx.DiGraph()
258 query = _make_node("getUser", graphql_type="Query")
259 create = _make_node("createUser", graphql_type="Mutation", mutation_type="CREATE")
260 graph.add_nodes_from([query, create])
261 gen = ChainGenerator()
262 chains = gen.generate(graph, [query, create])
263 for chain in chains:
264 for node in chain.nodes:
265 assert node.graphql_type != "Mutation"
266 finally:
267 config.DISABLE_MUTATIONS = original
269 def test_disable_mutations_false_includes_mutations(self):
270 original = config.DISABLE_MUTATIONS
271 try:
272 config.DISABLE_MUTATIONS = False
273 graph = networkx.DiGraph()
274 create = _make_node("createUser", graphql_type="Mutation", mutation_type="CREATE")
275 graph.add_node(create)
276 gen = ChainGenerator()
277 chains = gen.generate(graph, [create])
278 mutation_chains = [c for c in chains if any(n.graphql_type == "Mutation" for n in c.nodes)]
279 assert len(mutation_chains) > 0
280 finally:
281 config.DISABLE_MUTATIONS = original
284# ---------------------------------------------------------------------------
285# AllDependenciesChainStrategy
286# ---------------------------------------------------------------------------
290def _names(chain: "Chain") -> list:
291 """Return the list of node names in a chain."""
292 return [n.name for n in chain.nodes]
295class TestAllDependenciesChainStrategy:
296 def _strategy(self) -> AllDependenciesChainStrategy:
297 return AllDependenciesChainStrategy()
299 def test_single_node(self):
300 graph, nodes = _linear_graph("A")
301 chains = self._strategy().generate(graph, [])
302 assert len(chains) == 1
303 assert _names(chains[0]) == ["A"]
305 def test_linear_chain_produces_one_chain_per_node(self):
306 """A -> B -> C -> D: each node gets its own self-sufficient chain."""
307 graph, nodes = _linear_graph("A", "B", "C", "D")
308 chains = self._strategy().generate(graph, [])
309 assert len(chains) == 4
311 chain_by_last = {_names(c)[-1]: _names(c) for c in chains}
312 assert chain_by_last["A"] == ["A"]
313 assert chain_by_last["B"] == ["A", "B"]
314 assert chain_by_last["C"] == ["A", "B", "C"]
315 assert chain_by_last["D"] == ["A", "B", "C", "D"]
317 def test_diamond_multi_parent(self):
318 """B and C both depend on A; D depends on both B and C."""
319 A, B, C, D = [_make_node(n) for n in "ABCD"]
320 graph = networkx.DiGraph()
321 graph.add_edge(A, B)
322 graph.add_edge(A, C)
323 graph.add_edge(B, D)
324 graph.add_edge(C, D)
326 chains = self._strategy().generate(graph, [])
327 chain_by_last = {_names(c)[-1]: _names(c) for c in chains}
329 d_chain = chain_by_last["D"]
330 assert set(d_chain) == {"A", "B", "C", "D"}
331 assert d_chain[-1] == "D"
332 assert d_chain.index("A") < d_chain.index("B")
333 assert d_chain.index("A") < d_chain.index("C")
334 assert d_chain.index("B") < d_chain.index("D")
335 assert d_chain.index("C") < d_chain.index("D")
337 def test_deep_three_layer_multi_dependency(self):
338 """Deep 3-layer graph where Z has 2 mid-level deps each with 2-3 root deps.
340 Layer 0 (roots): A, B, C, E, F
341 Layer 1 (mid): D <- (A, B, C); H <- (E, F)
342 Layer 2 (final): Z <- (D, H)
343 """
344 A, B, C, D, E, F, H, Z = [_make_node(n) for n in ["A", "B", "C", "D", "E", "F", "H", "Z"]]
345 graph = networkx.DiGraph()
346 graph.add_edge(A, D)
347 graph.add_edge(B, D)
348 graph.add_edge(C, D)
349 graph.add_edge(E, H)
350 graph.add_edge(F, H)
351 graph.add_edge(D, Z)
352 graph.add_edge(H, Z)
354 chains = self._strategy().generate(graph, [])
355 chain_by_last = {_names(c)[-1]: _names(c) for c in chains}
357 z_chain = chain_by_last["Z"]
358 assert set(z_chain) == {"A", "B", "C", "D", "E", "F", "H", "Z"}
359 assert z_chain[-1] == "Z"
360 for root in ("A", "B", "C"):
361 assert z_chain.index(root) < z_chain.index("D")
362 for root in ("E", "F"):
363 assert z_chain.index(root) < z_chain.index("H")
364 assert z_chain.index("D") < z_chain.index("Z")
365 assert z_chain.index("H") < z_chain.index("Z")
367 d_chain = chain_by_last["D"]
368 assert set(d_chain) == {"A", "B", "C", "D"}
369 assert d_chain[-1] == "D"
371 h_chain = chain_by_last["H"]
372 assert set(h_chain) == {"E", "F", "H"}
373 assert h_chain[-1] == "H"
375 def test_filter_excludes_nodes(self):
376 """A filtered node should produce no chain."""
377 create = _make_node("create", graphql_type="Mutation", mutation_type="CREATE")
378 obj = _make_node("Obj", graphql_type="Object")
379 delete = _make_node("delete", graphql_type="Mutation", mutation_type="DELETE")
380 graph = networkx.DiGraph()
381 graph.add_edge(create, obj)
382 graph.add_edge(obj, delete)
384 chains = self._strategy().generate(graph, [], filter_mutation_type=["DELETE"])
385 all_last_nodes = {_names(c)[-1] for c in chains}
386 assert "delete" not in all_last_nodes
388 def test_filtered_ancestor_excluded_from_descendant_chain(self):
389 """If an ancestor is filtered, it should be absent from the descendant chain."""
390 update = _make_node("update", graphql_type="Mutation", mutation_type="UPDATE")
391 obj = _make_node("Obj", graphql_type="Object")
392 query = _make_node("getObj", graphql_type="Query")
393 graph = networkx.DiGraph()
394 graph.add_edge(update, obj)
395 graph.add_edge(obj, query)
397 chains = self._strategy().generate(graph, [], filter_mutation_type=["UPDATE"])
398 chain_by_last = {_names(c)[-1]: _names(c) for c in chains}
400 assert "update" not in chain_by_last
401 q_chain = chain_by_last.get("getObj", [])
402 assert "update" not in q_chain
404 def test_starter_nodes_ignored(self):
405 """AllDependenciesChainStrategy ignores starter_nodes; all nodes get a chain."""
406 graph, nodes = _linear_graph("A", "B", "C")
407 chains_empty = self._strategy().generate(graph, [])
408 chains_with = self._strategy().generate(graph, [nodes[0]])
409 assert len(chains_empty) == len(chains_with)
412class TestChainGeneratorUsesAllDependenciesDefault:
413 def test_default_strategy_is_all_dependencies(self):
414 gen = ChainGenerator()
415 assert isinstance(gen._strategy, AllDependenciesChainStrategy)
417 def test_generate_with_diamond_includes_all_ancestors(self):
418 """ChainGenerator default strategy handles diamond graphs correctly."""
419 A, B, C, D = [_make_node(n) for n in "ABCD"]
420 graph = networkx.DiGraph()
421 graph.add_edge(A, B)
422 graph.add_edge(A, C)
423 graph.add_edge(B, D)
424 graph.add_edge(C, D)
426 gen = ChainGenerator()
427 chains = gen.generate(graph, [])
428 # All 3 passes x 4 nodes each = 12 chains (all Query type, never filtered)
429 all_d_chains = [c for c in chains if D in c.nodes and c.nodes[-1] == D]
430 assert any(len(c.nodes) == 4 for c in all_d_chains)