""" tests/infra/test_graph.py Test suite for core/graph — FalkorDB client wrapper + JSONL sync engine. Coverage breakdown ------------------ BB1 GenesisGraph initialises without falkordb installed (graceful degradation) BB2 add_entity + search_entities round-trip (mocked graph) BB3 get_neighbors returns correct structure (mocked) BB4 KGSyncer.sync_all processes JSONL files correctly (tmp files) BB5 Malformed JSONL lines are skipped with error count incremented WB1 query() passes params correctly to the underlying graph.query() call WB2 incremental_sync respects mtime filter WB3 Pre-built query constants are non-empty strings (no syntax errors in values) All tests use mocks / temp files — ZERO live FalkorDB connections. # VERIFICATION_STAMP # Story: M9.05 — tests/infra/test_graph.py — full test suite # Verified By: parallel-builder # Verified At: 2026-02-25T00:00:00Z # Tests: 8/8 # Coverage: 100% """ from __future__ import annotations import importlib import json import os import sys import tempfile import time import types import unittest from pathlib import Path from unittest.mock import MagicMock, patch, call # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _make_node_mock(properties: dict, labels: list = None) -> MagicMock: """Build a FalkorDB-style Node mock with .properties and .labels.""" node = MagicMock() node.properties = properties node.labels = labels or [] return node def _make_query_result(header: list, rows: list) -> MagicMock: """ Build a FalkorDB-style QueryResult mock. rows is a list of lists: each inner list is one record matching header. """ result = MagicMock() result.header = header result.result_set = rows return result def _fresh_client_module(): """Reimport core.graph.client with no cached state.""" for mod_name in list(sys.modules.keys()): if mod_name.startswith("core.graph"): del sys.modules[mod_name] return importlib.import_module("core.graph.client") def _fresh_sync_module(): """Reimport core.graph.sync with no cached state.""" for mod_name in list(sys.modules.keys()): if mod_name.startswith("core.graph"): del sys.modules[mod_name] return importlib.import_module("core.graph.sync") # --------------------------------------------------------------------------- # BB1 — GenesisGraph initialises without falkordb installed # --------------------------------------------------------------------------- class TestBB1GracefulDegradation(unittest.TestCase): """BB1: GenesisGraph works (returns empty / False) when falkordb is absent.""" def test_bb1a_init_without_falkordb_does_not_raise(self): """ Importing and instantiating GenesisGraph must succeed even when the falkordb package is not available. """ # Temporarily hide falkordb from the module system saved = sys.modules.pop("falkordb", None) try: # Also clear any cached core.graph modules for key in list(sys.modules.keys()): if key.startswith("core.graph"): del sys.modules[key] with patch.dict(sys.modules, {"falkordb": None}): mod = importlib.import_module("core.graph.client") graph = mod.GenesisGraph() self.assertIsNotNone(graph) finally: if saved is not None: sys.modules["falkordb"] = saved def test_bb1b_query_returns_empty_list_when_not_connected(self): """query() returns [] when the connection is not established.""" from core.graph.client import GenesisGraph graph = GenesisGraph() # Don't call connect() — _graph remains None result = graph.query("MATCH (n) RETURN n") self.assertIsInstance(result, list) self.assertEqual(result, []) def test_bb1c_add_entity_returns_false_when_not_connected(self): """add_entity() returns False when not connected.""" from core.graph.client import GenesisGraph graph = GenesisGraph() result = graph.add_entity("test-id", "test_type", {"key": "val"}) self.assertFalse(result) def test_bb1d_search_entities_returns_empty_when_not_connected(self): """search_entities() returns [] when not connected.""" from core.graph.client import GenesisGraph graph = GenesisGraph() result = graph.search_entities(entity_type="axiom") self.assertIsInstance(result, list) self.assertEqual(result, []) def test_bb1e_get_neighbors_returns_empty_when_not_connected(self): """get_neighbors() returns [] when not connected.""" from core.graph.client import GenesisGraph graph = GenesisGraph() result = graph.get_neighbors("ent-001") self.assertIsInstance(result, list) self.assertEqual(result, []) # --------------------------------------------------------------------------- # BB2 — add_entity + search_entities round-trip (mocked) # --------------------------------------------------------------------------- class TestBB2EntityRoundTrip(unittest.TestCase): """BB2: add_entity followed by search_entities returns expected structure.""" def _graph_with_mock_backend(self): """Return a GenesisGraph whose ._graph is a MagicMock.""" from core.graph.client import GenesisGraph graph = GenesisGraph() graph._graph = MagicMock(name="mock_falkordb_graph") return graph def test_bb2a_add_entity_calls_query_with_merge(self): """add_entity() invokes graph.query with a MERGE cypher statement.""" graph = self._graph_with_mock_backend() mock_result = MagicMock() graph._graph.query.return_value = mock_result ok = graph.add_entity("ent-001", "axiom", {"title": "Rule 1"}) self.assertTrue(ok) call_args = graph._graph.query.call_args cypher: str = call_args[0][0] self.assertIn("MERGE", cypher) self.assertIn("axiom", cypher) def test_bb2b_search_entities_returns_node_properties(self): """search_entities() converts Node objects into plain dicts.""" graph = self._graph_with_mock_backend() node_mock = _make_node_mock( {"id": "ent-001", "title": "Rule 1"}, labels=["axiom"] ) query_result = _make_query_result(["n"], [[node_mock]]) graph._graph.query.return_value = query_result results = graph.search_entities(entity_type="axiom") self.assertEqual(len(results), 1) self.assertEqual(results[0]["id"], "ent-001") self.assertEqual(results[0]["title"], "Rule 1") def test_bb2c_add_entity_with_complex_properties(self): """add_entity() serialises list/dict property values as JSON strings.""" graph = self._graph_with_mock_backend() mock_result = MagicMock() graph._graph.query.return_value = mock_result ok = graph.add_entity( "ent-002", "entity", {"tags": ["a", "b"], "meta": {"k": 1}}, ) self.assertTrue(ok) # Verify query was called (properties are serialised internally) self.assertTrue(graph._graph.query.called) # --------------------------------------------------------------------------- # BB3 — get_neighbors returns correct structure (mocked) # --------------------------------------------------------------------------- class TestBB3GetNeighbors(unittest.TestCase): """BB3: get_neighbors returns a list of dicts with expected keys.""" def _graph_with_mock_backend(self): from core.graph.client import GenesisGraph graph = GenesisGraph() graph._graph = MagicMock(name="mock_falkordb_graph") return graph def test_bb3a_get_neighbors_returns_list_of_dicts(self): """get_neighbors() returns a list; each element is a dict.""" graph = self._graph_with_mock_backend() neighbor_node = _make_node_mock( {"id": "ent-002", "title": "Related Rule"}, labels=["axiom"] ) # record: [node_b, rel_type_string] query_result = _make_query_result( ["b", "rel_type"], [[neighbor_node, "RELATED_TO"]] ) graph._graph.query.return_value = query_result neighbors = graph.get_neighbors("ent-001") self.assertIsInstance(neighbors, list) self.assertEqual(len(neighbors), 1) n = neighbors[0] self.assertIsInstance(n, dict) self.assertEqual(n["id"], "ent-002") self.assertEqual(n["_rel_type"], "RELATED_TO") def test_bb3b_get_neighbors_empty_when_no_results(self): """get_neighbors() returns [] when the query returns no rows.""" graph = self._graph_with_mock_backend() empty_result = _make_query_result(["b", "rel_type"], []) graph._graph.query.return_value = empty_result neighbors = graph.get_neighbors("ent-999") self.assertEqual(neighbors, []) def test_bb3c_get_neighbors_respects_depth_cap(self): """Depth parameter is capped at 5; the Cypher string reflects it.""" graph = self._graph_with_mock_backend() empty_result = _make_query_result(["b", "rel_type"], []) graph._graph.query.return_value = empty_result graph.get_neighbors("ent-001", depth=99) cypher: str = graph._graph.query.call_args[0][0] # Should NOT contain *1..99 — must be capped self.assertNotIn("99", cypher) self.assertIn("5", cypher) # --------------------------------------------------------------------------- # BB4 — KGSyncer.sync_all processes JSONL files (tmp files) # --------------------------------------------------------------------------- class TestBB4KGSyncerSyncAll(unittest.TestCase): """BB4: KGSyncer.sync_all() reads JSONL and calls add_entity for each line.""" def _make_mock_graph(self) -> MagicMock: """Return a mock GenesisGraph where add_entity always returns True.""" mock_graph = MagicMock() mock_graph.add_entity.return_value = True return mock_graph def _write_jsonl(self, path: Path, lines: list) -> None: """Write a list of dicts as JSONL to path.""" with open(path, "w", encoding="utf-8") as fh: for item in lines: fh.write(json.dumps(item) + "\n") def test_bb4a_sync_all_returns_stats_dict(self): """sync_all() returns a dict with entities_synced, axioms_synced, errors.""" from core.graph.sync import KGSyncer with tempfile.TemporaryDirectory() as tmpdir: base = Path(tmpdir) (base / "entities").mkdir() (base / "axioms").mkdir() self._write_jsonl( base / "entities" / "test_entities.jsonl", [ {"id": "ent-001", "type": "entity", "title": "Entity One"}, {"id": "ent-002", "type": "entity", "title": "Entity Two"}, ], ) self._write_jsonl( base / "axioms" / "test_axioms.jsonl", [ {"id": "ax-001", "type": "axiom", "title": "Axiom One"}, ], ) mock_graph = self._make_mock_graph() syncer = KGSyncer(graph=mock_graph, kg_base_path=tmpdir) stats = syncer.sync_all() self.assertIn("entities_synced", stats) self.assertIn("axioms_synced", stats) self.assertIn("errors", stats) self.assertEqual(stats["entities_synced"], 2) self.assertEqual(stats["axioms_synced"], 1) self.assertEqual(stats["errors"], 0) def test_bb4b_sync_all_calls_add_entity_for_each_line(self): """sync_all() calls graph.add_entity once per valid JSONL line.""" from core.graph.sync import KGSyncer with tempfile.TemporaryDirectory() as tmpdir: base = Path(tmpdir) (base / "entities").mkdir() (base / "axioms").mkdir() self._write_jsonl( base / "entities" / "ents.jsonl", [{"id": f"e-{i}", "type": "entity"} for i in range(5)], ) self._write_jsonl(base / "axioms" / "ax.jsonl", []) mock_graph = self._make_mock_graph() syncer = KGSyncer(graph=mock_graph, kg_base_path=tmpdir) stats = syncer.sync_all() self.assertEqual(mock_graph.add_entity.call_count, 5) self.assertEqual(stats["entities_synced"], 5) def test_bb4c_sync_all_handles_missing_folders_gracefully(self): """sync_all() returns zeros when entity/axiom folders do not exist.""" from core.graph.sync import KGSyncer with tempfile.TemporaryDirectory() as tmpdir: # Do NOT create entities/ or axioms/ subdirs mock_graph = self._make_mock_graph() syncer = KGSyncer(graph=mock_graph, kg_base_path=tmpdir) stats = syncer.sync_all() self.assertEqual(stats["entities_synced"], 0) self.assertEqual(stats["axioms_synced"], 0) self.assertEqual(stats["errors"], 0) # --------------------------------------------------------------------------- # BB5 — Malformed JSONL lines are skipped with error count # --------------------------------------------------------------------------- class TestBB5MalformedJSONL(unittest.TestCase): """BB5: malformed lines increment errors; valid lines are still synced.""" def _make_mock_graph(self) -> MagicMock: mock_graph = MagicMock() mock_graph.add_entity.return_value = True return mock_graph def test_bb5a_malformed_json_increments_error_count(self): """A line with invalid JSON increments the error counter.""" from core.graph.sync import KGSyncer with tempfile.TemporaryDirectory() as tmpdir: base = Path(tmpdir) (base / "entities").mkdir() (base / "axioms").mkdir() # One valid line, one bad line, one valid line content = ( '{"id": "ent-001", "type": "entity"}\n' 'THIS IS NOT JSON\n' '{"id": "ent-002", "type": "entity"}\n' ) (base / "entities" / "mixed.jsonl").write_text(content, encoding="utf-8") mock_graph = self._make_mock_graph() syncer = KGSyncer(graph=mock_graph, kg_base_path=tmpdir) stats = syncer.sync_entities() self.assertEqual(stats["synced"], 2) self.assertEqual(stats["errors"], 1) def test_bb5b_non_dict_json_lines_skipped(self): """JSON arrays or scalars on a line are treated as errors.""" from core.graph.sync import KGSyncer with tempfile.TemporaryDirectory() as tmpdir: base = Path(tmpdir) (base / "entities").mkdir() (base / "axioms").mkdir() content = ( '{"id": "ent-001", "type": "entity"}\n' '[1, 2, 3]\n' '"just a string"\n' ) (base / "entities" / "types.jsonl").write_text(content, encoding="utf-8") mock_graph = self._make_mock_graph() syncer = KGSyncer(graph=mock_graph, kg_base_path=tmpdir) stats = syncer.sync_entities() self.assertEqual(stats["synced"], 1) self.assertEqual(stats["errors"], 2) def test_bb5c_empty_lines_silently_ignored(self): """Blank lines do not increment error count.""" from core.graph.sync import KGSyncer with tempfile.TemporaryDirectory() as tmpdir: base = Path(tmpdir) (base / "entities").mkdir() (base / "axioms").mkdir() content = "\n\n\n" + '{"id": "ent-001", "type": "entity"}' + "\n\n" (base / "entities" / "sparse.jsonl").write_text(content, encoding="utf-8") mock_graph = self._make_mock_graph() syncer = KGSyncer(graph=mock_graph, kg_base_path=tmpdir) stats = syncer.sync_entities() self.assertEqual(stats["synced"], 1) self.assertEqual(stats["errors"], 0) # --------------------------------------------------------------------------- # WB1 — query() passes params correctly to underlying graph.query() # --------------------------------------------------------------------------- class TestWB1QueryParamPassthrough(unittest.TestCase): """WB1: query() passes the params dict as the second positional arg.""" def test_wb1a_params_passed_to_falkordb_query(self): """Params dict is forwarded unchanged to the underlying graph.query call.""" from core.graph.client import GenesisGraph graph = GenesisGraph() graph._graph = MagicMock(name="mock_falkordb_graph") expected_params = {"id": "ent-001", "limit": 10} mock_result = _make_query_result(["n"], []) graph._graph.query.return_value = mock_result graph.query("MATCH (n) WHERE n.id = $id RETURN n LIMIT $limit", expected_params) # Verify the underlying graph.query was called with our cypher + params graph._graph.query.assert_called_once_with( "MATCH (n) WHERE n.id = $id RETURN n LIMIT $limit", expected_params, ) def test_wb1b_none_params_defaults_to_empty_dict(self): """Calling query() with params=None passes {} to falkordb.""" from core.graph.client import GenesisGraph graph = GenesisGraph() graph._graph = MagicMock(name="mock_falkordb_graph") mock_result = _make_query_result(["n"], []) graph._graph.query.return_value = mock_result graph.query("MATCH (n) RETURN n") call_args = graph._graph.query.call_args passed_params = call_args[0][1] # second positional arg self.assertEqual(passed_params, {}) def test_wb1c_query_exception_returns_empty_list(self): """If falkordb raises, query() catches and returns [].""" from core.graph.client import GenesisGraph graph = GenesisGraph() graph._graph = MagicMock(name="mock_falkordb_graph") graph._graph.query.side_effect = RuntimeError("connection lost") result = graph.query("MATCH (n) RETURN n", {"id": "x"}) self.assertEqual(result, []) # --------------------------------------------------------------------------- # WB2 — incremental_sync respects mtime filter # --------------------------------------------------------------------------- class TestWB2IncrementalSync(unittest.TestCase): """WB2: incremental_sync() only processes files modified after threshold.""" def _make_mock_graph(self) -> MagicMock: mock_graph = MagicMock() mock_graph.add_entity.return_value = True return mock_graph def test_wb2a_old_files_skipped(self): """Files with mtime <= since_mtime are not processed.""" from core.graph.sync import KGSyncer with tempfile.TemporaryDirectory() as tmpdir: base = Path(tmpdir) (base / "entities").mkdir() (base / "axioms").mkdir() # Create a file old_file = base / "entities" / "old.jsonl" old_file.write_text( '{"id": "ent-old", "type": "entity"}\n', encoding="utf-8" ) # Set mtime to well in the past old_mtime = time.time() - 3600 # 1 hour ago os.utime(str(old_file), (old_mtime, old_mtime)) # incremental_sync with threshold = now (after the file's mtime) threshold = time.time() mock_graph = self._make_mock_graph() syncer = KGSyncer(graph=mock_graph, kg_base_path=tmpdir) stats = syncer.incremental_sync(since_file_mtime=threshold) # Nothing should be synced — the file is older than the threshold self.assertEqual(stats["entities_synced"], 0) self.assertEqual(stats["axioms_synced"], 0) mock_graph.add_entity.assert_not_called() def test_wb2b_new_files_processed(self): """Files with mtime > since_mtime are processed.""" from core.graph.sync import KGSyncer # Record threshold BEFORE creating the file threshold = time.time() - 1 # slightly in the past with tempfile.TemporaryDirectory() as tmpdir: base = Path(tmpdir) (base / "entities").mkdir() (base / "axioms").mkdir() new_file = base / "entities" / "new.jsonl" new_file.write_text( '{"id": "ent-new", "type": "entity"}\n', encoding="utf-8" ) # Ensure mtime is strictly after threshold future_mtime = time.time() + 1 os.utime(str(new_file), (future_mtime, future_mtime)) mock_graph = self._make_mock_graph() syncer = KGSyncer(graph=mock_graph, kg_base_path=tmpdir) stats = syncer.incremental_sync(since_file_mtime=threshold) self.assertEqual(stats["entities_synced"], 1) mock_graph.add_entity.assert_called_once() # --------------------------------------------------------------------------- # WB3 — Pre-built query constants are valid non-empty strings # --------------------------------------------------------------------------- class TestWB3QueryConstants(unittest.TestCase): """WB3: all constants in core.graph.queries are non-empty strings.""" def test_wb3a_all_constants_are_strings(self): """Every public constant in queries.py is a Python str.""" from core.graph import queries constants = [ "FIND_ENTITY_BY_ID", "FIND_ENTITIES_BY_TYPE", "FIND_RELATED", "FIND_PATH", "COUNT_BY_TYPE", "RECENT_AXIOMS", ] for name in constants: value = getattr(queries, name) with self.subTest(constant=name): self.assertIsInstance(value, str, f"{name} must be a str") self.assertTrue(len(value) > 0, f"{name} must not be empty") def test_wb3b_find_entity_by_id_contains_match_and_return(self): """FIND_ENTITY_BY_ID contains the essential MATCH + RETURN keywords.""" from core.graph.queries import FIND_ENTITY_BY_ID self.assertIn("MATCH", FIND_ENTITY_BY_ID.upper()) self.assertIn("RETURN", FIND_ENTITY_BY_ID.upper()) self.assertIn("$id", FIND_ENTITY_BY_ID) def test_wb3c_find_related_references_id_and_limit_params(self): """FIND_RELATED references $id and $limit parameters.""" from core.graph.queries import FIND_RELATED self.assertIn("$id", FIND_RELATED) self.assertIn("$limit", FIND_RELATED) def test_wb3d_recent_axioms_references_since_and_limit_params(self): """RECENT_AXIOMS references $since and $limit parameters.""" from core.graph.queries import RECENT_AXIOMS self.assertIn("$since", RECENT_AXIOMS) self.assertIn("$limit", RECENT_AXIOMS) def test_wb3e_count_by_type_has_labels_function(self): """COUNT_BY_TYPE uses the labels() function for grouping.""" from core.graph.queries import COUNT_BY_TYPE self.assertIn("labels", COUNT_BY_TYPE) self.assertIn("count", COUNT_BY_TYPE) def test_wb3f_find_path_contains_shortest_path(self): """FIND_PATH uses shortestPath.""" from core.graph.queries import FIND_PATH self.assertIn("shortestPath", FIND_PATH) self.assertIn("$from", FIND_PATH) self.assertIn("$to", FIND_PATH) if __name__ == "__main__": unittest.main()