generated from coulomb/repo-seed
tegwick
34b0c539f3
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>
68 lines
2.5 KiB
Python
68 lines
2.5 KiB
Python
"""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
|