gabriel / muse public
test_social_plugin.py python
376 lines 15.5 KB
Raw
sha256:d11a87833d5fad6059b7662844bf5448a8911a17cce7a51811f71ad394f248eb bump to v0.2.0rc13 Human patch 7 days ago
1 """Phase 00 TDD — SocialPlugin core.
2
3 RED → GREEN cycle. Run before implementation to confirm failures, then
4 implement until all pass.
5
6 Coverage matrix:
7 TestSocialPluginShape — protocol satisfaction, registry, domain string
8 TestSocialPluginSchema — DomainSchema structure, dimensions, merge_mode
9 TestSocialPluginInMemory — snapshot pass-through, diff, drift, apply
10 TestSocialPluginMerge — set-union for posts/reactions, profile conflict
11 TestSocialPluginDescribe — schema.description is non-empty string
12 TestSocialPluginDocstrings — module, class, and method docstrings present
13 """
14 from __future__ import annotations
15
16 import hashlib
17 import inspect
18 import json
19 import pathlib
20 import sys
21 import time
22
23 import pytest
24
25 from muse.domain import (
26 DriftReport,
27 MergeResult,
28 MuseDomainPlugin,
29 SnapshotManifest,
30 StateDelta,
31 StateSnapshot,
32 StructuredDelta,
33 )
34 from muse.core.schema import DomainSchema, SetSchema, DimensionSpec
35 from muse.plugins.social.plugin import SocialPlugin
36 from muse.plugins.registry import registered_domains, resolve_plugin_by_domain
37 from muse.core.types import blob_id
38
39
40 # ---------------------------------------------------------------------------
41 # Helpers
42 # ---------------------------------------------------------------------------
43
44 def _snap(*path_content_pairs: tuple[str, bytes]) -> SnapshotManifest:
45 """Build a SnapshotManifest from (path, content) pairs."""
46 files = {path: blob_id(content) for path, content in path_content_pairs}
47 return SnapshotManifest(files=files, domain="social", directories=[])
48
49
50 def _empty_snap() -> SnapshotManifest:
51 return SnapshotManifest(files={}, domain="social", directories=[])
52
53
54 _POST_A = json.dumps({"body": "hello muse", "created_at": "2026-05-01T00:00:00Z"}).encode()
55 _POST_B = json.dumps({"body": "second post", "created_at": "2026-05-01T01:00:00Z"}).encode()
56 _POST_C = json.dumps({"body": "third post", "created_at": "2026-05-01T02:00:00Z"}).encode()
57 _PROFILE_V1 = json.dumps({"handle": "gabriel", "bio": "building the future"}).encode()
58 _PROFILE_V2 = json.dumps({"handle": "gabriel", "bio": "building the singularity"}).encode()
59 _REACTION_A = json.dumps({"emoji": "❤️", "post_id": "sha256:aaa"}).encode()
60 _FOLLOWING = json.dumps({"following": ["alice", "bob"]}).encode()
61
62
63 # ---------------------------------------------------------------------------
64 # Fixtures
65 # ---------------------------------------------------------------------------
66
67 @pytest.fixture()
68 def plugin() -> SocialPlugin:
69 return SocialPlugin()
70
71
72 # ===========================================================================
73 # Shape — protocol, registry, domain constant
74 # ===========================================================================
75
76 class TestSocialPluginShape:
77
78 def test_satisfies_protocol(self, plugin: SocialPlugin) -> None:
79 assert isinstance(plugin, MuseDomainPlugin)
80
81 def test_has_all_six_methods(self, plugin: SocialPlugin) -> None:
82 for method in ("snapshot", "diff", "merge", "drift", "apply", "schema"):
83 assert callable(getattr(plugin, method, None)), f"missing {method}"
84
85 def test_domain_constant_is_social(self) -> None:
86 from muse.plugins.social.plugin import _DOMAIN_NAME
87 assert _DOMAIN_NAME == "social"
88
89 def test_registered_in_registry(self) -> None:
90 assert "social" in registered_domains()
91
92 def test_resolve_plugin_by_domain_returns_social_plugin(self) -> None:
93 p = resolve_plugin_by_domain("social")
94 assert isinstance(p, SocialPlugin)
95
96
97 # ===========================================================================
98 # Schema
99 # ===========================================================================
100
101 class TestSocialPluginSchema:
102
103 def test_schema_returns_domain_schema(self, plugin: SocialPlugin) -> None:
104 schema = plugin.schema()
105 assert isinstance(schema, dict)
106 assert "domain" in schema
107 assert "description" in schema
108 assert "top_level" in schema
109 assert "dimensions" in schema
110 assert "merge_mode" in schema
111
112 def test_schema_domain_is_social(self, plugin: SocialPlugin) -> None:
113 assert plugin.schema()["domain"] == "social"
114
115 def test_schema_top_level_is_set(self, plugin: SocialPlugin) -> None:
116 top = plugin.schema()["top_level"]
117 assert isinstance(top, dict)
118 assert top["kind"] == "set"
119
120 def test_schema_merge_mode_three_way(self, plugin: SocialPlugin) -> None:
121 assert plugin.schema()["merge_mode"] == "three_way"
122
123 def test_schema_has_four_dimensions(self, plugin: SocialPlugin) -> None:
124 assert len(plugin.schema()["dimensions"]) == 4
125
126 def test_schema_dimension_names(self, plugin: SocialPlugin) -> None:
127 names = {d["name"] for d in plugin.schema()["dimensions"]}
128 assert names == {"posts", "reactions", "graph", "profile"}
129
130 def test_posts_dimension_independent(self, plugin: SocialPlugin) -> None:
131 posts_dim = next(d for d in plugin.schema()["dimensions"] if d["name"] == "posts")
132 assert posts_dim["independent_merge"] is True
133
134 def test_reactions_dimension_independent(self, plugin: SocialPlugin) -> None:
135 d = next(d for d in plugin.schema()["dimensions"] if d["name"] == "reactions")
136 assert d["independent_merge"] is True
137
138 def test_graph_dimension_independent(self, plugin: SocialPlugin) -> None:
139 d = next(d for d in plugin.schema()["dimensions"] if d["name"] == "graph")
140 assert d["independent_merge"] is True
141
142 def test_profile_dimension_not_independent(self, plugin: SocialPlugin) -> None:
143 d = next(d for d in plugin.schema()["dimensions"] if d["name"] == "profile")
144 assert d["independent_merge"] is False
145
146 def test_schema_is_json_serialisable(self, plugin: SocialPlugin) -> None:
147 schema = plugin.schema()
148 assert json.dumps(schema["domain"])
149 assert json.dumps(schema["merge_mode"])
150 assert json.dumps(schema["description"])
151
152 def test_schema_description_non_empty(self, plugin: SocialPlugin) -> None:
153 assert len(plugin.schema()["description"]) > 20
154
155
156 # ===========================================================================
157 # In-memory snapshot / diff / drift / apply
158 # ===========================================================================
159
160 class TestSocialPluginInMemory:
161
162 def test_snapshot_passes_through_manifest(self, plugin: SocialPlugin) -> None:
163 snap = _snap(("posts/abc.json", _POST_A))
164 result = plugin.snapshot(snap)
165 assert result["files"] == snap["files"]
166 assert result["domain"] == "social"
167
168 def test_diff_empty_to_empty_has_no_ops(self, plugin: SocialPlugin) -> None:
169 delta = plugin.diff(_empty_snap(), _empty_snap())
170 assert len(delta["ops"]) == 0
171
172 def test_diff_add_post_produces_insert_op(self, plugin: SocialPlugin) -> None:
173 base = _empty_snap()
174 target = _snap(("posts/abc.json", _POST_A))
175 delta = plugin.diff(base, target)
176 assert len(delta["ops"]) == 1
177 assert delta["ops"][0]["op"] == "insert"
178
179 def test_diff_remove_post_produces_delete_op(self, plugin: SocialPlugin) -> None:
180 base = _snap(("posts/abc.json", _POST_A))
181 target = _empty_snap()
182 delta = plugin.diff(base, target)
183 assert len(delta["ops"]) == 1
184 assert delta["ops"][0]["op"] == "delete"
185
186 def test_diff_replace_post_produces_replace_op(self, plugin: SocialPlugin) -> None:
187 base = _snap(("posts/abc.json", _POST_A))
188 target = _snap(("posts/abc.json", _POST_B))
189 delta = plugin.diff(base, target)
190 assert len(delta["ops"]) == 1
191 assert delta["ops"][0]["op"] == "replace"
192
193 def test_diff_multiple_adds(self, plugin: SocialPlugin) -> None:
194 base = _empty_snap()
195 target = _snap(
196 ("posts/aaa.json", _POST_A),
197 ("posts/bbb.json", _POST_B),
198 ("reactions/r1.json", _REACTION_A),
199 )
200 delta = plugin.diff(base, target)
201 assert len(delta["ops"]) == 3
202
203 def test_drift_no_drift_when_identical(self, plugin: SocialPlugin) -> None:
204 snap = _snap(("posts/abc.json", _POST_A))
205 report = plugin.drift(snap, snap)
206 assert report.has_drift is False
207
208 def test_drift_detects_new_post(self, plugin: SocialPlugin) -> None:
209 committed = _snap(("posts/aaa.json", _POST_A))
210 live = _snap(("posts/aaa.json", _POST_A), ("posts/bbb.json", _POST_B))
211 report = plugin.drift(committed, live)
212 assert report.has_drift is True
213
214 def test_drift_detects_deleted_post(self, plugin: SocialPlugin) -> None:
215 committed = _snap(("posts/aaa.json", _POST_A))
216 live = _empty_snap()
217 report = plugin.drift(committed, live)
218 assert report.has_drift is True
219
220 def test_drift_report_has_summary(self, plugin: SocialPlugin) -> None:
221 committed = _empty_snap()
222 live = _snap(("posts/aaa.json", _POST_A))
223 report = plugin.drift(committed, live)
224 assert isinstance(report.summary, str)
225 assert len(report.summary) > 0
226
227 def test_apply_returns_live_state_unchanged(self, plugin: SocialPlugin) -> None:
228 snap = _snap(("posts/abc.json", _POST_A))
229 delta = plugin.diff(_empty_snap(), snap)
230 result = plugin.apply(delta, snap)
231 assert result is snap
232
233
234 # ===========================================================================
235 # Merge — set-union for posts/reactions/graph, conflict on profile
236 # ===========================================================================
237
238 class TestSocialPluginMerge:
239
240 def test_merge_both_sides_add_different_posts_no_conflict(
241 self, plugin: SocialPlugin
242 ) -> None:
243 base = _empty_snap()
244 left = _snap(("posts/aaa.json", _POST_A))
245 right = _snap(("posts/bbb.json", _POST_B))
246 result = plugin.merge(base, left, right)
247 assert result.conflicts == []
248 assert "posts/aaa.json" in result.merged["files"]
249 assert "posts/bbb.json" in result.merged["files"]
250
251 def test_merge_both_sides_add_same_post_no_conflict(
252 self, plugin: SocialPlugin
253 ) -> None:
254 base = _empty_snap()
255 left = _snap(("posts/aaa.json", _POST_A))
256 right = _snap(("posts/aaa.json", _POST_A))
257 result = plugin.merge(base, left, right)
258 assert result.conflicts == []
259 assert "posts/aaa.json" in result.merged["files"]
260
261 def test_merge_only_left_adds_post(self, plugin: SocialPlugin) -> None:
262 base = _empty_snap()
263 left = _snap(("posts/aaa.json", _POST_A))
264 right = _empty_snap()
265 result = plugin.merge(base, left, right)
266 assert result.conflicts == []
267 assert "posts/aaa.json" in result.merged["files"]
268
269 def test_merge_only_right_adds_post(self, plugin: SocialPlugin) -> None:
270 base = _empty_snap()
271 left = _empty_snap()
272 right = _snap(("posts/bbb.json", _POST_B))
273 result = plugin.merge(base, left, right)
274 assert result.conflicts == []
275 assert "posts/bbb.json" in result.merged["files"]
276
277 def test_merge_reactions_set_union(self, plugin: SocialPlugin) -> None:
278 base = _empty_snap()
279 react_b = json.dumps({"emoji": "🔥", "post_id": "sha256:bbb"}).encode()
280 left = _snap(("reactions/r1.json", _REACTION_A))
281 right = _snap(("reactions/r2.json", react_b))
282 result = plugin.merge(base, left, right)
283 assert result.conflicts == []
284 assert len(result.merged["files"]) == 2
285
286 def test_merge_concurrent_profile_edits_conflict(
287 self, plugin: SocialPlugin
288 ) -> None:
289 base = _snap(("profile.json", _PROFILE_V1))
290 left = _snap(("profile.json", _PROFILE_V2))
291 right_profile = json.dumps({"handle": "gabriel", "bio": "different edit"}).encode()
292 right = _snap(("profile.json", right_profile))
293 result = plugin.merge(base, left, right)
294 assert "profile.json" in result.conflicts
295
296 def test_merge_only_left_edits_profile_no_conflict(
297 self, plugin: SocialPlugin
298 ) -> None:
299 base = _snap(("profile.json", _PROFILE_V1))
300 left = _snap(("profile.json", _PROFILE_V2))
301 right = _snap(("profile.json", _PROFILE_V1))
302 result = plugin.merge(base, left, right)
303 assert result.conflicts == []
304 assert result.merged["files"]["profile.json"] == blob_id(_PROFILE_V2)
305
306 def test_merge_only_right_edits_profile_no_conflict(
307 self, plugin: SocialPlugin
308 ) -> None:
309 base = _snap(("profile.json", _PROFILE_V1))
310 left = _snap(("profile.json", _PROFILE_V1))
311 right = _snap(("profile.json", _PROFILE_V2))
312 result = plugin.merge(base, left, right)
313 assert result.conflicts == []
314 assert result.merged["files"]["profile.json"] == blob_id(_PROFILE_V2)
315
316 def test_merge_graph_following_set_union(self, plugin: SocialPlugin) -> None:
317 # Each follow is a separate content-addressed file — set-union is automatic.
318 follow_alice = json.dumps({"handle": "alice", "since": "2026-05-01T00:00:00Z"}).encode()
319 follow_bob = json.dumps({"handle": "bob", "since": "2026-05-01T00:01:00Z"}).encode()
320 base = _empty_snap()
321 left = _snap(("graph/follows/alice.json", follow_alice))
322 right = _snap(("graph/follows/bob.json", follow_bob))
323 result = plugin.merge(base, left, right)
324 assert result.conflicts == []
325 assert "graph/follows/alice.json" in result.merged["files"]
326 assert "graph/follows/bob.json" in result.merged["files"]
327
328 def test_merge_returns_merge_result(self, plugin: SocialPlugin) -> None:
329 result = plugin.merge(_empty_snap(), _empty_snap(), _empty_snap())
330 assert isinstance(result, MergeResult)
331
332 def test_merge_result_domain_is_social(self, plugin: SocialPlugin) -> None:
333 result = plugin.merge(_empty_snap(), _empty_snap(), _empty_snap())
334 assert result.merged["domain"] == "social"
335
336 def test_merge_combined_scenario(self, plugin: SocialPlugin) -> None:
337 # Base: one post + profile
338 base = _snap(("posts/aaa.json", _POST_A), ("profile.json", _PROFILE_V1))
339 # Left: adds a new post, doesn't touch profile
340 left = _snap(
341 ("posts/aaa.json", _POST_A),
342 ("posts/bbb.json", _POST_B),
343 ("profile.json", _PROFILE_V1),
344 )
345 # Right: adds a reaction, doesn't touch profile
346 right = _snap(
347 ("posts/aaa.json", _POST_A),
348 ("reactions/r1.json", _REACTION_A),
349 ("profile.json", _PROFILE_V1),
350 )
351 result = plugin.merge(base, left, right)
352 assert result.conflicts == []
353 assert "posts/aaa.json" in result.merged["files"]
354 assert "posts/bbb.json" in result.merged["files"]
355 assert "reactions/r1.json" in result.merged["files"]
356 assert "profile.json" in result.merged["files"]
357
358
359 # ===========================================================================
360 # Docstrings
361 # ===========================================================================
362
363 class TestSocialPluginDocstrings:
364
365 def test_module_has_docstring(self) -> None:
366 import muse.plugins.social.plugin as mod
367 assert mod.__doc__ and len(mod.__doc__.strip()) > 20
368
369 def test_class_has_docstring(self, plugin: SocialPlugin) -> None:
370 assert SocialPlugin.__doc__ and len(SocialPlugin.__doc__.strip()) > 20
371
372 def test_each_method_has_docstring(self, plugin: SocialPlugin) -> None:
373 for name in ("snapshot", "diff", "merge", "drift", "apply", "schema"):
374 method = getattr(SocialPlugin, name)
375 assert method.__doc__ and len(method.__doc__.strip()) > 10, \
376 f"{name}() missing docstring"
File History 1 commit
sha256:d11a87833d5fad6059b7662844bf5448a8911a17cce7a51811f71ad394f248eb bump to v0.2.0rc13 Human patch 7 days ago