generated from coulomb/repo-seed
feat(coordination): apply-under-drift (WP-0008 T4)
OverlayEngine.apply: read-only target → KEPT_DRAFT; base_rev==current → fast-forward write-through (APPLIED, MERGE_DECIDED closes overlay); drift → REFUSED_DRIFT (no clobber, I-5). 5 tests green, pyflakes clean. (blueprint §8.6) Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
66
tests/test_apply.py
Normal file
66
tests/test_apply.py
Normal file
@@ -0,0 +1,66 @@
|
||||
"""Tests for apply-under-drift (SHARD-WP-0008 T4)."""
|
||||
|
||||
from shard_wiki.adapters import FolderAdapter
|
||||
from shard_wiki.coordination import ApplyStatus, DecisionLog, OverlayEngine
|
||||
from shard_wiki.model import Identity
|
||||
|
||||
|
||||
def _writable(tmp_path, files):
|
||||
for rel, text in files.items():
|
||||
(tmp_path / rel).write_text(text, encoding="utf-8")
|
||||
return FolderAdapter("w", tmp_path, writable=True)
|
||||
|
||||
|
||||
def test_fast_forward_apply_writes_through(tmp_path):
|
||||
shard = _writable(tmp_path, {"Home.md": "old"})
|
||||
eng = OverlayEngine("space", DecisionLog())
|
||||
base = shard.current_rev("Home")
|
||||
ov = eng.draft(Identity("w", "Home"), "new", base_rev=base)
|
||||
|
||||
result = eng.apply(ov.overlay_id, shard)
|
||||
assert result.status is ApplyStatus.APPLIED
|
||||
assert shard.read("Home").body == "new" # written through
|
||||
assert eng.get(ov.overlay_id) is None # overlay closed in the fold
|
||||
|
||||
|
||||
def test_drift_refuses_without_clobber(tmp_path):
|
||||
shard = _writable(tmp_path, {"Home.md": "old"})
|
||||
eng = OverlayEngine("space", DecisionLog())
|
||||
ov = eng.draft(Identity("w", "Home"), "mine", base_rev="STALE-REV")
|
||||
|
||||
result = eng.apply(ov.overlay_id, shard)
|
||||
assert result.status is ApplyStatus.REFUSED_DRIFT
|
||||
assert shard.read("Home").body == "old" # not clobbered (I-5)
|
||||
assert eng.get(ov.overlay_id) is not None # overlay still open
|
||||
|
||||
|
||||
def test_read_only_target_keeps_draft(tmp_path):
|
||||
(tmp_path / "Home.md").write_text("canon", encoding="utf-8")
|
||||
ro = FolderAdapter("ro", tmp_path) # not writable
|
||||
eng = OverlayEngine("space", DecisionLog())
|
||||
ov = eng.draft(Identity("ro", "Home"), "my edit", base_rev=ro.current_rev("Home"))
|
||||
|
||||
result = eng.apply(ov.overlay_id, ro)
|
||||
assert result.status is ApplyStatus.KEPT_DRAFT
|
||||
assert ro.read("Home").body == "canon" # source untouched
|
||||
assert eng.get(ov.overlay_id) is not None # local truth retained
|
||||
|
||||
|
||||
def test_new_page_fast_forwards(tmp_path):
|
||||
shard = _writable(tmp_path, {})
|
||||
eng = OverlayEngine("space", DecisionLog())
|
||||
ov = eng.draft(Identity("w", "Fresh"), "brand new", base_rev=None) # didn't exist
|
||||
result = eng.apply(ov.overlay_id, shard)
|
||||
assert result.status is ApplyStatus.APPLIED
|
||||
assert shard.read("Fresh").body == "brand new"
|
||||
|
||||
|
||||
def test_wrong_adapter_is_rejected(tmp_path):
|
||||
shard = _writable(tmp_path, {"Home.md": "x"})
|
||||
eng = OverlayEngine("space", DecisionLog())
|
||||
ov = eng.draft(Identity("other", "Home"), "y", base_rev=None)
|
||||
try:
|
||||
eng.apply(ov.overlay_id, shard)
|
||||
raise AssertionError("expected ValueError")
|
||||
except ValueError:
|
||||
pass
|
||||
Reference in New Issue
Block a user