Update tests to reflect recent fixes

Author: jacklinke

tests/test_all_paths.py

@@ -3,6 +3,7 @@
 from django.test import TestCase
 
 from tests.helpers import DAGFixtureMixin, TenNodeDAGFixtureMixin
+from tests.testapp.models import NetworkNode
 
 
 class AllPathsFromDAGTestCase(DAGFixtureMixin, TestCase):
@@ -94,3 +95,26 @@ def test_all_paths_root_to_c2(self):
         # Only one path: root -> a3 -> b3 -> c2
         paths = self.root.all_paths_as_pk_lists(self.c2)
         self.assertEqual(len(paths), 1)
+
+
+class AllUpwardPathsMaxResultsTestCase(TenNodeDAGFixtureMixin, TestCase):
+    """Tests for upward all-paths with max_results (AllUpwardPathsQuery)."""
+
+    def test_upward_all_paths_max_results(self):
+        """Upward all-paths with max_results limits results."""
+        paths = self.c1.all_paths_as_pk_lists(self.root, directional=False, max_results=1)
+        self.assertEqual(len(paths), 1)
+        self.assertEqual(paths[0][0], self.c1.pk)
+        self.assertEqual(paths[0][-1], self.root.pk)
+
+    def test_upward_all_paths_returns_all_without_limit(self):
+        """Without max_results, all upward paths are returned."""
+        paths = self.c1.all_paths_as_pk_lists(self.root, directional=False)
+        # c1 -> b3 -> a3 -> root AND c1 -> b4 -> a3 -> root
+        self.assertEqual(len(paths), 2)
+
+    def test_upward_all_paths_no_path(self):
+        """Upward all-paths returns empty when no path exists."""
+        island = NetworkNode.objects.create(name="island")
+        paths = island.all_paths_as_pk_lists(self.root, directional=False)
+        self.assertEqual(len(paths), 0)

tests/test_connected_graph.py

@@ -67,7 +67,7 @@ class ConnectedComponentsTestCase(DAGFixtureMixin, TestCase):
     """Tests for NodeManager.connected_components()."""
 
     def test_returns_two_components(self):
-        """DAGFixtureMixin has a main graph and an island — should be 2 components."""
+        """DAGFixtureMixin has a main graph and an island - should be 2 components."""
         components = NetworkNode.objects.connected_components()
         self.assertEqual(len(components), 2)
 
@@ -82,7 +82,7 @@ def test_each_component_is_queryset(self):
         """Each component should be a standard Django QuerySet (not raw)."""
         components = NetworkNode.objects.connected_components()
         for component in components:
-            # Can call .filter() — this verifies it's a regular QuerySet
+            # Can call .filter() - this verifies it's a regular QuerySet
             self.assertTrue(component.filter(pk__isnull=False).exists())
 
     def test_empty_graph(self):
@@ -101,20 +101,92 @@ def test_single_island(self):
 
 
 class ConnectedGraphFilterTestCase(DAGFixtureMixin, TestCase):
-    """Tests that ConnectedGraphQuery no-op filter methods are exercised (lines 430-446)."""
+    """Tests that ConnectedGraphQuery filter methods work correctly."""
 
-    def test_connected_graph_with_all_filter_params(self):
-        """Pass all filter kwargs -- ConnectedGraphQuery stubs them all as no-ops."""
-        edge_set = EdgeSet.objects.create(name="cg_set")
+    def setUp(self):
+        super().setUp()
+        self.edge_set = EdgeSet.objects.create(name="cg_set")
+        # Assign all edges to edge_set
+        NetworkEdge.objects.all().update(edge_set=self.edge_set)
+
+    def test_connected_graph_disallow_nodes(self):
+        """Disallowing a node should exclude it and nodes only reachable through it."""
+        disallowed = NetworkNode.objects.filter(pk=self.a1.pk)
+        result = self.root.connected_graph(disallowed_nodes_queryset=disallowed)
+        result_names = set(result.values_list("name", flat=True))
+        self.assertNotIn("a1", result_names)
+        # root, a2, a3 should still be present
+        self.assertIn("root", result_names)
+        self.assertIn("a2", result_names)
+        self.assertIn("a3", result_names)
+
+    def test_connected_graph_disallow_edges(self):
+        """Disallowing an edge should prevent traversal along it."""
+        edge_root_a1 = NetworkEdge.objects.get(parent=self.root, child=self.a1)
+        disallowed = NetworkEdge.objects.filter(pk=edge_root_a1.pk)
+        result = self.root.connected_graph(disallowed_edges_queryset=disallowed)
+        result_names = set(result.values_list("name", flat=True))
+        # a1 and b1 may still be reachable via a2->b1 path, but a1 is only reachable via root->a1
+        self.assertIn("root", result_names)
+
+    def test_connected_graph_allow_nodes(self):
+        """Only traverse through allowed nodes."""
+        allowed = NetworkNode.objects.filter(name__in=["root", "a1", "b1"])
+        result = self.root.connected_graph(allowed_nodes_queryset=allowed)
+        result_names = set(result.values_list("name", flat=True))
+        self.assertIn("root", result_names)
+        self.assertIn("a1", result_names)
+        self.assertIn("b1", result_names)
+        self.assertNotIn("a2", result_names)
+        self.assertNotIn("a3", result_names)
+
+    def test_connected_graph_allow_edges(self):
+        """Only traverse along allowed edges."""
+        edge_root_a1 = NetworkEdge.objects.get(parent=self.root, child=self.a1)
+        edge_a1_b1 = NetworkEdge.objects.get(parent=self.a1, child=self.b1)
+        allowed = NetworkEdge.objects.filter(pk__in=[edge_root_a1.pk, edge_a1_b1.pk])
+        result = self.root.connected_graph(allowed_edges_queryset=allowed)
+        result_names = set(result.values_list("name", flat=True))
+        self.assertIn("root", result_names)
+        self.assertIn("a1", result_names)
+        self.assertIn("b1", result_names)
+        self.assertNotIn("a2", result_names)
+
+    def test_connected_graph_limiting_edges_set_fk(self):
+        """Limiting edges by FK should restrict traversal to those edges."""
+        other_set = EdgeSet.objects.create(name="other_set")
+        # Only assign root->a1 and a1->b1 to other_set
+        NetworkEdge.objects.filter(parent=self.root, child=self.a1).update(edge_set=other_set)
+        NetworkEdge.objects.filter(parent=self.a1, child=self.b1).update(edge_set=other_set)
+        result = self.root.connected_graph(limiting_edges_set_fk=other_set)
+        result_names = set(result.values_list("name", flat=True))
+        self.assertIn("root", result_names)
+        self.assertIn("a1", result_names)
+        self.assertIn("b1", result_names)
+        self.assertNotIn("a2", result_names)
+
+    def test_connected_graph_limiting_nodes_set_fk_noop(self):
+        """limiting_nodes_set_fk is still a no-op but should not error."""
         node_set = NodeSet.objects.create(name="cg_ns")
-        result = self.root.connected_graph(
-            limiting_nodes_set_fk=node_set,
-            limiting_edges_set_fk=edge_set,
-            disallowed_nodes_queryset=NetworkNode.objects.filter(pk=self.island.pk),
-            disallowed_edges_queryset=NetworkEdge.objects.all(),
-            allowed_nodes_queryset=NetworkNode.objects.all(),
-            allowed_edges_queryset=NetworkEdge.objects.all(),
-        )
-        # All filters are no-ops, so result should include the full connected component
-        self.assertIn(self.root, result)
-        self.assertIn(self.a1, result)
+        result = self.root.connected_graph(limiting_nodes_set_fk=node_set)
+        # No-op, so all connected nodes should be returned
+        result_names = set(result.values_list("name", flat=True))
+        self.assertIn("root", result_names)
+        self.assertIn("a1", result_names)
+
+    def test_connected_graph_max_depth(self):
+        """max_depth should limit how far connected_graph traverses."""
+        # With max_depth=1, from root we should reach immediate neighbors only
+        result = self.root.connected_graph(max_depth=2)
+        result_names = set(result.values_list("name", flat=True))
+        # max_depth=2 means path array can be at most length 2 (root + 1 hop)
+        self.assertIn("root", result_names)
+        self.assertIn("a1", result_names)
+        self.assertIn("a2", result_names)
+        self.assertIn("a3", result_names)
+
+    def test_connected_graph_no_duplicate_rows(self):
+        """Ensure the rewritten CTE does not produce duplicate rows."""
+        result = self.root.connected_graph()
+        pks = list(result.values_list("pk", flat=True))
+        self.assertEqual(len(pks), len(set(pks)))

