gabriel / muse public
test_identity_domain_schema.py python
145 lines 5.8 KB
Raw
sha256:b636f72dcba9e190afb980bece906fa5b717fbde014b76ef023df8cb96e01eb9 docs: expand cache plan with all seven testing tiers and do… Sonnet 4.6 134 days ago
1 """TDD — IdentityPlugin.schema() and protocol conformance.
2
3 Covers:
4 - schema() returns a valid DomainSchema
5 - domain field is "identity"
6 - Two dimensions: "identities" and "relationships"
7 - Both dimensions are independently mergeable
8 - "identities" dimension uses SetSchema keyed by handle
9 - "relationships" dimension uses SetSchema keyed by (from, edge_type, to)
10 - merge_mode is "three_way"
11 - Plugin satisfies MuseDomainPlugin protocol (runtime isinstance check)
12 - Plugin satisfies StructuredMergePlugin protocol (has merge_ops)
13 - Plugin satisfies HarmonyPlugin protocol (has conflict_fingerprint)
14 - drift() delegates to snapshot + diff correctly
15 - apply() returns live_state unchanged (post-processing no-op for now)
16 """
17 from __future__ import annotations
18
19 import pytest
20
21 from muse.domain import (
22 DriftReport,
23 LiveState,
24 MergeResult,
25 MuseDomainPlugin,
26 SnapshotManifest,
27 StateDelta,
28 StateSnapshot,
29 StructuredMergePlugin,
30 )
31 from muse.plugins.identity.plugin import IdentityPlugin
32
33
34 @pytest.fixture
35 def plugin() -> IdentityPlugin:
36 return IdentityPlugin()
37
38
39 # ── DomainSchema ──────────────────────────────────────────────────────────────
40
41 class TestSchema:
42 def test_domain_is_identity(self, plugin):
43 assert plugin.schema()["domain"] == "identity"
44
45 def test_description_non_empty(self, plugin):
46 assert plugin.schema()["description"]
47
48 def test_has_two_dimensions(self, plugin):
49 dims = plugin.schema()["dimensions"]
50 assert len(dims) == 2
51
52 def test_identities_dimension_exists(self, plugin):
53 names = {d["name"] for d in plugin.schema()["dimensions"]}
54 assert "identities" in names
55
56 def test_relationships_dimension_exists(self, plugin):
57 names = {d["name"] for d in plugin.schema()["dimensions"]}
58 assert "relationships" in names
59
60 def test_both_dimensions_independently_mergeable(self, plugin):
61 for dim in plugin.schema()["dimensions"]:
62 assert dim["independent_merge"] is True
63
64 def test_identities_dimension_is_set_schema(self, plugin):
65 dim = next(d for d in plugin.schema()["dimensions"] if d["name"] == "identities")
66 assert dim["schema"]["kind"] == "set"
67
68 def test_relationships_dimension_is_set_schema(self, plugin):
69 dim = next(d for d in plugin.schema()["dimensions"] if d["name"] == "relationships")
70 assert dim["schema"]["kind"] == "set"
71
72 def test_merge_mode_is_three_way(self, plugin):
73 assert plugin.schema()["merge_mode"] == "three_way"
74
75 def test_schema_version_is_string(self, plugin):
76 assert isinstance(plugin.schema()["schema_version"], str)
77
78
79 # ── Protocol conformance ──────────────────────────────────────────────────────
80
81 class TestProtocolConformance:
82 def test_satisfies_muse_domain_plugin(self, plugin):
83 assert isinstance(plugin, MuseDomainPlugin)
84
85 def test_satisfies_structured_merge_plugin(self, plugin):
86 assert isinstance(plugin, StructuredMergePlugin)
87
88 def test_has_conflict_fingerprint_harmony(self, plugin):
89 assert callable(getattr(plugin, "conflict_fingerprint", None))
90
91 def test_has_all_six_required_methods(self, plugin):
92 for method in ("snapshot", "diff", "merge", "drift", "apply", "schema"):
93 assert callable(getattr(plugin, method, None)), f"missing: {method}"
94
95 def test_has_merge_ops(self, plugin):
96 assert callable(getattr(plugin, "merge_ops", None))
97
98
99 # ── drift() ───────────────────────────────────────────────────────────────────
100
101 class TestDrift:
102 def test_no_drift_when_tree_matches_commit(self, plugin, tmp_path):
103 snap = plugin.snapshot(tmp_path)
104 report = plugin.drift(snap, tmp_path)
105 assert report.has_drift is False
106
107 def test_drift_detected_on_new_file(self, plugin, tmp_path):
108 snap = plugin.snapshot(tmp_path)
109 (tmp_path / "identities").mkdir()
110 (tmp_path / "identities" / "gabriel.json").write_bytes(b'{"handle":"gabriel"}')
111 report = plugin.drift(snap, tmp_path)
112 assert report.has_drift is True
113
114 def test_drift_report_has_summary(self, plugin, tmp_path):
115 snap = plugin.snapshot(tmp_path)
116 (tmp_path / "identities").mkdir()
117 (tmp_path / "identities" / "alice.json").write_bytes(b'{"handle":"alice"}')
118 report = plugin.drift(snap, tmp_path)
119 assert hasattr(report, "summary")
120
121 def test_no_drift_summary_indicates_clean(self, plugin, tmp_path):
122 snap = plugin.snapshot(tmp_path)
123 report = plugin.drift(snap, tmp_path)
124 # summary should convey "nothing changed"
125 assert not report.has_drift
126
127
128 # ── apply() ───────────────────────────────────────────────────────────────────
129
130 class TestApply:
131 def test_apply_returns_live_state(self, plugin, tmp_path):
132 snap = plugin.snapshot(tmp_path)
133 delta: StateDelta = {"ops": [], "summary": "", "domain": "identity"}
134 result = plugin.apply(delta, tmp_path)
135 assert result == tmp_path
136
137 def test_apply_with_manifest_returns_manifest(self, plugin):
138 manifest = SnapshotManifest(
139 files={"identities/alice.json": "sha256:" + "a" * 64},
140 domain="identity",
141 directories=[],
142 )
143 delta: StateDelta = {"ops": [], "summary": "", "domain": "identity"}
144 result = plugin.apply(delta, manifest)
145 assert result == manifest
File History 1 commit
sha256:b636f72dcba9e190afb980bece906fa5b717fbde014b76ef023df8cb96e01eb9 docs: expand cache plan with all seven testing tiers and do… Sonnet 4.6 134 days ago