fix(graph): correct TESTED_BY edge direction in tests_for queries

The parser stores TESTED_BY edges as source=production, target=test, but
query_graph(pattern="tests_for"), get_transitive_tests, and
load_flow_adjacency all traversed them in the wrong direction. The
naming-convention fallback (test_<name>/Test<name>) silently masked the
bug for tests following standard naming.

Changes:
- tools/query.py: flip tests_for branch to use get_edges_by_source +
  e.target_qualified
- graph.py::get_transitive_tests: flip three SQL queries to select
  target_qualified where source_qualified = ?
- graph.py::load_flow_adjacency: track production node (src) in
  has_tested_by so criticality scoring reflects real coverage

Regression tests use unconventional test names so the naming-convention
fallback cannot mask future regressions.

Fixes #515

Author: Devilthelegend

code_review_graph/graph.py

@@ -371,7 +371,10 @@ def get_transitive_tests(
     ) -> list[dict]:
         """Find tests covering a node, including indirect (transitive) coverage.
 
-        1. Direct: TESTED_BY edges targeting this node (+ bare-name fallback).
+        TESTED_BY edges are stored as source=production, target=test by
+        the parser, so look them up by source_qualified. See: #515
+
+        1. Direct: TESTED_BY edges originating at this node (+ bare-name fallback).
         2. Indirect: follow outgoing CALLS edges up to *max_depth* hops,
            then collect TESTED_BY edges on each callee.
 
@@ -415,31 +418,31 @@ def _node_dict(qn: str, indirect: bool) -> dict | None:
                 "indirect": indirect,
             }
 
-        # Direct TESTED_BY
+        # Direct TESTED_BY (source=production, target=test). See: #515
         for qn in input_qns:
             for row in conn.execute(
-                "SELECT source_qualified FROM edges "
-                "WHERE target_qualified = ? AND kind = 'TESTED_BY'",
+                "SELECT target_qualified FROM edges "
+                "WHERE source_qualified = ? AND kind = 'TESTED_BY'",
                 (qn,),
             ).fetchall():
-                src = row["source_qualified"]
-                if src not in seen:
-                    seen.add(src)
-                    d = _node_dict(src, indirect=False)
+                tgt = row["target_qualified"]
+                if tgt not in seen:
+                    seen.add(tgt)
+                    d = _node_dict(tgt, indirect=False)
                     if d:
                         results.append(d)
 
         # Bare-name fallback for direct
         bare = qualified_name.rsplit("::", 1)[-1] if "::" in qualified_name else qualified_name
         for row in conn.execute(
-            "SELECT source_qualified FROM edges "
-            "WHERE target_qualified = ? AND kind = 'TESTED_BY'",
+            "SELECT target_qualified FROM edges "
+            "WHERE source_qualified = ? AND kind = 'TESTED_BY'",
             (bare,),
         ).fetchall():
-            src = row["source_qualified"]
-            if src not in seen:
-                seen.add(src)
-                d = _node_dict(src, indirect=False)
+            tgt = row["target_qualified"]
+            if tgt not in seen:
+                seen.add(tgt)
+                d = _node_dict(tgt, indirect=False)
                 if d:
                     results.append(d)
 
@@ -458,14 +461,14 @@ def _node_dict(qn: str, indirect: bool) -> dict | None:
                 next_frontier = set(list(next_frontier)[:max_frontier])
             for callee in next_frontier:
                 for row in conn.execute(
-                    "SELECT source_qualified FROM edges "
-                    "WHERE target_qualified = ? AND kind = 'TESTED_BY'",
+                    "SELECT target_qualified FROM edges "
+                    "WHERE source_qualified = ? AND kind = 'TESTED_BY'",
                     (callee,),
                 ).fetchall():
-                    src = row["source_qualified"]
-                    if src not in seen:
-                        seen.add(src)
-                        d = _node_dict(src, indirect=True)
+                    tgt = row["target_qualified"]
+                    if tgt not in seen:
+                        seen.add(tgt)
+                        d = _node_dict(tgt, indirect=True)
                         if d:
                             results.append(d)
             frontier = next_frontier
@@ -1263,8 +1266,8 @@ def load_flow_adjacency(self) -> "FlowAdjacency":
             kind, src, tgt = row["kind"], row["source_qualified"], row["target_qualified"]
             if kind == "CALLS":
                 calls_out.setdefault(src, []).append(tgt)
-            else:  # TESTED_BY
-                has_tested_by.add(tgt)
+            else:  # TESTED_BY: source is the production node being tested. See: #515
+                has_tested_by.add(src)
 
         return FlowAdjacency(
             calls_out=calls_out,

code_review_graph/tools/query.py

@@ -293,9 +293,11 @@ def query_graph(
                         results.append(node_to_dict(child))
 
         elif pattern == "tests_for":
-            for e in store.get_edges_by_target(qn):
+            # 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.source_qualified)
+                    test = store.get_node(e.target_qualified)
                     if test:
                         results.append(node_to_dict(test))
             # Also search by naming convention

tests/test_graph.py

@@ -226,6 +226,66 @@ def test_metadata(self):
         assert self.store.get_metadata("test_key") == "test_value"
         assert self.store.get_metadata("nonexistent") is None
 
+    def test_get_transitive_tests_follows_direct_tested_by_edge(self):
+        """Regression test for #515: get_transitive_tests must follow
+        TESTED_BY edges by source_qualified (production) since the parser
+        stores source=production, target=test. The test function uses an
+        unconventional name so the bare-name fallback cannot mask the bug.
+        """
+        self.store.upsert_node(self._make_file_node("/src/calc.py"))
+        self.store.upsert_node(self._make_func_node("add", "/src/calc.py"))
+        self.store.upsert_node(self._make_file_node("/tests/check.py"))
+        self.store.upsert_node(self._make_func_node(
+            "verify_addition", "/tests/check.py", is_test=True,
+        ))
+        self.store.upsert_edge(EdgeInfo(
+            kind="TESTED_BY",
+            source="/src/calc.py::add",
+            target="/tests/check.py::verify_addition",
+            file_path="/tests/check.py", line=1,
+        ))
+        self.store.commit()
+
+        results = self.store.get_transitive_tests("/src/calc.py::add")
+        qns = {r["qualified_name"] for r in results}
+        assert "/tests/check.py::verify_addition" in qns
+        assert all(not r["indirect"] for r in results)
+
+    def test_get_transitive_tests_follows_calls_then_tested_by(self):
+        """Transitive coverage: caller -> CALLS -> callee -> TESTED_BY -> test.
+        Uses an unconventional test name so the bare-name fallback cannot
+        match. See: #515.
+        """
+        self.store.upsert_node(self._make_file_node("/src/svc.py"))
+        self.store.upsert_node(self._make_func_node("orchestrate", "/src/svc.py"))
+        self.store.upsert_node(self._make_func_node("compute", "/src/svc.py"))
+        self.store.upsert_node(self._make_file_node("/tests/check.py"))
+        self.store.upsert_node(self._make_func_node(
+            "verify_compute", "/tests/check.py", is_test=True,
+        ))
+        self.store.upsert_edge(EdgeInfo(
+            kind="CALLS", source="/src/svc.py::orchestrate",
+            target="/src/svc.py::compute", file_path="/src/svc.py", line=2,
+        ))
+        self.store.upsert_edge(EdgeInfo(
+            kind="TESTED_BY",
+            source="/src/svc.py::compute",
+            target="/tests/check.py::verify_compute",
+            file_path="/tests/check.py", line=1,
+        ))
+        self.store.commit()
+
+        results = self.store.get_transitive_tests(
+            "/src/svc.py::orchestrate", max_depth=2,
+        )
+        qns = {r["qualified_name"] for r in results}
+        assert "/tests/check.py::verify_compute" in qns
+        match = next(
+            r for r in results
+            if r["qualified_name"] == "/tests/check.py::verify_compute"
+        )
+        assert match["indirect"] is True
+
     def test_get_all_community_ids_logs_when_column_missing(self, caplog):
         conn = sqlite3.connect(":memory:")
         conn.row_factory = sqlite3.Row

tests/test_tools.py

@@ -366,6 +366,72 @@ def test_validate_repo_root_error_mentions_svn_marker(self, tmp_path):
             _validate_repo_root(tmp_path)
 
 
+class TestQueryGraphTestsFor:
+    """Regression tests for #515: query_graph(pattern='tests_for')
+    must follow direct TESTED_BY edges (source=production, target=test)
+    rather than relying on the naming-convention fallback.
+    """
+
+    def setup_method(self, tmp_path_factory=None):
+        # tmp_path_factory is unavailable in setup_method; use a TemporaryDirectory.
+        import tempfile as _tempfile
+        self._tmpdir = _tempfile.TemporaryDirectory()
+        self.repo_root = Path(self._tmpdir.name)
+        # _validate_repo_root requires .git or .code-review-graph.
+        (self.repo_root / ".code-review-graph").mkdir()
+        # find_project_root / get_db_path look here for the DB.
+        from code_review_graph.incremental import get_db_path
+        self.db_path = get_db_path(self.repo_root)
+        self.store = GraphStore(str(self.db_path))
+        self._seed_graph()
+
+    def teardown_method(self):
+        self.store.close()
+        self._tmpdir.cleanup()
+
+    def _seed_graph(self):
+        # Production function with an unconventional name so the
+        # naming-convention fallback (test_<name> / Test<name>) cannot match.
+        self.store.upsert_node(NodeInfo(
+            kind="File", name="/src/calc.py", file_path="/src/calc.py",
+            line_start=1, line_end=20, language="python",
+        ))
+        self.store.upsert_node(NodeInfo(
+            kind="Function", name="combine", file_path="/src/calc.py",
+            line_start=1, line_end=5, 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",
+            file_path="/tests/spec.py",
+            line_start=1, line_end=5, language="python", is_test=True,
+        ))
+        # 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",
+            file_path="/tests/spec.py", line=1,
+        ))
+        self.store.commit()
+        # Release the writer connection so query_graph can open its own.
+        self.store.close()
+
+    def test_query_graph_tests_for_finds_direct_edge(self):
+        from code_review_graph.tools import query_graph
+        result = query_graph(
+            pattern="tests_for",
+            target="/src/calc.py::combine",
+            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
+
+
 class TestGetDocsSection:
     """Tests for the get_docs_section tool."""