feat(query): surface bounded transitive test coverage

code_review_graph/tools/query.py

@@ -319,21 +319,29 @@ def query_graph(
                         results.append(node_to_dict(child))
 
         elif pattern == "tests_for":
-            # TESTED_BY edges are stored as source=production, target=test
-            # by the parser, so look them up by source. See: #515
-            for e in store.get_edges_by_source(qn):
-                if e.kind == "TESTED_BY":
-                    test = store.get_node(e.target_qualified)
-                    if test:
-                        results.append(node_to_dict(test))
+            # Keep the normal sanitized node response while adding the
+            # direct/indirect marker returned by the bounded store lookup.
+            seen: set[str] = set()
+            for match in store.get_transitive_tests(qn):
+                test_qn = match.get("qualified_name")
+                if not isinstance(test_qn, str) or test_qn in seen:
+                    continue
+                test = store.get_node(test_qn)
+                if test:
+                    result = node_to_dict(test)
+                    result["indirect"] = bool(match.get("indirect", False))
+                    results.append(result)
+                    seen.add(test_qn)
             # Also search by naming convention
             name = node.name if node else target
             test_nodes = store.search_nodes(f"test_{name}", limit=10)
             test_nodes += store.search_nodes(f"Test{name}", limit=10)
-            seen = {r.get("qualified_name") for r in results}
             for t in test_nodes:
                 if t.qualified_name not in seen and t.is_test:
-                    results.append(node_to_dict(t))
+                    result = node_to_dict(t)
+                    result["indirect"] = False
+                    results.append(result)
+                    seen.add(t.qualified_name)
 
         elif pattern == "inheritors_of":
             for e in store.get_edges_by_target(qn):
@@ -368,7 +376,7 @@ def query_graph(
             minimal_results = [
                 {
                     k: r[k]
-                    for k in ("name", "kind", "file_path")
+                    for k in ("name", "kind", "file_path", "indirect")
                     if k in r
                 }
                 for r in results[:5]

tests/test_tools.py

@@ -12,6 +12,7 @@
 from code_review_graph.graph import GraphStore, _sanitize_name, node_to_dict
 from code_review_graph.parser import EdgeInfo, NodeInfo
 from code_review_graph.tools import (
+    _validate_repo_root,
     get_affected_flows_func,
     get_architecture_overview_func,
     get_community_func,
@@ -22,7 +23,6 @@
     list_communities_func,
     list_flows,
     query_graph,
-    _validate_repo_root,
 )
 
 
@@ -423,22 +423,40 @@ def _seed_graph(self):
             kind="Function", name="combine", file_path="/src/calc.py",
             line_start=1, line_end=5, language="python",
         ))
+        self.store.upsert_node(NodeInfo(
+            kind="Function", name="orchestrate", file_path="/src/calc.py",
+            line_start=7, line_end=12, language="python",
+        ))
         self.store.upsert_node(NodeInfo(
             kind="File", name="/tests/spec.py", file_path="/tests/spec.py",
             line_start=1, line_end=20, language="python",
         ))
         self.store.upsert_node(NodeInfo(
-            kind="Test", name="verify_combine_behaviour",
+            kind="Test", name="verify_\x01combine_behaviour",
             file_path="/tests/spec.py",
             line_start=1, line_end=5, language="python", is_test=True,
         ))
+        self.store.upsert_node(NodeInfo(
+            kind="Function", name="shared_name", file_path="/src/first.py",
+            line_start=1, line_end=5, language="python",
+        ))
+        self.store.upsert_node(NodeInfo(
+            kind="Function", name="shared_name", file_path="/src/second.py",
+            line_start=1, line_end=5, language="python",
+        ))
         # Parser-canonical direction: source=production, target=test.
         self.store.upsert_edge(EdgeInfo(
             kind="TESTED_BY",
             source="/src/calc.py::combine",
-            target="/tests/spec.py::verify_combine_behaviour",
+            target="/tests/spec.py::verify_\x01combine_behaviour",
             file_path="/tests/spec.py", line=1,
         ))
+        self.store.upsert_edge(EdgeInfo(
+            kind="CALLS",
+            source="/src/calc.py::orchestrate",
+            target="/src/calc.py::combine",
+            file_path="/src/calc.py", line=9,
+        ))
         self.store.commit()
         # Release the writer connection so query_graph can open its own.
         self.store.close()
@@ -451,8 +469,54 @@ def test_query_graph_tests_for_finds_direct_edge(self):
             repo_root=str(self.repo_root),
         )
         assert result["status"] == "ok"
-        qns = {r["qualified_name"] for r in result["results"]}
-        assert "/tests/spec.py::verify_combine_behaviour" in qns
+        match = next(
+            r for r in result["results"]
+            if r["qualified_name"] == "/tests/spec.py::verify_combine_behaviour"
+        )
+        assert match["name"] == "verify_combine_behaviour"
+        assert match["indirect"] is False
+        assert set(match) == {
+            "id", "kind", "name", "qualified_name", "file_path",
+            "line_start", "line_end", "language", "parent_name", "is_test",
+            "indirect",
+        }
+
+    def test_query_graph_tests_for_finds_one_hop_indirect_test(self):
+        from code_review_graph.tools import query_graph
+
+        result = query_graph(
+            pattern="tests_for",
+            target="/src/calc.py::orchestrate",
+            repo_root=str(self.repo_root),
+        )
+
+        assert result["status"] == "ok"
+        match = next(
+            r for r in result["results"]
+            if r["qualified_name"] == "/tests/spec.py::verify_combine_behaviour"
+        )
+        assert match["indirect"] is True
+        assert match["is_test"] is True
+
+        minimal = query_graph(
+            pattern="tests_for",
+            target="/src/calc.py::orchestrate",
+            repo_root=str(self.repo_root),
+            detail_level="minimal",
+        )
+        assert minimal["results"][0]["indirect"] is True
+
+    def test_query_graph_tests_for_keeps_ambiguous_target_explicit(self):
+        from code_review_graph.tools import query_graph
+
+        result = query_graph(
+            pattern="tests_for",
+            target="shared_name",
+            repo_root=str(self.repo_root),
+        )
+
+        assert result["status"] == "ambiguous"
+        assert len(result["candidates"]) == 2
 
 
 class TestGetDocsSection: