generated from coulomb/repo-seed
feat(views): RecentChanges merged change feed (WP-0010 T3)
One newest-first feed merging the coordination journal (overlay/alias/fork/merge/ binding decisions, with actor + payload) and shard change signals (page source_rev / mtime). Each entry carries provenance: the originating shard for an edit, or 'coordination' (and the actor) for a decision. Non-temporal revision tokens are skipped gracefully. Derived/recomputable; notify-streaming later. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
67
tests/test_views_recentchanges.py
Normal file
67
tests/test_views_recentchanges.py
Normal file
@@ -0,0 +1,67 @@
|
||||
"""Tests for the RecentChanges merged feed (SHARD-WP-0010 T3)."""
|
||||
|
||||
import os
|
||||
from datetime import datetime, timezone
|
||||
|
||||
from shard_wiki.adapters import FolderAdapter
|
||||
from shard_wiki.coordination import DecisionLog, EventType
|
||||
from shard_wiki.union import UnionGraph
|
||||
from shard_wiki.views import recent_changes
|
||||
|
||||
|
||||
def _shard(tmp_path, name, files, mtime=None):
|
||||
root = tmp_path / name
|
||||
for rel, text in files.items():
|
||||
p = root / rel
|
||||
p.parent.mkdir(parents=True, exist_ok=True)
|
||||
p.write_text(text, encoding="utf-8")
|
||||
if mtime is not None:
|
||||
os.utime(p, (mtime, mtime))
|
||||
return FolderAdapter(name, root)
|
||||
|
||||
|
||||
def test_edit_and_alias_both_appear_newest_first(tmp_path):
|
||||
# Page edit signal pinned to an old mtime; the alias decision happens "now" → alias is newest.
|
||||
old = datetime(2020, 1, 1, tzinfo=timezone.utc).timestamp()
|
||||
u = UnionGraph("space")
|
||||
u.attach(_shard(tmp_path, "shardA", {"Home.md": "home"}, mtime=old))
|
||||
log = DecisionLog()
|
||||
log.append("space", EventType.ALIAS_SET, {"alias": "Start", "target": "shardA:Home"})
|
||||
|
||||
feed = recent_changes(u, log, "space")
|
||||
kinds = [e.kind for e in feed]
|
||||
assert "edit" in kinds and "alias" in kinds
|
||||
assert feed[0].kind == "alias" # newest first
|
||||
assert feed[-1].kind == "edit"
|
||||
# Monotonic non-increasing by time.
|
||||
assert all(feed[i].when >= feed[i + 1].when for i in range(len(feed) - 1))
|
||||
|
||||
|
||||
def test_per_shard_attribution_present(tmp_path):
|
||||
u = UnionGraph("space")
|
||||
u.attach(_shard(tmp_path, "shardA", {"A.md": "a"}))
|
||||
u.attach(_shard(tmp_path, "shardB", {"B.md": "b"}))
|
||||
feed = recent_changes(u, DecisionLog(), "space")
|
||||
edits = {e.ref: e.source for e in feed if e.kind == "edit"}
|
||||
assert edits["shardA:A"] == "shardA"
|
||||
assert edits["shardB:B"] == "shardB" # each edit attributed to its shard
|
||||
|
||||
|
||||
def test_coordination_entries_carry_actor_and_ref(tmp_path):
|
||||
u = UnionGraph("space")
|
||||
u.attach(_shard(tmp_path, "shardA", {"Doc.md": "x"}))
|
||||
log = DecisionLog()
|
||||
log.append(
|
||||
"space", EventType.PAGE_FORKED, {"source": "shardA:Doc", "fork": "shardB:Doc"}, actor="ana"
|
||||
)
|
||||
fork = next(e for e in recent_changes(u, log, "space") if e.kind == "fork")
|
||||
assert fork.source == "coordination"
|
||||
assert fork.actor == "ana"
|
||||
assert fork.ref == "shardA:Doc→shardB:Doc"
|
||||
|
||||
|
||||
def test_limit_truncates_to_newest(tmp_path):
|
||||
u = UnionGraph("space")
|
||||
u.attach(_shard(tmp_path, "shardA", {"A.md": "a", "B.md": "b", "C.md": "c"}))
|
||||
feed = recent_changes(u, DecisionLog(), "space", limit=2)
|
||||
assert len(feed) == 2
|
||||
Reference in New Issue
Block a user