From a8948b3833a9f7e6e3c6d8e3044da37837f36bbb Mon Sep 17 00:00:00 2001 From: Tirth Kanani Date: Fri, 17 Jul 2026 15:41:46 +0100 Subject: [PATCH] feat(query): surface bounded transitive test coverage Port the safe tests_for behavior from PR #334 while preserving sanitized node responses, explicit direct/indirect markers, bounded traversal, and explicit ambiguity. Co-authored-by: Gideon Zenz <91069374+gzenz@users.noreply.github.com> --- code_review_graph/tools/query.py | 28 +++++++----- tests/test_tools.py | 74 +++++++++++++++++++++++++++++--- 2 files changed, 87 insertions(+), 15 deletions(-) diff --git a/code_review_graph/tools/query.py b/code_review_graph/tools/query.py index 8d865a47..41646c00 100644 --- a/code_review_graph/tools/query.py +++ b/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] diff --git a/tests/test_tools.py b/tests/test_tools.py index 27845d54..5e88c066 100644 --- a/tests/test_tools.py +++ b/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: