PlanOpticon

planopticon / tests / test_graph_discovery.py
Source Blame History 104 lines
b363c5b… noreply 1 """Tests for graph discovery (find_knowledge_graphs, describe_graph)."""
b363c5b… noreply 2
b363c5b… noreply 3 import json
b363c5b… noreply 4
b363c5b… noreply 5 from video_processor.integrators.graph_discovery import (
b363c5b… noreply 6 describe_graph,
b363c5b… noreply 7 find_knowledge_graphs,
b363c5b… noreply 8 find_nearest_graph,
b363c5b… noreply 9 )
b363c5b… noreply 10
b363c5b… noreply 11
b363c5b… noreply 12 class TestFindKnowledgeGraphs:
b363c5b… noreply 13 def test_finds_db_in_current_dir(self, tmp_path):
b363c5b… noreply 14 db = tmp_path / "knowledge_graph.db"
b363c5b… noreply 15 db.write_bytes(b"") # placeholder
b363c5b… noreply 16 graphs = find_knowledge_graphs(tmp_path, walk_up=False)
b363c5b… noreply 17 assert db.resolve() in graphs
b363c5b… noreply 18
b363c5b… noreply 19 def test_finds_in_results_subdir(self, tmp_path):
b363c5b… noreply 20 results = tmp_path / "results"
b363c5b… noreply 21 results.mkdir()
b363c5b… noreply 22 db = results / "knowledge_graph.db"
b363c5b… noreply 23 db.write_bytes(b"")
b363c5b… noreply 24 graphs = find_knowledge_graphs(tmp_path, walk_up=False)
b363c5b… noreply 25 assert db.resolve() in graphs
b363c5b… noreply 26
b363c5b… noreply 27 def test_finds_in_output_subdir(self, tmp_path):
b363c5b… noreply 28 output = tmp_path / "output"
b363c5b… noreply 29 output.mkdir()
b363c5b… noreply 30 db = output / "knowledge_graph.db"
b363c5b… noreply 31 db.write_bytes(b"")
b363c5b… noreply 32 graphs = find_knowledge_graphs(tmp_path, walk_up=False)
b363c5b… noreply 33 assert db.resolve() in graphs
b363c5b… noreply 34
b363c5b… noreply 35 def test_walks_up_parents(self, tmp_path):
b363c5b… noreply 36 db = tmp_path / "knowledge_graph.db"
b363c5b… noreply 37 db.write_bytes(b"")
b363c5b… noreply 38 child = tmp_path / "sub" / "deep"
b363c5b… noreply 39 child.mkdir(parents=True)
b363c5b… noreply 40 graphs = find_knowledge_graphs(child, walk_up=True)
b363c5b… noreply 41 assert db.resolve() in graphs
b363c5b… noreply 42
b363c5b… noreply 43 def test_returns_empty_when_none_found(self, tmp_path):
b363c5b… noreply 44 graphs = find_knowledge_graphs(tmp_path, walk_up=False)
b363c5b… noreply 45 assert graphs == []
b363c5b… noreply 46
b363c5b… noreply 47 def test_finds_json_fallback(self, tmp_path):
b363c5b… noreply 48 jf = tmp_path / "knowledge_graph.json"
b363c5b… noreply 49 jf.write_text('{"nodes":[], "relationships":[]}')
b363c5b… noreply 50 graphs = find_knowledge_graphs(tmp_path, walk_up=False)
b363c5b… noreply 51 assert jf.resolve() in graphs
b363c5b… noreply 52
b363c5b… noreply 53 def test_db_before_json(self, tmp_path):
b363c5b… noreply 54 db = tmp_path / "knowledge_graph.db"
b363c5b… noreply 55 db.write_bytes(b"")
b363c5b… noreply 56 jf = tmp_path / "knowledge_graph.json"
b363c5b… noreply 57 jf.write_text('{"nodes":[], "relationships":[]}')
b363c5b… noreply 58 graphs = find_knowledge_graphs(tmp_path, walk_up=False)
b363c5b… noreply 59 assert graphs.index(db.resolve()) < graphs.index(jf.resolve())
b363c5b… noreply 60
b363c5b… noreply 61 def test_closest_first_ordering(self, tmp_path):
b363c5b… noreply 62 # Deeper file
b363c5b… noreply 63 deep = tmp_path / "a" / "b"
b363c5b… noreply 64 deep.mkdir(parents=True)
b363c5b… noreply 65 deep_db = deep / "knowledge_graph.db"
b363c5b… noreply 66 deep_db.write_bytes(b"")
b363c5b… noreply 67 # Closer file
b363c5b… noreply 68 close_db = tmp_path / "knowledge_graph.db"
b363c5b… noreply 69 close_db.write_bytes(b"")
b363c5b… noreply 70 graphs = find_knowledge_graphs(tmp_path, walk_up=False)
b363c5b… noreply 71 assert graphs.index(close_db.resolve()) < graphs.index(deep_db.resolve())
b363c5b… noreply 72
b363c5b… noreply 73
b363c5b… noreply 74 class TestFindNearestGraph:
b363c5b… noreply 75 def test_returns_closest(self, tmp_path):
b363c5b… noreply 76 db = tmp_path / "knowledge_graph.db"
b363c5b… noreply 77 db.write_bytes(b"")
b363c5b… noreply 78 result = find_nearest_graph(tmp_path)
b363c5b… noreply 79 assert result == db.resolve()
b363c5b… noreply 80
b363c5b… noreply 81 def test_returns_none_when_empty(self, tmp_path):
b363c5b… noreply 82 assert find_nearest_graph(tmp_path) is None
b363c5b… noreply 83
b363c5b… noreply 84
b363c5b… noreply 85 class TestDescribeGraph:
b363c5b… noreply 86 def test_describe_json_graph(self, tmp_path):
b363c5b… noreply 87 data = {
b363c5b… noreply 88 "nodes": [
b363c5b… noreply 89 {"name": "Python", "type": "technology", "descriptions": ["A language"]},
b363c5b… noreply 90 {"name": "Django", "type": "technology", "descriptions": ["A framework"]},
b363c5b… noreply 91 {"name": "Alice", "type": "person", "descriptions": ["Engineer"]},
b363c5b… noreply 92 ],
b363c5b… noreply 93 "relationships": [
b363c5b… noreply 94 {"source": "Django", "target": "Python", "type": "uses"},
b363c5b… noreply 95 ],
b363c5b… noreply 96 }
b363c5b… noreply 97 jf = tmp_path / "knowledge_graph.json"
b363c5b… noreply 98 jf.write_text(json.dumps(data))
b363c5b… noreply 99 info = describe_graph(jf)
b363c5b… noreply 100 assert info["entity_count"] == 3
b363c5b… noreply 101 assert info["relationship_count"] == 1
b363c5b… noreply 102 assert info["entity_types"]["technology"] == 2
b363c5b… noreply 103 assert info["entity_types"]["person"] == 1
b363c5b… noreply 104 assert info["store_type"] == "json"

Keyboard Shortcuts

Open search /
Next entry (timeline) j
Previous entry (timeline) k
Open focused entry Enter
Show this help ?
Toggle theme Top nav button