1291 lines
50 KiB
Python
1291 lines
50 KiB
Python
"""Tests for the graph storage and query engine."""
|
|
|
|
import logging
|
|
import sqlite3
|
|
import tempfile
|
|
import time
|
|
from pathlib import Path, PureWindowsPath
|
|
|
|
import pytest
|
|
|
|
import code_review_graph.constants as constants_module
|
|
from code_review_graph.graph import GraphStore
|
|
from code_review_graph.incremental import full_build
|
|
from code_review_graph.parser import EdgeInfo, NodeInfo
|
|
|
|
|
|
class TestGraphStore:
|
|
def setup_method(self):
|
|
self.tmp = tempfile.NamedTemporaryFile(suffix=".db", delete=False)
|
|
self.tmp.close() # release the handle before GraphStore reopens it on Windows
|
|
self.store = GraphStore(self.tmp.name)
|
|
|
|
def teardown_method(self):
|
|
self.store.close()
|
|
Path(self.tmp.name).unlink(missing_ok=True)
|
|
|
|
def _make_file_node(self, path="/test/file.py"):
|
|
return NodeInfo(
|
|
kind="File", name=path, file_path=path,
|
|
line_start=1, line_end=100, language="python",
|
|
)
|
|
|
|
def _make_func_node(self, name="my_func", path="/test/file.py", parent=None, is_test=False):
|
|
return NodeInfo(
|
|
kind="Test" if is_test else "Function",
|
|
name=name, file_path=path,
|
|
line_start=10, line_end=20, language="python",
|
|
parent_name=parent, is_test=is_test,
|
|
)
|
|
|
|
def _make_class_node(self, name="MyClass", path="/test/file.py"):
|
|
return NodeInfo(
|
|
kind="Class", name=name, file_path=path,
|
|
line_start=5, line_end=50, language="python",
|
|
)
|
|
|
|
def test_upsert_and_get_node(self):
|
|
node = self._make_file_node()
|
|
self.store.upsert_node(node)
|
|
self.store.commit()
|
|
|
|
result = self.store.get_node("/test/file.py")
|
|
assert result is not None
|
|
assert result.kind == "File"
|
|
assert result.name == "/test/file.py"
|
|
|
|
def test_upsert_function_node(self):
|
|
func = self._make_func_node()
|
|
self.store.upsert_node(func)
|
|
self.store.commit()
|
|
|
|
result = self.store.get_node("/test/file.py::my_func")
|
|
assert result is not None
|
|
assert result.kind == "Function"
|
|
assert result.name == "my_func"
|
|
|
|
def test_upsert_method_node(self):
|
|
method = self._make_func_node(name="do_thing", parent="MyClass")
|
|
self.store.upsert_node(method)
|
|
self.store.commit()
|
|
|
|
result = self.store.get_node("/test/file.py::MyClass.do_thing")
|
|
assert result is not None
|
|
assert result.parent_name == "MyClass"
|
|
|
|
def test_get_node_bridges_windows_native_qualified_names(self):
|
|
"""A Windows-native path prefix still finds the POSIX-keyed node (#774)."""
|
|
path = "repo/pkg/mod.py"
|
|
self.store.upsert_node(self._make_file_node(path))
|
|
self.store.upsert_node(self._make_func_node("my_func", path))
|
|
self.store.commit()
|
|
|
|
native_prefix = str(PureWindowsPath(path))
|
|
assert native_prefix == "repo\\pkg\\mod.py"
|
|
native_qn = f"{native_prefix}::my_func"
|
|
result = self.store.get_node(native_qn)
|
|
assert result is not None
|
|
assert result.qualified_name == "repo/pkg/mod.py::my_func"
|
|
|
|
file_node = self.store.get_node(native_prefix)
|
|
assert file_node is not None
|
|
assert file_node.qualified_name == "repo/pkg/mod.py"
|
|
|
|
assert self.store.get_node(f"{native_prefix}::missing") is None
|
|
|
|
def test_get_node_bridge_keeps_php_backslashes_in_symbol_part(self):
|
|
"""Only the path component is normalized; PHP FQN symbols keep ``\\``."""
|
|
node = NodeInfo(
|
|
kind="Class", name="App\\Domain\\Job", file_path="src/App.php",
|
|
line_start=1, line_end=10, language="php",
|
|
)
|
|
self.store.upsert_node(node)
|
|
self.store.commit()
|
|
|
|
posix_qn = "src/App.php::App\\Domain\\Job"
|
|
assert self.store.get_node(posix_qn) is not None
|
|
native_qn = "src\\App.php::App\\Domain\\Job"
|
|
bridged = self.store.get_node(native_qn)
|
|
assert bridged is not None
|
|
assert bridged.qualified_name == posix_qn
|
|
|
|
def test_upsert_edge(self):
|
|
edge = EdgeInfo(
|
|
kind="CALLS",
|
|
source="/test/file.py::func_a",
|
|
target="/test/file.py::func_b",
|
|
file_path="/test/file.py",
|
|
line=15,
|
|
)
|
|
self.store.upsert_edge(edge)
|
|
self.store.commit()
|
|
|
|
edges = self.store.get_edges_by_source("/test/file.py::func_a")
|
|
assert len(edges) == 1
|
|
assert edges[0].kind == "CALLS"
|
|
assert edges[0].target_qualified == "/test/file.py::func_b"
|
|
|
|
def test_remove_file_data(self):
|
|
node = self._make_file_node()
|
|
func = self._make_func_node()
|
|
self.store.upsert_node(node)
|
|
self.store.upsert_node(func)
|
|
self.store.commit()
|
|
|
|
self.store.remove_file_data("/test/file.py")
|
|
self.store.commit()
|
|
|
|
assert self.store.get_node("/test/file.py") is None
|
|
assert self.store.get_node("/test/file.py::my_func") is None
|
|
|
|
def test_remove_file_permanently_removes_references_and_same_db_embeddings(self):
|
|
deleted_path = "/test/deleted.py"
|
|
survivor_path = "/test/survivor.py"
|
|
deleted_qn = f"{deleted_path}::removed"
|
|
survivor_qn = f"{survivor_path}::caller"
|
|
self.store.store_file_nodes_edges(
|
|
deleted_path,
|
|
[
|
|
self._make_file_node(deleted_path),
|
|
self._make_func_node("removed", deleted_path),
|
|
],
|
|
[],
|
|
)
|
|
self.store.store_file_nodes_edges(
|
|
survivor_path,
|
|
[
|
|
self._make_file_node(survivor_path),
|
|
self._make_func_node("caller", survivor_path),
|
|
],
|
|
[
|
|
EdgeInfo(
|
|
kind="CALLS",
|
|
source=survivor_qn,
|
|
target=deleted_qn,
|
|
file_path=survivor_path,
|
|
),
|
|
],
|
|
)
|
|
self.store._conn.execute(
|
|
"CREATE TABLE embeddings ("
|
|
"qualified_name TEXT PRIMARY KEY, vector BLOB NOT NULL, "
|
|
"text_hash TEXT NOT NULL, provider TEXT NOT NULL)"
|
|
)
|
|
self.store._conn.executemany(
|
|
"INSERT INTO embeddings VALUES (?, ?, ?, ?)",
|
|
[
|
|
(deleted_qn, b"deleted", "deleted", "test"),
|
|
(survivor_qn, b"survivor", "survivor", "test"),
|
|
("unrelated::orphan", b"orphan", "orphan", "test"),
|
|
],
|
|
)
|
|
self.store.commit()
|
|
|
|
self.store.remove_file_permanently(deleted_path)
|
|
self.store.commit()
|
|
|
|
assert self.store.get_nodes_by_file(deleted_path) == []
|
|
assert self.store.get_node(survivor_qn) is not None
|
|
assert self.store.get_edges_by_source(survivor_qn) == []
|
|
embeddings = self.store._conn.execute(
|
|
"SELECT qualified_name FROM embeddings ORDER BY qualified_name"
|
|
).fetchall()
|
|
assert [row["qualified_name"] for row in embeddings] == [
|
|
survivor_qn,
|
|
"unrelated::orphan",
|
|
]
|
|
|
|
def test_remove_file_permanently_handles_more_than_sqlite_variable_limit(self):
|
|
deleted_path = "/test/large.py"
|
|
rows = [
|
|
(
|
|
"Function",
|
|
f"node_{index}",
|
|
f"{deleted_path}::node_{index}",
|
|
deleted_path,
|
|
index + 1,
|
|
index + 1,
|
|
"python",
|
|
0,
|
|
0.0,
|
|
)
|
|
for index in range(16_384)
|
|
]
|
|
self.store._conn.executemany(
|
|
"INSERT INTO nodes "
|
|
"(kind, name, qualified_name, file_path, line_start, line_end, language, "
|
|
"is_test, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)",
|
|
rows,
|
|
)
|
|
self.store.commit()
|
|
|
|
changed = self.store.remove_file_permanently(deleted_path)
|
|
|
|
assert changed == 1
|
|
assert self.store.get_nodes_by_file(deleted_path) == []
|
|
|
|
def test_remove_files_permanently_rolls_back_every_table_on_failure(self):
|
|
deleted_path = "/test/deleted.py"
|
|
survivor_path = "/test/survivor.py"
|
|
deleted_qn = f"{deleted_path}::removed"
|
|
survivor_qn = f"{survivor_path}::caller"
|
|
self.store.store_file_nodes_edges(
|
|
deleted_path,
|
|
[self._make_file_node(deleted_path), self._make_func_node("removed", deleted_path)],
|
|
[
|
|
EdgeInfo(
|
|
kind="CONTAINS",
|
|
source=deleted_path,
|
|
target=deleted_qn,
|
|
file_path=deleted_path,
|
|
)
|
|
],
|
|
)
|
|
self.store.store_file_nodes_edges(
|
|
survivor_path,
|
|
[self._make_file_node(survivor_path), self._make_func_node("caller", survivor_path)],
|
|
[
|
|
EdgeInfo(
|
|
kind="CALLS",
|
|
source=survivor_qn,
|
|
target=deleted_qn,
|
|
file_path=survivor_path,
|
|
)
|
|
],
|
|
)
|
|
self.store._conn.execute(
|
|
"CREATE TABLE embeddings (qualified_name TEXT PRIMARY KEY, vector BLOB NOT NULL, "
|
|
"text_hash TEXT NOT NULL, provider TEXT NOT NULL)"
|
|
)
|
|
self.store._conn.execute(
|
|
"INSERT INTO embeddings VALUES (?, ?, ?, ?)",
|
|
(deleted_qn, b"deleted", "deleted", "test"),
|
|
)
|
|
self.store.commit()
|
|
before = {
|
|
"nodes": self.store._conn.execute("SELECT COUNT(*) FROM nodes").fetchone()[0],
|
|
"edges": self.store._conn.execute("SELECT COUNT(*) FROM edges").fetchone()[0],
|
|
"embeddings": self.store._conn.execute("SELECT COUNT(*) FROM embeddings").fetchone()[0],
|
|
}
|
|
|
|
self.store._conn.execute(
|
|
"CREATE TRIGGER fail_deleted_node BEFORE DELETE ON nodes "
|
|
f"WHEN OLD.file_path = '{deleted_path}' "
|
|
"BEGIN SELECT RAISE(ABORT, 'injected deletion failure'); END"
|
|
)
|
|
self.store.commit()
|
|
|
|
with pytest.raises(sqlite3.IntegrityError, match="injected deletion failure"):
|
|
self.store.remove_files_permanently([deleted_path])
|
|
|
|
after = {
|
|
"nodes": self.store._conn.execute("SELECT COUNT(*) FROM nodes").fetchone()[0],
|
|
"edges": self.store._conn.execute("SELECT COUNT(*) FROM edges").fetchone()[0],
|
|
"embeddings": self.store._conn.execute("SELECT COUNT(*) FROM embeddings").fetchone()[0],
|
|
}
|
|
assert after == before
|
|
|
|
def test_remove_files_permanently_counts_changed_paths_and_commits_once(self):
|
|
paths = ["/test/first.py", "/test/second.py", "/test/missing.py"]
|
|
for path in paths[:2]:
|
|
self.store.store_file_nodes_edges(path, [self._make_file_node(path)], [])
|
|
|
|
commits = 0
|
|
|
|
def count_commits() -> int:
|
|
nonlocal commits
|
|
commits += 1
|
|
return 0
|
|
|
|
self.store._conn.set_trace_callback(
|
|
lambda statement: count_commits() if statement == "COMMIT" else None
|
|
)
|
|
changed = self.store.remove_files_permanently(paths)
|
|
|
|
assert changed == 2
|
|
assert commits == 1
|
|
|
|
def test_replacement_preserves_incoming_edges_from_other_files(self):
|
|
target_path = "/test/target.py"
|
|
caller_path = "/test/caller.py"
|
|
target_qn = f"{target_path}::target"
|
|
caller_qn = f"{caller_path}::caller"
|
|
self.store.store_file_nodes_edges(
|
|
target_path,
|
|
[
|
|
self._make_file_node(target_path),
|
|
self._make_func_node("target", target_path),
|
|
],
|
|
[],
|
|
)
|
|
self.store.store_file_nodes_edges(
|
|
caller_path,
|
|
[
|
|
self._make_file_node(caller_path),
|
|
self._make_func_node("caller", caller_path),
|
|
],
|
|
[
|
|
EdgeInfo(
|
|
kind="CALLS",
|
|
source=caller_qn,
|
|
target=target_qn,
|
|
file_path=caller_path,
|
|
),
|
|
],
|
|
)
|
|
|
|
self.store.store_file_nodes_edges(
|
|
target_path,
|
|
[
|
|
self._make_file_node(target_path),
|
|
self._make_func_node("target", target_path),
|
|
],
|
|
[],
|
|
)
|
|
|
|
incoming = self.store.get_edges_by_target(target_qn)
|
|
assert [(edge.source_qualified, edge.file_path) for edge in incoming] == [
|
|
(caller_qn, caller_path),
|
|
]
|
|
|
|
def test_store_file_nodes_edges(self):
|
|
nodes = [self._make_file_node(), self._make_func_node()]
|
|
edges = [
|
|
EdgeInfo(
|
|
kind="CONTAINS", source="/test/file.py",
|
|
target="/test/file.py::my_func", file_path="/test/file.py",
|
|
)
|
|
]
|
|
self.store.store_file_nodes_edges("/test/file.py", nodes, edges)
|
|
|
|
result = self.store.get_nodes_by_file("/test/file.py")
|
|
assert len(result) == 2
|
|
|
|
def test_store_after_remove_no_transaction_error(self):
|
|
"""Regression test for #135: store_file_nodes_edges after
|
|
remove_file_data must not raise 'cannot start a transaction
|
|
within a transaction'.
|
|
"""
|
|
# Seed initial data for two files
|
|
nodes_a = [self._make_file_node("/test/a.py")]
|
|
nodes_b = [self._make_file_node("/test/b.py")]
|
|
self.store.store_file_nodes_edges("/test/a.py", nodes_a, [])
|
|
self.store.store_file_nodes_edges("/test/b.py", nodes_b, [])
|
|
|
|
# Without the isolation_level=None fix, this would leave an
|
|
# implicit transaction open and the next call would crash.
|
|
self.store.remove_file_data("/test/a.py")
|
|
# Must not raise sqlite3.OperationalError
|
|
nodes_c = [self._make_file_node("/test/c.py")]
|
|
self.store.store_file_nodes_edges("/test/c.py", nodes_c, [])
|
|
|
|
assert self.store.get_node("/test/a.py") is None
|
|
assert self.store.get_node("/test/c.py") is not None
|
|
|
|
def test_store_after_multiple_removes_no_transaction_error(self):
|
|
"""Regression test for #181: full_build stale-file purge leaves
|
|
implicit transaction open after multiple remove_file_data calls.
|
|
"""
|
|
# Seed data for several files
|
|
for i in range(5):
|
|
path = f"/test/file_{i}.py"
|
|
self.store.store_file_nodes_edges(
|
|
path, [self._make_file_node(path)], [],
|
|
)
|
|
|
|
# Simulates full_build's stale-file purge: multiple deletes in a
|
|
# row without explicit commit between them.
|
|
for i in range(3):
|
|
self.store.remove_file_data(f"/test/file_{i}.py")
|
|
|
|
# Next store call must succeed regardless of prior connection state.
|
|
new_path = "/test/new_file.py"
|
|
nodes = [self._make_file_node(new_path)]
|
|
self.store.store_file_nodes_edges(new_path, nodes, [])
|
|
|
|
assert self.store.get_node(new_path) is not None
|
|
assert self.store.get_node("/test/file_0.py") is None
|
|
|
|
def test_store_with_open_transaction_no_error(self):
|
|
"""Regression test for #489: store_file_nodes_edges and
|
|
store_file_batch must not raise 'cannot start a transaction
|
|
within a transaction' when the caller has an explicit BEGIN open.
|
|
"""
|
|
node_a = self._make_file_node("/test/a.py")
|
|
node_b = self._make_file_node("/test/b.py")
|
|
|
|
# Force an open transaction on the shared connection.
|
|
self.store._conn.execute("BEGIN")
|
|
assert self.store._conn.in_transaction
|
|
|
|
# Must not raise sqlite3.OperationalError.
|
|
self.store.store_file_nodes_edges("/test/a.py", [node_a], [])
|
|
assert self.store.get_node("/test/a.py") is not None
|
|
|
|
# Re-open the transaction and verify the batch path is guarded too.
|
|
self.store._conn.execute("BEGIN")
|
|
assert self.store._conn.in_transaction
|
|
self.store.store_file_batch([("/test/b.py", [node_b], [], "")])
|
|
assert self.store.get_node("/test/b.py") is not None
|
|
|
|
def test_search_nodes(self):
|
|
self.store.upsert_node(self._make_func_node("authenticate"))
|
|
self.store.upsert_node(self._make_func_node("authorize"))
|
|
self.store.upsert_node(self._make_func_node("process"))
|
|
self.store.commit()
|
|
|
|
results = self.store.search_nodes("auth")
|
|
names = {r.name for r in results}
|
|
assert "authenticate" in names
|
|
assert "authorize" in names
|
|
assert "process" not in names
|
|
|
|
def test_get_stats(self):
|
|
self.store.upsert_node(self._make_file_node())
|
|
self.store.upsert_node(self._make_func_node())
|
|
self.store.upsert_node(self._make_class_node())
|
|
self.store.upsert_edge(EdgeInfo(
|
|
kind="CONTAINS", source="/test/file.py",
|
|
target="/test/file.py::my_func", file_path="/test/file.py",
|
|
))
|
|
self.store.commit()
|
|
|
|
stats = self.store.get_stats()
|
|
assert stats.total_nodes == 3
|
|
assert stats.total_edges == 1
|
|
assert stats.nodes_by_kind["File"] == 1
|
|
assert stats.nodes_by_kind["Function"] == 1
|
|
assert stats.nodes_by_kind["Class"] == 1
|
|
assert "python" in stats.languages
|
|
|
|
def test_has_nodes(self):
|
|
assert self.store.has_nodes() is False
|
|
|
|
self.store.upsert_node(self._make_file_node())
|
|
|
|
assert self.store.has_nodes() is True
|
|
|
|
def test_impact_radius(self):
|
|
# func_b depends on the changed func_a, so func_b is impacted.
|
|
self.store.upsert_node(self._make_file_node("/a.py"))
|
|
self.store.upsert_node(self._make_func_node("func_a", "/a.py"))
|
|
self.store.upsert_node(self._make_file_node("/b.py"))
|
|
self.store.upsert_node(self._make_func_node("func_b", "/b.py"))
|
|
self.store.upsert_edge(EdgeInfo(
|
|
kind="CALLS", source="/b.py::func_b",
|
|
target="/a.py::func_a", file_path="/b.py", line=10,
|
|
))
|
|
self.store.commit()
|
|
|
|
result = self.store.get_impact_radius(["/a.py"], max_depth=2)
|
|
assert len(result["changed_nodes"]) > 0
|
|
# func_b in /b.py should be impacted
|
|
impacted_qns = {n.qualified_name for n in result["impacted_nodes"]}
|
|
assert "/b.py::func_b" in impacted_qns or "/b.py" in impacted_qns
|
|
|
|
def test_upsert_edge_preserves_multiple_call_sites(self):
|
|
"""Multiple CALLS edges to the same target from the same source on different lines."""
|
|
edge1 = EdgeInfo(
|
|
kind="CALLS", source="/test/file.py::caller",
|
|
target="/test/file.py::helper", file_path="/test/file.py", line=10,
|
|
)
|
|
edge2 = EdgeInfo(
|
|
kind="CALLS", source="/test/file.py::caller",
|
|
target="/test/file.py::helper", file_path="/test/file.py", line=20,
|
|
)
|
|
self.store.upsert_edge(edge1)
|
|
self.store.upsert_edge(edge2)
|
|
self.store.commit()
|
|
|
|
edges = self.store.get_edges_by_source("/test/file.py::caller")
|
|
assert len(edges) == 2
|
|
lines = {e.line for e in edges}
|
|
assert lines == {10, 20}
|
|
|
|
def test_metadata(self):
|
|
self.store.set_metadata("test_key", "test_value")
|
|
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_parse_store_get_transitive_tests_end_to_end(self):
|
|
"""End-to-end producer->store->consumer guard for #515.
|
|
|
|
Parse a real fixture pair (production + test) through the parser,
|
|
persist the emitted nodes/edges, and confirm get_transitive_tests
|
|
surfaces the test as covering the production code. This couples the
|
|
parser's canonical TESTED_BY direction (source=production,
|
|
target=test) to the consumer query, so a future parser flip would
|
|
break this test even if every hand-seeded fixture test still passed.
|
|
"""
|
|
from code_review_graph.parser import CodeParser
|
|
|
|
fixtures = Path(__file__).parent / "fixtures"
|
|
parser = CodeParser()
|
|
all_nodes: list[NodeInfo] = []
|
|
all_edges: list[EdgeInfo] = []
|
|
for fixture in ("sample_python.py", "test_sample.py"):
|
|
nodes, edges = parser.parse_file(fixtures / fixture)
|
|
all_nodes.extend(nodes)
|
|
all_edges.extend(edges)
|
|
|
|
for n in all_nodes:
|
|
self.store.upsert_node(n)
|
|
for e in all_edges:
|
|
self.store.upsert_edge(e)
|
|
self.store.commit()
|
|
|
|
tested_by = [e for e in all_edges if e.kind == "TESTED_BY"]
|
|
assert tested_by, "fixture pair should yield at least one TESTED_BY edge"
|
|
|
|
# Producer direction guard: every TESTED_BY target must be a stored
|
|
# Test node, and querying the consumer (get_transitive_tests) by the
|
|
# edge's *source* (production) must surface that test target. If a
|
|
# future parser flip swapped the direction, the target would point at
|
|
# production code and this end-to-end assertion would fail.
|
|
checked = 0
|
|
for edge in tested_by:
|
|
target = self.store.get_node(edge.target)
|
|
assert target is not None, f"missing test node {edge.target}"
|
|
assert target.is_test, (
|
|
f"TESTED_BY target {edge.target!r} should be a test node; "
|
|
f"a flipped parser would put production code here"
|
|
)
|
|
|
|
results = self.store.get_transitive_tests(edge.source)
|
|
qns = {r["qualified_name"] for r in results}
|
|
assert edge.target in qns, (
|
|
f"get_transitive_tests({edge.source!r}) should surface test "
|
|
f"{edge.target!r}; got {sorted(qns)}"
|
|
)
|
|
checked += 1
|
|
assert checked >= 1
|
|
|
|
def test_get_all_community_ids_logs_when_column_missing(self, caplog):
|
|
conn = sqlite3.connect(":memory:")
|
|
conn.row_factory = sqlite3.Row
|
|
conn.execute(
|
|
"CREATE TABLE nodes (qualified_name TEXT PRIMARY KEY)"
|
|
)
|
|
store = GraphStore.__new__(GraphStore)
|
|
store._conn = conn
|
|
|
|
with caplog.at_level(logging.DEBUG, logger="code_review_graph.graph"):
|
|
result = store.get_all_community_ids()
|
|
|
|
assert result == {}
|
|
assert "Community IDs unavailable" in caplog.text
|
|
conn.close()
|
|
|
|
def test_get_communities_list_logs_when_table_missing(self, caplog):
|
|
conn = sqlite3.connect(":memory:")
|
|
conn.row_factory = sqlite3.Row
|
|
store = GraphStore.__new__(GraphStore)
|
|
store._conn = conn
|
|
|
|
with caplog.at_level(logging.DEBUG, logger="code_review_graph.graph"):
|
|
result = store.get_communities_list()
|
|
|
|
assert result == []
|
|
assert "Communities list unavailable" in caplog.text
|
|
conn.close()
|
|
|
|
|
|
class TestImpactRadiusSql:
|
|
"""Tests for get_impact_radius_sql vs NetworkX BFS."""
|
|
|
|
def setup_method(self):
|
|
self.tmp = tempfile.NamedTemporaryFile(suffix=".db", delete=False)
|
|
self.tmp.close() # release the handle before GraphStore reopens it on Windows
|
|
self.store = GraphStore(self.tmp.name)
|
|
self._build_chain()
|
|
|
|
def teardown_method(self):
|
|
self.store.close()
|
|
Path(self.tmp.name).unlink(missing_ok=True)
|
|
|
|
def _build_chain(self):
|
|
"""Build D -> C -> B -> A dependency chain for testing."""
|
|
for name, path in [
|
|
("func_a", "/a.py"), ("func_b", "/b.py"),
|
|
("func_c", "/c.py"), ("func_d", "/d.py"),
|
|
]:
|
|
self.store.upsert_node(NodeInfo(
|
|
kind="File", name=path, file_path=path,
|
|
line_start=1, line_end=50, language="python",
|
|
))
|
|
self.store.upsert_node(NodeInfo(
|
|
kind="Function", name=name, file_path=path,
|
|
line_start=5, line_end=20, language="python",
|
|
))
|
|
self.store.upsert_edge(EdgeInfo(
|
|
kind="CALLS", source="/b.py::func_b",
|
|
target="/a.py::func_a", file_path="/b.py", line=10,
|
|
))
|
|
self.store.upsert_edge(EdgeInfo(
|
|
kind="CALLS", source="/c.py::func_c",
|
|
target="/b.py::func_b", file_path="/c.py", line=10,
|
|
))
|
|
self.store.upsert_edge(EdgeInfo(
|
|
kind="CALLS", source="/d.py::func_d",
|
|
target="/c.py::func_c", file_path="/d.py", line=10,
|
|
))
|
|
self.store.commit()
|
|
|
|
def test_sql_matches_networkx(self):
|
|
"""SQL and NetworkX BFS produce identical impacted node sets."""
|
|
sql_result = self.store.get_impact_radius_sql(["/a.py"], max_depth=2)
|
|
nx_result = self.store._get_impact_radius_networkx(["/a.py"], max_depth=2)
|
|
|
|
sql_qns = {n.qualified_name for n in sql_result["impacted_nodes"]}
|
|
nx_qns = {n.qualified_name for n in nx_result["impacted_nodes"]}
|
|
assert sql_qns == {"/b.py::func_b", "/c.py::func_c"}
|
|
assert sql_qns == nx_qns
|
|
|
|
def test_max_nodes_truncation(self):
|
|
"""Setting max_nodes=2 should truncate results."""
|
|
result = self.store.get_impact_radius_sql(
|
|
["/a.py"], max_depth=3, max_nodes=2,
|
|
)
|
|
assert result["truncated"] is True
|
|
assert result["total_impacted"] == 3
|
|
assert len(result["impacted_nodes"]) == 2
|
|
|
|
def test_empty_changed_files(self):
|
|
result = self.store.get_impact_radius_sql([], max_depth=2)
|
|
assert result["changed_nodes"] == []
|
|
assert result["impacted_nodes"] == []
|
|
assert result["total_impacted"] == 0
|
|
|
|
|
|
def test_impact_radius_real_build_includes_importer_not_imported_dependency(
|
|
tmp_path: Path,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
"""A real parsed import graph follows impact toward dependents only."""
|
|
monkeypatch.setenv("CRG_SERIAL_PARSE", "1")
|
|
dependency = tmp_path / "dependency.py"
|
|
changed = tmp_path / "changed.py"
|
|
importer = tmp_path / "importer.py"
|
|
dependency.write_text("VALUE = 1\n", encoding="utf-8")
|
|
changed.write_text(
|
|
"from dependency import VALUE\n\n"
|
|
"def changed_value():\n"
|
|
" return VALUE\n",
|
|
encoding="utf-8",
|
|
)
|
|
importer.write_text(
|
|
"from changed import changed_value\n\n"
|
|
"def consume():\n"
|
|
" return changed_value()\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
with GraphStore(tmp_path / "graph.db") as store:
|
|
built = full_build(tmp_path, store)
|
|
assert built["errors"] == []
|
|
|
|
sql = store.get_impact_radius_sql([str(changed)], max_depth=1)
|
|
networkx = store._get_impact_radius_networkx(
|
|
[str(changed)],
|
|
max_depth=1,
|
|
)
|
|
|
|
expected = {importer.as_posix()}
|
|
assert set(sql["impacted_files"]) == expected
|
|
assert set(networkx["impacted_files"]) == expected
|
|
assert dependency.as_posix() not in sql["impacted_files"]
|
|
assert sql["impact_scores"] == networkx["impact_scores"]
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("raw", "expected"),
|
|
[
|
|
("0.75", 0.75),
|
|
("", 0.6),
|
|
("not-a-number", 0.6),
|
|
("nan", 0.6),
|
|
("inf", 0.6),
|
|
("-0.1", 0.6),
|
|
("0", 0.6),
|
|
("1", 0.6),
|
|
("1.2", 0.6),
|
|
],
|
|
)
|
|
def test_impact_float_configuration_is_finite_and_bounded(
|
|
monkeypatch, raw, expected,
|
|
):
|
|
monkeypatch.setenv("CRG_TEST_IMPACT_FLOAT", raw)
|
|
assert constants_module._bounded_float_env(
|
|
"CRG_TEST_IMPACT_FLOAT", 0.6, lower=0.0, upper=1.0,
|
|
) == pytest.approx(expected)
|
|
|
|
|
|
class TestWeightedImpactScoring:
|
|
"""Best-path scoring stays ranked, bounded, and engine-independent."""
|
|
|
|
def setup_method(self):
|
|
self.tmp = tempfile.NamedTemporaryFile(suffix=".db", delete=False)
|
|
self.tmp.close()
|
|
self.store = GraphStore(self.tmp.name)
|
|
|
|
def teardown_method(self):
|
|
self.store.close()
|
|
Path(self.tmp.name).unlink(missing_ok=True)
|
|
|
|
def _add_func(self, name: str, path: str) -> str:
|
|
self.store.upsert_node(NodeInfo(
|
|
kind="Function", name=name, file_path=path,
|
|
line_start=1, line_end=10, language="python",
|
|
))
|
|
return f"{path}::{name}"
|
|
|
|
def _add_edge(
|
|
self, kind: str, source: str, target: str, line: int = 1,
|
|
) -> None:
|
|
self.store.upsert_edge(EdgeInfo(
|
|
kind=kind, source=source, target=target,
|
|
file_path="/seed.py", line=line,
|
|
))
|
|
|
|
@staticmethod
|
|
def _ordered_qns(result) -> list[str]:
|
|
return [node.qualified_name for node in result["impacted_nodes"]]
|
|
|
|
@pytest.mark.parametrize(
|
|
"kind",
|
|
[
|
|
"CALLS",
|
|
"IMPORTS_FROM",
|
|
"DEPENDS_ON",
|
|
"REFERENCES",
|
|
"INHERITS",
|
|
"OVERRIDES",
|
|
"IMPLEMENTS",
|
|
],
|
|
)
|
|
def test_dependency_edges_include_dependents_not_dependencies(self, kind):
|
|
seed = self._add_func("seed", "/seed.py")
|
|
dependent = self._add_func("dependent", "/dependent.py")
|
|
dependency = self._add_func("dependency", "/dependency.py")
|
|
self._add_edge(kind, dependent, seed)
|
|
self._add_edge(kind, seed, dependency, line=2)
|
|
self.store.commit()
|
|
|
|
sql = self.store.get_impact_radius_sql(["/seed.py"], max_depth=1)
|
|
nx_result = self.store._get_impact_radius_networkx(
|
|
["/seed.py"], max_depth=1,
|
|
)
|
|
|
|
assert self._ordered_qns(sql) == [dependent]
|
|
assert self._ordered_qns(nx_result) == [dependent]
|
|
assert sql["impact_scores"] == nx_result["impact_scores"]
|
|
|
|
def test_tested_by_traverses_from_production_to_test_only(self):
|
|
seed = self._add_func("seed", "/seed.py")
|
|
test = self._add_func("test_seed", "/test_seed.py")
|
|
unrelated_production = self._add_func(
|
|
"unrelated_production", "/unrelated.py",
|
|
)
|
|
self._add_edge("TESTED_BY", seed, test)
|
|
self._add_edge("TESTED_BY", unrelated_production, seed, line=2)
|
|
self.store.commit()
|
|
|
|
sql = self.store.get_impact_radius_sql(["/seed.py"], max_depth=1)
|
|
nx_result = self.store._get_impact_radius_networkx(
|
|
["/seed.py"], max_depth=1,
|
|
)
|
|
|
|
assert self._ordered_qns(sql) == [test]
|
|
assert self._ordered_qns(nx_result) == [test]
|
|
assert sql["impact_scores"] == nx_result["impact_scores"]
|
|
|
|
def test_contains_edge_cannot_bridge_impact(self):
|
|
seed = self._add_func("seed", "/seed.py")
|
|
stale_container = "stale.py::Container"
|
|
dependent = self._add_func("dependent", "/dependent.py")
|
|
self._add_edge("CONTAINS", stale_container, seed)
|
|
self._add_edge("CALLS", dependent, stale_container, line=2)
|
|
self.store.commit()
|
|
|
|
sql = self.store.get_impact_radius_sql(["/seed.py"], max_depth=2)
|
|
nx_result = self.store._get_impact_radius_networkx(
|
|
["/seed.py"], max_depth=2,
|
|
)
|
|
|
|
assert self._ordered_qns(sql) == []
|
|
assert self._ordered_qns(nx_result) == []
|
|
assert sql["impact_scores"] == nx_result["impact_scores"]
|
|
|
|
def test_unknown_edge_kind_defaults_to_incoming_dependency_direction(self):
|
|
seed = self._add_func("seed", "/seed.py")
|
|
dependent = self._add_func("dependent", "/dependent.py")
|
|
dependency = self._add_func("dependency", "/dependency.py")
|
|
self._add_edge("UNKNOWN_KIND", dependent, seed)
|
|
self._add_edge("UNKNOWN_KIND", seed, dependency, line=2)
|
|
self.store.commit()
|
|
|
|
sql = self.store.get_impact_radius_sql(["/seed.py"], max_depth=1)
|
|
nx_result = self.store._get_impact_radius_networkx(
|
|
["/seed.py"], max_depth=1,
|
|
)
|
|
|
|
assert self._ordered_qns(sql) == [dependent]
|
|
assert sql["impact_scores"][dependent] == pytest.approx(0.3)
|
|
assert self._ordered_qns(nx_result) == [dependent]
|
|
assert sql["impact_scores"] == nx_result["impact_scores"]
|
|
|
|
def test_edge_weights_rank_best_path_and_engines_match(self):
|
|
seed = self._add_func("seed", "/seed.py")
|
|
caller = self._add_func("caller", "/caller.py")
|
|
importer = self._add_func("importer", "/importer.py")
|
|
indirect_caller = self._add_func(
|
|
"indirect_caller", "/indirect_caller.py",
|
|
)
|
|
self._add_edge("CALLS", caller, seed)
|
|
self._add_edge("IMPORTS_FROM", importer, seed)
|
|
self._add_edge("CALLS", indirect_caller, caller)
|
|
self.store.commit()
|
|
|
|
sql = self.store.get_impact_radius_sql(["/seed.py"], max_depth=2)
|
|
nx_result = self.store._get_impact_radius_networkx(
|
|
["/seed.py"], max_depth=2,
|
|
)
|
|
|
|
assert sql["impact_scores"][caller] == pytest.approx(0.6)
|
|
assert sql["impact_scores"][indirect_caller] == pytest.approx(0.36)
|
|
assert sql["impact_scores"][importer] == pytest.approx(0.3)
|
|
assert self._ordered_qns(sql) == [
|
|
caller, indirect_caller, importer,
|
|
]
|
|
assert sql["impact_scores"] == nx_result["impact_scores"]
|
|
assert self._ordered_qns(sql) == self._ordered_qns(nx_result)
|
|
|
|
def test_deeper_strong_path_beats_shallow_weak_path(self):
|
|
seed = self._add_func("seed", "/seed.py")
|
|
middle = self._add_func("middle", "/middle.py")
|
|
target = self._add_func("target", "/target.py")
|
|
self._add_edge("IMPORTS_FROM", target, seed)
|
|
self._add_edge("CALLS", middle, seed, line=2)
|
|
self._add_edge("CALLS", target, middle, line=3)
|
|
self.store.commit()
|
|
|
|
sql = self.store.get_impact_radius_sql(["/seed.py"], max_depth=2)
|
|
nx_result = self.store._get_impact_radius_networkx(
|
|
["/seed.py"], max_depth=2,
|
|
)
|
|
|
|
assert sql["impact_scores"][target] == pytest.approx(0.36)
|
|
assert sql["impact_scores"] == nx_result["impact_scores"]
|
|
|
|
def test_score_floor_stops_expansion_in_both_engines(self):
|
|
qns = [
|
|
self._add_func(f"node_{index}", f"/node_{index}.py")
|
|
for index in range(8)
|
|
]
|
|
for index, (source, target) in enumerate(zip(qns[1:], qns)):
|
|
self._add_edge("CALLS", source, target, line=index + 1)
|
|
self.store.commit()
|
|
|
|
sql = self.store.get_impact_radius_sql(
|
|
["/node_0.py"], max_depth=8,
|
|
)
|
|
nx_result = self.store._get_impact_radius_networkx(
|
|
["/node_0.py"], max_depth=8,
|
|
)
|
|
|
|
assert qns[5] in sql["impact_scores"]
|
|
assert qns[6] not in sql["impact_scores"]
|
|
assert sql["impact_scores"] == nx_result["impact_scores"]
|
|
|
|
def test_unknown_edge_kind_uses_default_weight(self):
|
|
seed = self._add_func("seed", "/seed.py")
|
|
target = self._add_func("target", "/target.py")
|
|
self._add_edge("UNKNOWN_KIND", target, seed)
|
|
self.store.commit()
|
|
|
|
sql = self.store.get_impact_radius_sql(["/seed.py"], max_depth=1)
|
|
nx_result = self.store._get_impact_radius_networkx(
|
|
["/seed.py"], max_depth=1,
|
|
)
|
|
|
|
assert sql["impact_scores"][target] == pytest.approx(0.3)
|
|
assert sql["impact_scores"] == nx_result["impact_scores"]
|
|
|
|
def test_truncation_is_exact_at_boundary_and_uses_sentinel(self):
|
|
seed = self._add_func("seed", "/seed.py")
|
|
targets = [
|
|
self._add_func(f"target_{index}", f"/target_{index}.py")
|
|
for index in range(3)
|
|
]
|
|
for index, target in enumerate(targets):
|
|
self._add_edge("CALLS", target, seed, line=index + 1)
|
|
self.store.commit()
|
|
|
|
exact = self.store.get_impact_radius_sql(
|
|
["/seed.py"], max_depth=1, max_nodes=3,
|
|
)
|
|
capped = self.store.get_impact_radius_sql(
|
|
["/seed.py"], max_depth=1, max_nodes=2,
|
|
)
|
|
|
|
assert exact["truncated"] is False
|
|
assert exact["total_impacted"] == 3
|
|
assert capped["truncated"] is True
|
|
assert capped["total_impacted"] == 3
|
|
assert len(capped["impacted_nodes"]) == 2
|
|
|
|
def test_ghost_endpoint_bridges_without_consuming_limit(self):
|
|
seed = self._add_func("seed", "/seed.py")
|
|
target = self._add_func("target", "/target.py")
|
|
ghost = "external.package::ghost"
|
|
self._add_edge("CALLS", ghost, seed)
|
|
self._add_edge("CALLS", target, ghost, line=2)
|
|
self.store.commit()
|
|
|
|
result = self.store.get_impact_radius_sql(
|
|
["/seed.py"], max_depth=2, max_nodes=1,
|
|
)
|
|
|
|
assert self._ordered_qns(result) == [target]
|
|
assert ghost not in result["impact_scores"]
|
|
assert result["truncated"] is False
|
|
|
|
def test_parallel_edges_use_strongest_weight_in_both_engines(self):
|
|
seed = self._add_func("seed", "/seed.py")
|
|
target = self._add_func("target", "/target.py")
|
|
self._add_edge("CALLS", target, seed, line=1)
|
|
self._add_edge("IMPORTS_FROM", target, seed, line=2)
|
|
self.store.commit()
|
|
|
|
sql = self.store.get_impact_radius_sql(["/seed.py"], max_depth=1)
|
|
nx_result = self.store._get_impact_radius_networkx(
|
|
["/seed.py"], max_depth=1,
|
|
)
|
|
assert sql["impact_scores"][target] == pytest.approx(0.6)
|
|
assert sql["impact_scores"] == nx_result["impact_scores"]
|
|
|
|
def test_parallel_edges_preserve_each_direction_in_both_engines(self):
|
|
source = self._add_func("source", "/source.py")
|
|
target = self._add_func("target", "/target.py")
|
|
self._add_edge("CALLS", source, target, line=1)
|
|
self._add_edge("TESTED_BY", source, target, line=2)
|
|
self.store.commit()
|
|
|
|
for path, expected_qn, expected_score in (
|
|
("/source.py", target, 0.42),
|
|
("/target.py", source, 0.6),
|
|
):
|
|
sql = self.store.get_impact_radius_sql([path], max_depth=1)
|
|
nx_result = self.store._get_impact_radius_networkx(
|
|
[path], max_depth=1,
|
|
)
|
|
|
|
assert self._ordered_qns(sql) == [expected_qn]
|
|
assert sql["impact_scores"][expected_qn] == pytest.approx(
|
|
expected_score,
|
|
)
|
|
assert sql["impact_scores"] == nx_result["impact_scores"]
|
|
|
|
def test_dense_mixed_cycle_is_bounded(self):
|
|
qns = [self._add_func(f"node_{i}", f"/node_{i}.py") for i in range(12)]
|
|
line = 1
|
|
for source_index, source in enumerate(qns):
|
|
for target_index, target in enumerate(qns):
|
|
if source_index == target_index:
|
|
continue
|
|
kind = "CALLS" if (source_index + target_index) % 2 else "IMPORTS_FROM"
|
|
self._add_edge(kind, source, target, line=line)
|
|
line += 1
|
|
self.store.commit()
|
|
|
|
started = time.monotonic()
|
|
result = self.store.get_impact_radius_sql(
|
|
["/node_0.py"], max_depth=25, max_nodes=20,
|
|
)
|
|
elapsed = time.monotonic() - started
|
|
|
|
assert len(result["impacted_nodes"]) == 11
|
|
assert result["truncated"] is False
|
|
assert elapsed < 5.0
|
|
|
|
|
|
class TestGetTransitiveTestsFrontierCap:
|
|
"""Regression tests for O(N*M) query explosion in get_transitive_tests."""
|
|
|
|
def setup_method(self):
|
|
self.tmp = tempfile.NamedTemporaryFile(suffix=".db", delete=False)
|
|
self.tmp.close() # release the handle before GraphStore reopens it on Windows
|
|
self.store = GraphStore(self.tmp.name)
|
|
|
|
def teardown_method(self):
|
|
self.store.close()
|
|
Path(self.tmp.name).unlink(missing_ok=True)
|
|
|
|
def _add_func(self, name: str, path: str) -> str:
|
|
node = NodeInfo(
|
|
kind="Function", name=name, file_path=path,
|
|
line_start=1, line_end=5, language="python",
|
|
)
|
|
self.store.upsert_node(node)
|
|
return f"{path}::{name}"
|
|
|
|
def _add_calls_edge(self, source_qn: str, target_qn: str) -> None:
|
|
self.store.upsert_edge(EdgeInfo(
|
|
kind="CALLS", source=source_qn, target=target_qn,
|
|
file_path=source_qn.split("::")[0], line=1,
|
|
))
|
|
|
|
def test_frontier_capped_limits_sql_queries(self):
|
|
"""Hub function with 200 callees must not issue 200 TESTED_BY queries."""
|
|
hub_qn = self._add_func("hub", "/t/hub.py")
|
|
for i in range(200):
|
|
callee_qn = self._add_func(f"callee_{i}", "/t/callee.py")
|
|
self._add_calls_edge(hub_qn, callee_qn)
|
|
self.store.commit()
|
|
|
|
query_count = 0
|
|
|
|
def _trace(stmt: str) -> None:
|
|
nonlocal query_count
|
|
query_count += 1
|
|
|
|
self.store._conn.set_trace_callback(_trace)
|
|
self.store.get_transitive_tests(hub_qn, max_frontier=50)
|
|
self.store._conn.set_trace_callback(None)
|
|
|
|
# Without cap: 200 callee TESTED_BY queries + overhead = ~204
|
|
# With cap of 50: ~54 queries max
|
|
assert query_count <= 60, (
|
|
f"Expected <=60 queries with frontier cap, got {query_count}"
|
|
)
|
|
|
|
def test_uncapped_small_frontier_unchanged(self):
|
|
"""Small fan-out (< cap) returns same results regardless of cap."""
|
|
hub_qn = self._add_func("hub", "/t/hub.py")
|
|
test_qn = self._add_func("test_hub", "/t/test_hub.py")
|
|
for i in range(5):
|
|
callee_qn = self._add_func(f"callee_{i}", "/t/callee.py")
|
|
self._add_calls_edge(hub_qn, callee_qn)
|
|
# Only callee_2 has a test
|
|
if i == 2:
|
|
self.store.upsert_edge(EdgeInfo(
|
|
kind="TESTED_BY", source=callee_qn, target=test_qn,
|
|
file_path="/t/test_hub.py", line=1,
|
|
))
|
|
self.store.commit()
|
|
|
|
results_default = self.store.get_transitive_tests(hub_qn)
|
|
results_capped = self.store.get_transitive_tests(hub_qn, max_frontier=50)
|
|
|
|
indirect_default = [r for r in results_default if r["indirect"]]
|
|
indirect_capped = [r for r in results_capped if r["indirect"]]
|
|
assert len(indirect_default) == 1
|
|
assert len(indirect_capped) == 1
|
|
assert indirect_default[0]["name"] == indirect_capped[0]["name"]
|
|
|
|
|
|
class TestResolveBareEndpoints:
|
|
"""Only graph evidence may turn a bare call/test endpoint into a node."""
|
|
|
|
def setup_method(self):
|
|
self.tmp = tempfile.NamedTemporaryFile(suffix=".db", delete=False)
|
|
self.tmp.close()
|
|
self.store = GraphStore(self.tmp.name)
|
|
|
|
def teardown_method(self):
|
|
self.store.close()
|
|
Path(self.tmp.name).unlink(missing_ok=True)
|
|
|
|
def _func(self, name: str, path: str, *, is_test: bool = False) -> str:
|
|
self.store.upsert_node(NodeInfo(
|
|
kind="Test" if is_test else "Function",
|
|
name=name,
|
|
file_path=path,
|
|
line_start=1,
|
|
line_end=5,
|
|
language="python",
|
|
is_test=is_test,
|
|
))
|
|
return f"{path}::{name}"
|
|
|
|
def _edge(
|
|
self, kind: str, source: str, target: str, file_path: str,
|
|
) -> None:
|
|
self.store.upsert_edge(EdgeInfo(
|
|
kind=kind,
|
|
source=source,
|
|
target=target,
|
|
file_path=file_path,
|
|
line=1,
|
|
))
|
|
|
|
def _endpoints(self, kind: str) -> list[tuple[str, str]]:
|
|
rows = self.store._conn.execute(
|
|
"SELECT source_qualified, target_qualified FROM edges "
|
|
"WHERE kind = ? ORDER BY id",
|
|
(kind,),
|
|
).fetchall()
|
|
return [
|
|
(row["source_qualified"], row["target_qualified"])
|
|
for row in rows
|
|
]
|
|
|
|
def test_unique_tested_by_source_without_evidence_stays_bare(self):
|
|
"""A globally unique name in an unrelated file is still not evidence."""
|
|
self._func("parse", "/repo/src/app.py")
|
|
test_qn = self._func(
|
|
"test_parse", "/repo/tests/test_other.py", is_test=True,
|
|
)
|
|
self._edge("TESTED_BY", "parse", test_qn, "/repo/tests/test_other.py")
|
|
self.store.commit()
|
|
|
|
assert self.store.resolve_bare_tested_by_sources() == 0
|
|
assert self._endpoints("TESTED_BY") == [("parse", test_qn)]
|
|
|
|
def test_unique_tested_by_source_resolves_with_import_evidence(self):
|
|
source_qn = self._func("parse", "/repo/src/app.py")
|
|
test_file = "/repo/tests/test_app.py"
|
|
test_qn = self._func("test_parse", test_file, is_test=True)
|
|
self._edge("IMPORTS_FROM", test_file, "/repo/src/app.py", test_file)
|
|
self._edge("TESTED_BY", "parse", test_qn, test_file)
|
|
self.store.commit()
|
|
|
|
assert self.store.resolve_bare_tested_by_sources() == 1
|
|
assert self._endpoints("TESTED_BY") == [(source_qn, test_qn)]
|
|
|
|
def test_ambiguous_tested_by_source_uses_one_imported_candidate(self):
|
|
source_qn = self._func("parse", "/repo/src/app.py")
|
|
self._func("parse", "/repo/vendor/app.py")
|
|
test_file = "/repo/tests/test_app.py"
|
|
test_qn = self._func("test_parse", test_file, is_test=True)
|
|
self._edge("IMPORTS_FROM", test_file, "/repo/src/app.py", test_file)
|
|
self._edge("TESTED_BY", "parse", test_qn, test_file)
|
|
self.store.commit()
|
|
|
|
assert self.store.resolve_bare_tested_by_sources() == 1
|
|
assert self._endpoints("TESTED_BY") == [(source_qn, test_qn)]
|
|
|
|
def test_same_file_call_target_is_strong_evidence(self):
|
|
file_path = "/repo/src/app.py"
|
|
caller_qn = self._func("caller", file_path)
|
|
helper_qn = self._func("helper", file_path)
|
|
self._edge("CALLS", caller_qn, "helper", file_path)
|
|
self.store.commit()
|
|
|
|
assert self.store.resolve_bare_call_targets() == 1
|
|
assert self._endpoints("CALLS") == [(caller_qn, helper_qn)]
|
|
|
|
def test_unique_unrelated_call_target_stays_bare(self):
|
|
caller_file = "/repo/src/app.py"
|
|
caller_qn = self._func("caller", caller_file)
|
|
self._func("helper", "/repo/unrelated/util.py")
|
|
self._edge("CALLS", caller_qn, "helper", caller_file)
|
|
self.store.commit()
|
|
|
|
assert self.store.resolve_bare_call_targets() == 0
|
|
assert self._endpoints("CALLS") == [(caller_qn, "helper")]
|
|
|
|
def test_tests_for_does_not_guess_unrelated_bare_source(self):
|
|
source_qn = self._func("parse", "/repo/src/app.py")
|
|
test_file = "/repo/tests/test_other.py"
|
|
test_qn = self._func("test_parse", test_file, is_test=True)
|
|
self._edge("TESTED_BY", "parse", test_qn, test_file)
|
|
self.store.commit()
|
|
|
|
assert self.store.get_transitive_tests(source_qn, max_depth=0) == []
|
|
|
|
def test_tests_for_accepts_unique_import_backed_bare_source(self):
|
|
source_qn = self._func("parse", "/repo/src/app.py")
|
|
test_file = "/repo/tests/test_app.py"
|
|
test_qn = self._func("test_parse", test_file, is_test=True)
|
|
self._edge("IMPORTS_FROM", test_file, "/repo/src/app.py", test_file)
|
|
self._edge("TESTED_BY", "parse", test_qn, test_file)
|
|
self.store.commit()
|
|
|
|
results = self.store.get_transitive_tests(source_qn, max_depth=0)
|
|
assert [result["qualified_name"] for result in results] == [test_qn]
|
|
|
|
def test_tests_for_rejects_bare_source_with_two_imported_candidates(self):
|
|
first_qn = self._func("parse", "/repo/src/app.py")
|
|
second_qn = self._func("parse", "/repo/vendor/app.py")
|
|
test_file = "/repo/tests/test_app.py"
|
|
test_qn = self._func("test_parse", test_file, is_test=True)
|
|
self._edge("IMPORTS_FROM", test_file, "/repo/src/app.py", test_file)
|
|
self._edge("IMPORTS_FROM", test_file, "/repo/vendor/app.py", test_file)
|
|
self._edge("TESTED_BY", "parse", test_qn, test_file)
|
|
self.store.commit()
|
|
|
|
assert self.store.get_transitive_tests(first_qn, max_depth=0) == []
|
|
assert self.store.get_transitive_tests(second_qn, max_depth=0) == []
|
|
|
|
def test_transitive_tests_do_not_follow_unresolved_bare_callee(self):
|
|
hub_qn = self._func("hub", "/repo/src/hub.py")
|
|
self._func("parse", "/repo/unrelated/app.py")
|
|
test_file = "/repo/tests/test_app.py"
|
|
test_qn = self._func("test_parse", test_file, is_test=True)
|
|
self._edge("CALLS", hub_qn, "parse", "/repo/src/hub.py")
|
|
self._edge("TESTED_BY", "parse", test_qn, test_file)
|
|
self.store.commit()
|
|
|
|
assert self.store.get_transitive_tests(hub_qn, max_depth=1) == []
|