tests/test_depth_annotation.py

@@ -3,6 +3,7 @@
 from django.test import TestCase
 
 from tests.helpers import DAGFixtureMixin, TenNodeDAGFixtureMixin
+from tests.testapp.models import EdgeSet, NetworkEdge, NetworkNode
 
 
 class DepthAnnotationFromDAGTestCase(DAGFixtureMixin, TestCase):
@@ -88,3 +89,78 @@ def test_descendants_returns_nodes_not_just_pks(self):
         for node, depth in result:
             self.assertTrue(hasattr(node, "name"))
             self.assertIsInstance(depth, int)
+
+
+class DepthAnnotationFilterTestCase(DAGFixtureMixin, TestCase):
+    """Tests for filters on AncestorDepthQuery and DescendantDepthQuery."""
+
+    def setUp(self):
+        super().setUp()
+        self.edge_set = EdgeSet.objects.create(name="depth_set")
+        NetworkEdge.objects.all().update(edge_set=self.edge_set)
+
+    def test_descendants_with_depth_disallow_nodes(self):
+        """Disallow a1 from root's descendants_with_depth."""
+        disallowed = NetworkNode.objects.filter(pk=self.a1.pk)
+        result = self.root.descendants_with_depth(disallowed_nodes_queryset=disallowed)
+        depth_dict = {node.name: depth for node, depth in result}
+        self.assertNotIn("a1", depth_dict)
+        # a2 and a3 should still be present
+        self.assertIn("a2", depth_dict)
+        self.assertIn("a3", depth_dict)
+
+    def test_ancestors_with_depth_disallow_nodes(self):
+        """Disallow a1 from b1's ancestors_with_depth."""
+        disallowed = NetworkNode.objects.filter(pk=self.a1.pk)
+        result = self.b1.ancestors_with_depth(disallowed_nodes_queryset=disallowed)
+        depth_dict = {node.name: depth for node, depth in result}
+        self.assertNotIn("a1", depth_dict)
+        # a2 should still be reachable
+        self.assertIn("a2", depth_dict)
+
+    def test_descendants_with_depth_allow_nodes(self):
+        """Only allow specific nodes in root's descendants_with_depth."""
+        allowed = NetworkNode.objects.filter(pk__in=[self.a1.pk, self.b1.pk])
+        result = self.root.descendants_with_depth(allowed_nodes_queryset=allowed)
+        depth_dict = {node.name: depth for node, depth in result}
+        self.assertIn("a1", depth_dict)
+        self.assertNotIn("a2", depth_dict)
+
+    def test_ancestors_with_depth_allow_nodes(self):
+        """Only allow specific nodes in b1's ancestors_with_depth."""
+        allowed = NetworkNode.objects.filter(pk__in=[self.a1.pk, self.root.pk])
+        result = self.b1.ancestors_with_depth(allowed_nodes_queryset=allowed)
+        depth_dict = {node.name: depth for node, depth in result}
+        self.assertNotIn("a2", depth_dict)
+
+    def test_descendants_with_depth_limiting_edges_set_fk(self):
+        """Limit descendants_with_depth to edges in a specific edge set."""
+        other_set = EdgeSet.objects.create(name="other_set")
+        # Only assign root->a1 and a1->b1 to other_set
+        NetworkEdge.objects.filter(parent=self.root, child=self.a1).update(edge_set=other_set)
+        NetworkEdge.objects.filter(parent=self.a1, child=self.b1).update(edge_set=other_set)
+        result = self.root.descendants_with_depth(limiting_edges_set_fk=other_set)
+        depth_dict = {node.name: depth for node, depth in result}
+        self.assertIn("a1", depth_dict)
+        self.assertIn("b1", depth_dict)
+        self.assertNotIn("a2", depth_dict)
+
+    def test_ancestors_with_depth_limiting_edges_set_fk(self):
+        """Limit ancestors_with_depth to edges in a specific edge set."""
+        other_set = EdgeSet.objects.create(name="other_set")
+        NetworkEdge.objects.filter(parent=self.root, child=self.a1).update(edge_set=other_set)
+        NetworkEdge.objects.filter(parent=self.a1, child=self.b1).update(edge_set=other_set)
+        result = self.b1.ancestors_with_depth(limiting_edges_set_fk=other_set)
+        depth_dict = {node.name: depth for node, depth in result}
+        self.assertIn("a1", depth_dict)
+        self.assertIn("root", depth_dict)
+        self.assertNotIn("a2", depth_dict)
+
+    def test_descendants_with_depth_disallow_edges(self):
+        """Disallow specific edges in descendants_with_depth."""
+        edge_root_a1 = NetworkEdge.objects.get(parent=self.root, child=self.a1)
+        disallowed = NetworkEdge.objects.filter(pk=edge_root_a1.pk)
+        result = self.root.descendants_with_depth(disallowed_edges_queryset=disallowed)
+        depth_dict = {node.name: depth for node, depth in result}
+        self.assertNotIn("a1", depth_dict)
+        self.assertIn("a2", depth_dict)

