75 lines
3.1 KiB
Python
75 lines
3.1 KiB
Python
|
"""Wire the incremental tier behind InformationSpace views (SHARD-WP-0011 T4)."""
|
||
|
|
|
||
|
|
from shard_wiki.adapters import FolderAdapter
|
||
|
|
from shard_wiki.coordination import EventType
|
||
|
|
from shard_wiki.model import Identity
|
||
|
|
from shard_wiki.space import InformationSpace
|
||
|
|
from shard_wiki.views import all_pages
|
||
|
|
|
||
|
|
|
||
|
|
def _shard(tmp_path, name, files):
|
||
|
|
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")
|
||
|
|
return FolderAdapter(name, root)
|
||
|
|
|
||
|
|
|
||
|
|
def test_all_pages_via_index_matches_direct_fold(tmp_path):
|
||
|
|
space = InformationSpace("space")
|
||
|
|
space.attach(_shard(tmp_path, "wiki", {"Home.md": "welcome", "Guide.md": "the guide"}))
|
||
|
|
space.attach(_shard(tmp_path, "notes", {"Daily.md": "today"}))
|
||
|
|
# Routed-through-index result equals the direct fold-based computation (behaviour unchanged).
|
||
|
|
via_index = {(e.name, e.members) for e in space.all_pages()}
|
||
|
|
direct = {(e.name, e.members) for e in all_pages(space.union)}
|
||
|
|
assert via_index == direct
|
||
|
|
|
||
|
|
|
||
|
|
def test_curator_binding_collapses_via_maintained_index(tmp_path):
|
||
|
|
space = InformationSpace("space")
|
||
|
|
space.attach(_shard(tmp_path, "a", {"Foo.md": "x"}))
|
||
|
|
space.attach(_shard(tmp_path, "b", {"Bar.md": "y"}))
|
||
|
|
space.log.append(
|
||
|
|
"space", EventType.BINDING_MADE, {"members": ["a:Foo", "b:Bar"]}
|
||
|
|
)
|
||
|
|
# The maintained index re-syncs curator edges live from the log fold.
|
||
|
|
collapsed = [e for e in space.all_pages() if len(e.members) == 2]
|
||
|
|
assert len(collapsed) == 1
|
||
|
|
assert set(collapsed[0].members) == {Identity("a", "Foo"), Identity("b", "Bar")}
|
||
|
|
|
||
|
|
|
||
|
|
def test_content_duplicate_collapses_via_index(tmp_path):
|
||
|
|
space = InformationSpace("space")
|
||
|
|
space.attach(_shard(tmp_path, "a", {"Foo.md": "the very same body content here"}))
|
||
|
|
space.attach(_shard(tmp_path, "b", {"Bar.md": "the very same body content here"}))
|
||
|
|
dup = [e for e in space.all_pages() if len(e.members) == 2]
|
||
|
|
assert len(dup) == 1 # content equivalence detected by the maintained index
|
||
|
|
assert set(dup[0].members) == {Identity("a", "Foo"), Identity("b", "Bar")}
|
||
|
|
|
||
|
|
|
||
|
|
def test_attach_invalidates_index(tmp_path):
|
||
|
|
space = InformationSpace("space")
|
||
|
|
space.attach(_shard(tmp_path, "a", {"Foo.md": "same body"}))
|
||
|
|
assert space.all_pages() # builds the index (one page, no groups)
|
||
|
|
space.attach(_shard(tmp_path, "b", {"Bar.md": "same body"})) # marks index stale
|
||
|
|
dup = [e for e in space.all_pages() if len(e.members) == 2]
|
||
|
|
assert len(dup) == 1 # rebuilt fallback picks up the new equivalent page
|
||
|
|
|
||
|
|
|
||
|
|
def test_verify_index_reports_healthy_when_consistent(tmp_path):
|
||
|
|
space = InformationSpace("space")
|
||
|
|
space.attach(_shard(tmp_path, "a", {"Foo.md": "same body"}))
|
||
|
|
space.attach(_shard(tmp_path, "b", {"Bar.md": "same body"}))
|
||
|
|
space.all_pages() # ensure built
|
||
|
|
report = space.verify_index()
|
||
|
|
assert report.healthy is True
|
||
|
|
|
||
|
|
|
||
|
|
def test_reindex_is_an_explicit_fallback(tmp_path):
|
||
|
|
space = InformationSpace("space")
|
||
|
|
space.attach(_shard(tmp_path, "a", {"Foo.md": "content"}))
|
||
|
|
before = space.index.digest()
|
||
|
|
space.reindex()
|
||
|
|
assert space.index.digest() == before # rebuild is deterministic
|