PlanOpticon

planopticon / tests / test_exchange.py
Source Blame History 224 lines
0981a08… noreply 1 """Tests for the PlanOpticonExchange interchange format."""
0981a08… noreply 2
0981a08… noreply 3 import json
0981a08… noreply 4
0981a08… noreply 5 from video_processor.exchange import (
0981a08… noreply 6 ArtifactMeta,
0981a08… noreply 7 PlanOpticonExchange,
0981a08… noreply 8 ProjectMeta,
0981a08… noreply 9 )
0981a08… noreply 10 from video_processor.models import Entity, Relationship, SourceRecord
0981a08… noreply 11
0981a08… noreply 12 # ------------------------------------------------------------------
0981a08… noreply 13 # Fixtures
0981a08… noreply 14 # ------------------------------------------------------------------
0981a08… noreply 15
0981a08… noreply 16
0981a08… noreply 17 def _sample_entity(name: str = "Python", etype: str = "technology"):
0981a08… noreply 18 return Entity(
0981a08… noreply 19 name=name,
0981a08… noreply 20 type=etype,
0981a08… noreply 21 descriptions=["A programming language"],
0981a08… noreply 22 )
0981a08… noreply 23
0981a08… noreply 24
0981a08… noreply 25 def _sample_relationship():
0981a08… noreply 26 return Relationship(
0981a08… noreply 27 source="Alice",
0981a08… noreply 28 target="Python",
0981a08… noreply 29 type="uses",
0981a08… noreply 30 )
0981a08… noreply 31
0981a08… noreply 32
0981a08… noreply 33 def _sample_source():
0981a08… noreply 34 return SourceRecord(
0981a08… noreply 35 source_id="src-1",
0981a08… noreply 36 source_type="video",
0981a08… noreply 37 title="Intro recording",
0981a08… noreply 38 )
0981a08… noreply 39
0981a08… noreply 40
0981a08… noreply 41 def _sample_artifact():
0981a08… noreply 42 return ArtifactMeta(
0981a08… noreply 43 name="roadmap",
0981a08… noreply 44 content="# Roadmap\n- Phase 1",
0981a08… noreply 45 artifact_type="roadmap",
0981a08… noreply 46 format="markdown",
0981a08… noreply 47 )
0981a08… noreply 48
0981a08… noreply 49
0981a08… noreply 50 def _sample_project():
0981a08… noreply 51 return ProjectMeta(name="TestProject", description="A test")
0981a08… noreply 52
0981a08… noreply 53
0981a08… noreply 54 # ------------------------------------------------------------------
0981a08… noreply 55 # Tests
0981a08… noreply 56 # ------------------------------------------------------------------
0981a08… noreply 57
0981a08… noreply 58
0981a08… noreply 59 def test_create_empty_exchange():
0981a08… noreply 60 ex = PlanOpticonExchange(project=_sample_project())
0981a08… noreply 61 assert ex.version == "1.0"
0981a08… noreply 62 assert ex.entities == []
0981a08… noreply 63 assert ex.relationships == []
0981a08… noreply 64 assert ex.artifacts == []
0981a08… noreply 65 assert ex.sources == []
0981a08… noreply 66 assert ex.project.name == "TestProject"
0981a08… noreply 67
0981a08… noreply 68
0981a08… noreply 69 def test_create_with_data():
0981a08… noreply 70 ex = PlanOpticonExchange(
0981a08… noreply 71 project=_sample_project(),
0981a08… noreply 72 entities=[_sample_entity()],
0981a08… noreply 73 relationships=[_sample_relationship()],
0981a08… noreply 74 artifacts=[_sample_artifact()],
0981a08… noreply 75 sources=[_sample_source()],
0981a08… noreply 76 )
0981a08… noreply 77 assert len(ex.entities) == 1
0981a08… noreply 78 assert ex.entities[0].name == "Python"
0981a08… noreply 79 assert len(ex.relationships) == 1
0981a08… noreply 80 assert len(ex.artifacts) == 1
0981a08… noreply 81 assert len(ex.sources) == 1
0981a08… noreply 82
0981a08… noreply 83
0981a08… noreply 84 def test_json_roundtrip(tmp_path):
0981a08… noreply 85 original = PlanOpticonExchange(
0981a08… noreply 86 project=_sample_project(),
0981a08… noreply 87 entities=[_sample_entity()],
0981a08… noreply 88 relationships=[_sample_relationship()],
0981a08… noreply 89 artifacts=[_sample_artifact()],
0981a08… noreply 90 sources=[_sample_source()],
0981a08… noreply 91 )
0981a08… noreply 92 out = tmp_path / "exchange.json"
0981a08… noreply 93 original.to_file(out)
0981a08… noreply 94
0981a08… noreply 95 assert out.exists()
0981a08… noreply 96 loaded = PlanOpticonExchange.from_file(out)
0981a08… noreply 97 assert loaded.project.name == original.project.name
0981a08… noreply 98 assert len(loaded.entities) == 1
0981a08… noreply 99 assert loaded.entities[0].name == "Python"
0981a08… noreply 100 assert len(loaded.relationships) == 1
0981a08… noreply 101 assert len(loaded.artifacts) == 1
0981a08… noreply 102 assert len(loaded.sources) == 1
0981a08… noreply 103
0981a08… noreply 104 # Verify valid JSON on disk
0981a08… noreply 105 raw = json.loads(out.read_text())
0981a08… noreply 106 assert raw["version"] == "1.0"
0981a08… noreply 107
0981a08… noreply 108
0981a08… noreply 109 def test_json_schema_export():
0981a08… noreply 110 schema = PlanOpticonExchange.json_schema()
0981a08… noreply 111 assert isinstance(schema, dict)
0981a08… noreply 112 assert "properties" in schema
0981a08… noreply 113 assert "version" in schema["properties"]
0981a08… noreply 114 assert "project" in schema["properties"]
0981a08… noreply 115 assert "entities" in schema["properties"]
0981a08… noreply 116
0981a08… noreply 117
0981a08… noreply 118 def test_from_knowledge_graph():
0981a08… noreply 119 kg_dict = {
0981a08… noreply 120 "nodes": [
0981a08… noreply 121 {
0981a08… noreply 122 "id": "python",
0981a08… noreply 123 "name": "Python",
0981a08… noreply 124 "type": "technology",
0981a08… noreply 125 "descriptions": ["A language"],
0981a08… noreply 126 "occurrences": [],
0981a08… noreply 127 },
0981a08… noreply 128 {
0981a08… noreply 129 "id": "alice",
0981a08… noreply 130 "name": "Alice",
0981a08… noreply 131 "type": "person",
0981a08… noreply 132 "descriptions": ["Engineer"],
0981a08… noreply 133 "occurrences": [],
0981a08… noreply 134 },
0981a08… noreply 135 ],
0981a08… noreply 136 "relationships": [
0981a08… noreply 137 {
0981a08… noreply 138 "source": "Alice",
0981a08… noreply 139 "target": "Python",
0981a08… noreply 140 "type": "uses",
0981a08… noreply 141 },
0981a08… noreply 142 ],
0981a08… noreply 143 "sources": [
0981a08… noreply 144 {
0981a08… noreply 145 "source_id": "s1",
0981a08… noreply 146 "source_type": "video",
0981a08… noreply 147 "title": "Recording",
0981a08… noreply 148 },
0981a08… noreply 149 ],
0981a08… noreply 150 }
0981a08… noreply 151
0981a08… noreply 152 ex = PlanOpticonExchange.from_knowledge_graph(
0981a08… noreply 153 kg_dict,
0981a08… noreply 154 project_name="Demo",
0981a08… noreply 155 tags=["test"],
0981a08… noreply 156 )
0981a08… noreply 157 assert ex.project.name == "Demo"
0981a08… noreply 158 assert len(ex.entities) == 2
0981a08… noreply 159 assert len(ex.relationships) == 1
0981a08… noreply 160 assert len(ex.sources) == 1
0981a08… noreply 161 assert "test" in ex.project.tags
0981a08… noreply 162
0981a08… noreply 163
0981a08… noreply 164 def test_merge_deduplicates_entities():
0981a08… noreply 165 ex1 = PlanOpticonExchange(
0981a08… noreply 166 project=_sample_project(),
0981a08… noreply 167 entities=[_sample_entity("Python"), _sample_entity("Rust")],
0981a08… noreply 168 relationships=[_sample_relationship()],
0981a08… noreply 169 sources=[_sample_source()],
0981a08… noreply 170 )
0981a08… noreply 171 ex2 = PlanOpticonExchange(
0981a08… noreply 172 project=ProjectMeta(name="Other"),
0981a08… noreply 173 entities=[
0981a08… noreply 174 _sample_entity("Python"), # duplicate
0981a08… noreply 175 _sample_entity("Go"), # new
0981a08… noreply 176 ],
0981a08… noreply 177 relationships=[
0981a08… noreply 178 Relationship(source="Bob", target="Go", type="uses"),
0981a08… noreply 179 ],
0981a08… noreply 180 sources=[
0981a08… noreply 181 SourceRecord(
0981a08… noreply 182 source_id="src-2",
0981a08… noreply 183 source_type="document",
0981a08… noreply 184 title="Notes",
0981a08… noreply 185 ),
0981a08… noreply 186 ],
0981a08… noreply 187 )
0981a08… noreply 188
0981a08… noreply 189 ex1.merge(ex2)
0981a08… noreply 190
0981a08… noreply 191 entity_names = [e.name for e in ex1.entities]
0981a08… noreply 192 assert entity_names.count("Python") == 1
0981a08… noreply 193 assert "Go" in entity_names
0981a08… noreply 194 assert "Rust" in entity_names
0981a08… noreply 195 assert len(ex1.entities) == 3
0981a08… noreply 196 assert len(ex1.relationships) == 2
0981a08… noreply 197 assert len(ex1.sources) == 2
0981a08… noreply 198
0981a08… noreply 199
0981a08… noreply 200 def test_version_field():
0981a08… noreply 201 ex = PlanOpticonExchange(
0981a08… noreply 202 version="2.0",
0981a08… noreply 203 project=_sample_project(),
0981a08… noreply 204 )
0981a08… noreply 205 assert ex.version == "2.0"
0981a08… noreply 206
0981a08… noreply 207
0981a08… noreply 208 def test_artifact_meta_model():
0981a08… noreply 209 art = ArtifactMeta(
0981a08… noreply 210 name="plan",
0981a08… noreply 211 content="# Plan\nDo stuff",
0981a08… noreply 212 artifact_type="project_plan",
0981a08… noreply 213 format="markdown",
0981a08… noreply 214 metadata={"author": "agent"},
0981a08… noreply 215 )
0981a08… noreply 216 assert art.name == "plan"
0981a08… noreply 217 assert art.artifact_type == "project_plan"
0981a08… noreply 218 assert art.format == "markdown"
0981a08… noreply 219 assert art.metadata == {"author": "agent"}
0981a08… noreply 220
0981a08… noreply 221 # Roundtrip via dict
0981a08… noreply 222 d = art.model_dump()
0981a08… noreply 223 restored = ArtifactMeta.model_validate(d)
0981a08… noreply 224 assert restored == art

Keyboard Shortcuts

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