tests/test_mutations.py

@@ -220,13 +220,55 @@ def test_disable_circular_check(self):
         self.assertTrue(NetworkEdge.objects.filter(parent=self.n1, child=self.n2).exists())
 
     def test_allow_duplicate_edges_false(self):
-        """Should raise ValidationError when duplicate edge exists"""
+        """Should raise ValidationError when exact duplicate edge exists"""
         self.n1.add_child(self.n2)
-        with self.assertRaises(ValidationError):
+        with self.assertRaises(ValidationError) as ctx:
             self.n1.add_child(self.n2, allow_duplicate_edges=False)
+        self.assertEqual(ctx.exception.message, "An edge already exists between these nodes.")
 
     def test_duplicate_edge_allowed_by_default(self):
         """By default, duplicate edges are allowed"""
         self.n1.add_child(self.n2)
         self.n1.add_child(self.n2)
         self.assertEqual(NetworkEdge.objects.filter(parent=self.n1, child=self.n2).count(), 2)
+
+
+class RedundantEdgeCheckerTestCase(TestCase):
+    """Tests for the redundant_edge_checker (transitive reachability check)."""
+
+    def setUp(self):
+        self.n1 = NetworkNode.objects.create(name="n1")
+        self.n2 = NetworkNode.objects.create(name="n2")
+        self.n3 = NetworkNode.objects.create(name="n3")
+        self.n1.add_child(self.n2)
+        self.n2.add_child(self.n3)
+
+    def test_redundant_edge_blocked_when_disabled(self):
+        """Adding n1->n3 when n3 is already reachable via n1->n2->n3 should raise."""
+        with self.assertRaises(ValidationError) as ctx:
+            self.n1.add_child(self.n3, allow_redundant_edges=False)
+        self.assertEqual(ctx.exception.message, "The child is already reachable from the parent.")
+
+    def test_redundant_edge_allowed_by_default(self):
+        """By default, redundant edges are allowed."""
+        self.n1.add_child(self.n3)
+        self.assertTrue(NetworkEdge.objects.filter(parent=self.n1, child=self.n3).exists())
+
+    def test_duplicate_checker_does_not_block_redundant(self):
+        """duplicate_edge_checker should not block n1->n3 (not an exact duplicate)."""
+        self.n1.add_child(self.n3, allow_duplicate_edges=False)
+        self.assertTrue(NetworkEdge.objects.filter(parent=self.n1, child=self.n3).exists())
+
+    def test_both_checkers_together(self):
+        """Both checks can be enabled at the same time."""
+        # First add n1->n3 (redundant but allowed since we only block duplicates here)
+        self.n1.add_child(self.n3, allow_duplicate_edges=False)
+        # Now adding the exact same edge should fail on duplicate check
+        with self.assertRaises(ValidationError) as ctx:
+            self.n1.add_child(self.n3, allow_duplicate_edges=False, allow_redundant_edges=False)
+        self.assertEqual(ctx.exception.message, "An edge already exists between these nodes.")
+
+    def test_add_parent_passes_through_allow_redundant_edges(self):
+        """add_parent should also support allow_redundant_edges."""
+        with self.assertRaises(ValidationError):
+            self.n3.add_parent(self.n1, allow_redundant_edges=False)