"""Tests for the graph storage and query engine.""" import logging import sqlite3 import tempfile from pathlib import Path from code_review_graph.graph import GraphStore from code_review_graph.parser import EdgeInfo, NodeInfo class TestGraphStore: def setup_method(self): self.tmp = tempfile.NamedTemporaryFile(suffix=".db", delete=False) 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_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_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_impact_radius(self): # Create a chain: file_a -> func_a -> (calls) -> func_b in file_b 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="/a.py::func_a", target="/b.py::func_b", file_path="/a.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_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.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 A -> B -> C -> D 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="/a.py::func_a", target="/b.py::func_b", file_path="/a.py", line=10, )) self.store.upsert_edge(EdgeInfo( kind="CALLS", source="/b.py::func_b", target="/c.py::func_c", file_path="/b.py", line=10, )) self.store.upsert_edge(EdgeInfo( kind="CALLS", source="/c.py::func_c", target="/d.py::func_d", file_path="/c.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 == 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, ) # With 4 files in chain + file nodes, max_nodes=2 should limit assert result["total_impacted"] <= 2 or result["truncated"] 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 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.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=test_qn, target=callee_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"]