Coverage for tests / unit / test_graphable.py: 100%
251 statements
« prev ^ index » next coverage.py v7.13.3, created at 2026-02-16 21:32 +0000
1from unittest.mock import MagicMock
3from pytest import raises
5from graphable.errors import GraphCycleError
6from graphable.graphable import Graphable
9class TestGraphable:
10 def test_initialization(self):
11 ref = "my_ref"
12 node = Graphable(ref)
13 assert node.reference == ref
14 assert len(node.dependents) == 0
15 assert len(node.depends_on) == 0
16 # Ensure reference is accessible and consistent
17 assert node.reference == ref
19 def test_dependencies_new(self):
20 node_a = Graphable("A")
21 node_b = Graphable("B")
23 # Test adding a new dependency using the public API
24 node_a.add_dependency(node_b)
25 assert node_b in node_a.depends_on
26 assert node_a in node_b.dependents
27 # This call should cover lines 47-51 in _add_depends_on and lines in add_dependency that call _add_dependent indirectly
29 def test_dependencies_existing(self):
30 node_a = Graphable("A")
31 node_b = Graphable("B")
33 # Add dependency once
34 node_a.add_dependency(node_b)
35 assert node_b in node_a.depends_on
36 assert node_a in node_b.dependents
37 initial_dependents_count_b = len(node_b.dependents)
38 initial_depends_on_count_a = len(node_a.depends_on)
40 # Add the same dependency again - should not change state or execute internal logic again
41 node_a.add_dependency(node_b)
42 assert node_b in node_a.depends_on
43 assert node_a in node_b.dependents
44 # Assert that adding an existing dependency does not increase counts
45 assert len(node_a.depends_on) == initial_depends_on_count_a
46 assert len(node_b.dependents) == initial_dependents_count_b
47 # This tests the 'else' path of _add_depends_on and _add_dependent
49 def test_add_dependency_and_dependent_new_items(self):
50 node_a = Graphable("A")
51 node_b = Graphable("B")
52 node_c = Graphable("C")
54 # Test _add_dependent for a new item, ensuring the 'if' block is covered
55 assert len(node_a.dependents) == 0
56 node_a._add_dependent(node_b) # This call should execute lines 36-38
57 assert node_b in node_a.dependents # Verifies line 37
58 assert len(node_a.dependents) == 1 # Verifies line 37
60 # Test _add_depends_on for a new item, ensuring the 'if' block is covered
61 assert len(node_b.depends_on) == 0
62 node_b._add_depends_on(node_a) # This call should execute lines 47-51
63 assert node_a in node_b.depends_on # Verifies line 48
64 assert len(node_b.depends_on) == 1 # Verifies line 48
66 # Test adding existing items to ensure they are not re-added (tests else path implicitly)
67 node_a._add_dependent(node_b) # Should hit the 'else' path for _add_dependent
68 assert len(node_a.dependents) == 1
69 node_b._add_depends_on(node_a) # Should hit the 'else' path for _add_depends_on
70 assert len(node_b.depends_on) == 1
72 # Test public API method add_dependency which calls _add_depends_on and dependency._add_dependent
73 node_c.add_dependency(node_a) # node_a is now a dependency of node_c
74 assert node_a in node_c.depends_on
75 assert node_c in node_a.dependents # tests _add_dependent called via public API
77 # Test public API method add_dependent which calls _add_dependent and dependent._add_depends_on
78 node_c.add_dependent(node_b) # node_b is now a dependent of node_c
79 assert node_b in node_c.dependents
80 assert (
81 node_c in node_b.depends_on
82 ) # tests _add_depends_on called via public API
84 def test_add_dependents_multiple(self):
85 node_a = Graphable("A")
86 node_b = Graphable("B")
87 node_c = Graphable("C")
88 dependents_set = {node_b, node_c}
90 # Test adding new dependents
91 node_a.add_dependents(dependents_set) # Covers lines 73-75 (logger and loop)
92 assert node_b in node_a.dependents
93 assert node_c in node_a.dependents
94 assert len(node_a.dependents) == 2
95 # Verify reciprocal dependencies (lines 73-75 should be covered by this loop)
96 assert node_a in node_b.depends_on
97 assert node_a in node_c.depends_on
98 assert len(node_b.depends_on) == 1
99 assert len(node_c.depends_on) == 1
101 # Test adding dependents again with some existing
102 node_d = Graphable("D")
103 node_a.add_dependents({node_b, node_d}) # node_b already exists, node_d is new
104 assert node_b in node_a.dependents
105 assert node_d in node_a.dependents
106 assert len(node_a.dependents) == 3 # node_b, node_c, node_d
107 assert node_a in node_d.depends_on
109 def test_add_dependencies_multiple(self):
110 node_a = Graphable("A")
111 node_b = Graphable("B")
112 node_c = Graphable("C")
113 dependencies_set = {node_b, node_c}
115 # Test adding multiple dependencies
116 node_a.add_dependencies(dependencies_set)
117 assert node_b in node_a.depends_on
118 assert node_c in node_a.depends_on
119 assert len(node_a.depends_on) == 2
120 # Verify reciprocal dependents
121 assert node_a in node_b.dependents
122 assert node_a in node_c.dependents
124 def test_tags(self):
125 node = Graphable("A")
126 node.add_tag("t1")
127 node.add_tag("t2")
128 assert node.tags == {"t1", "t2"}
129 # Ensure it's a copy
130 node.tags.add("t3")
131 assert "t3" not in node.tags
132 assert node.tags == {"t1", "t2"}
134 def test_remove_tag_existing(self):
135 node = Graphable("A")
136 node.add_tag("t1")
137 assert node.is_tagged("t1")
138 node.remove_tag("t1") # Covers line 175 (discard) and 176 (logger)
139 assert not node.is_tagged("t1")
140 assert len(node.tags) == 0
142 def test_remove_tag_non_existent(self):
143 node = Graphable("A")
144 assert not node.is_tagged("t1")
145 node.remove_tag(
146 "t1"
147 ) # Should not raise an error, covers line 175 (discard) and 176 (logger)
148 assert not node.is_tagged("t1")
149 assert len(node.tags) == 0
151 def test_reference_property_access(self):
152 ref_val = "test_ref"
153 node = Graphable(ref_val)
154 # Accessing reference property directly
155 assert node.reference == ref_val # Covers lines 153-156
156 # Accessing it again after some operations
157 node.add_tag("test")
158 assert node.reference == ref_val
159 node_b = Graphable("B")
160 node.add_dependency(node_b)
161 assert node.reference == ref_val
163 def test_dependents_property_is_copy(self):
164 node = Graphable("A")
165 deps = node.dependents
166 assert isinstance(deps, set)
167 # Modifying the returned set shouldn't modify the internal set
168 deps.add(Graphable("junk"))
169 assert len(node.dependents) == 0
171 def test_depends_on_property_is_copy(self):
172 node = Graphable("A")
173 deps = node.depends_on
174 assert isinstance(deps, set)
175 deps.add(Graphable("junk"))
176 assert len(node.depends_on) == 0
178 def test_provides_to_alias(self):
179 node_a = Graphable("A")
180 node_b = Graphable("B")
181 node_a.provides_to(node_b) # Test the alias
182 assert node_b in node_a.dependents
183 assert node_a in node_b.depends_on
185 def test_requires_alias(self):
186 node_a = Graphable("A")
187 node_b = Graphable("B")
188 node_a.requires(node_b) # Test the alias
189 assert node_b in node_a.depends_on
190 assert node_a in node_b.dependents
192 def test_cycle_detection(self):
193 a = Graphable("A")
194 b = Graphable("B")
195 c = Graphable("C")
197 a.add_dependency(b, check_cycles=True)
198 b.add_dependency(c, check_cycles=True)
200 # C -> B -> A. Adding A -> C creates a cycle.
201 with raises(GraphCycleError) as excinfo:
202 c.add_dependency(a, check_cycles=True)
203 assert "would create a cycle" in str(excinfo.value)
204 assert excinfo.value.cycle is not None
205 assert a in excinfo.value.cycle
206 assert b in excinfo.value.cycle
207 assert c in excinfo.value.cycle
209 def test_cycle_detection_dependent(self):
210 a = Graphable("A")
211 b = Graphable("B")
212 c = Graphable("C")
214 a.add_dependent(b, check_cycles=True)
215 b.add_dependent(c, check_cycles=True)
217 # A -> B -> C. Adding C -> A creates a cycle.
218 with raises(GraphCycleError) as excinfo:
219 c.add_dependent(a, check_cycles=True)
220 assert "would create a cycle" in str(excinfo.value)
222 def test_find_path(self):
223 a = Graphable("A")
224 b = Graphable("B")
225 c = Graphable("C")
226 d = Graphable("D")
228 a.add_dependent(b)
229 b.add_dependent(c)
230 c.add_dependent(d)
232 path = a.find_path(d)
233 assert path == [a, b, c, d]
235 assert (
236 a.find_path(a) is None
237 ) # Simple BFS doesn't find self unless there's a loop
239 a.add_dependent(a)
240 assert a.find_path(a) == [a, a]
242 def test_ordering(self):
243 a = Graphable("A")
244 b = Graphable("B")
245 c = Graphable("C")
247 a.add_dependent(b)
248 b.add_dependent(c)
250 # A is ancestor of B and C
251 # B is ancestor of C
252 assert a < b
253 assert b < c
254 assert a < c
256 assert c > b
257 assert b > a
258 assert c > a
260 assert a <= b
261 assert a <= a
263 assert c >= b
264 assert c >= c
266 # Incomparable nodes
267 d = Graphable("D")
268 assert not (a < d)
269 assert not (d < a)
270 assert not (a > d)
271 assert not (d > a)
273 # Equality
274 assert a == a
275 assert a != b
276 assert not (a == b)
278 def test_total_ordering_decorator(self):
279 # Verify that functools.total_ordering filled in the rest
280 a = Graphable("A")
281 b = Graphable("B")
282 a.add_dependent(b)
284 assert a <= b # Should be True
285 assert b >= a # Should be True
286 assert b > a # Should be True
288 def test_edge_attributes(self):
289 a = Graphable("A")
290 b = Graphable("B")
291 a.add_dependent(b, weight=10, label="primary")
293 assert a.edge_attributes(b)["weight"] == 10
294 assert a.edge_attributes(b)["label"] == "primary"
295 assert b.edge_attributes(a)["weight"] == 10
297 # Test updating attribute
298 a.set_edge_attribute(b, "weight", 20)
299 assert b.edge_attributes(a)["weight"] == 20
301 def test_node_duration_and_status(self):
302 a = Graphable("A")
303 assert a.duration == 0.0
304 assert a.status == "pending"
306 a.duration = 5.0
307 a.status = "running"
308 assert a.duration == 5.0
309 assert a.status == "running"
311 def test_comparison_not_implemented(self):
312 a = Graphable("A")
313 with raises(TypeError):
314 _ = a < 5
315 with raises(TypeError):
316 _ = a <= 5
317 with raises(TypeError):
318 _ = a > 5
319 with raises(TypeError):
320 _ = a >= 5
322 def test_edge_attributes_key_error(self):
323 a = Graphable("A")
324 b = Graphable("B")
325 with raises(KeyError, match="No edge between"):
326 a.edge_attributes(b)
328 def test_set_edge_attribute_upstream(self):
329 a = Graphable("A")
330 b = Graphable("B")
331 b.add_dependency(a)
333 # Set attribute from descendant to ancestor (upstream)
334 b.set_edge_attribute(a, "label", "upstream")
335 assert b.edge_attributes(a)["label"] == "upstream"
336 assert a.edge_attributes(b)["label"] == "upstream"
338 def test_set_edge_attribute_key_error(self):
339 a = Graphable("A")
340 b = Graphable("B")
341 with raises(KeyError, match="No edge between"):
342 a.set_edge_attribute(b, "key", "val")
344 def test_observer_discard_not_present(self):
345 a = Graphable("A")
346 mock_observer = MagicMock()
347 # Should not raise
348 a._unregister_observer(mock_observer)