generated from coulomb/repo-seed
tegwick
ff96ee0c48
engine/adapter.py: EngineShardAdapter implements adapters.ShardAdapter (read/write run extension transform hooks; profile derived from active extensions, E-5; current_rev for apply-under-drift) + build_engine_shard() helper (explicit ids or activation provider). runtime.available() added. Engine shard passes assert_conformant and attaches to an InformationSpace — resolve + edit (overlay->apply->write-through) work, and the declared profile reflects the active extensions. 5 tests green, pyflakes clean. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
81 lines
2.7 KiB
Python
81 lines
2.7 KiB
Python
"""Tests for EngineShardAdapter (SHARD-WP-0014 T5): engine as a canonical-mode shard."""
|
|
|
|
from shard_wiki import InformationSpace
|
|
from shard_wiki.adapters import assert_conformant
|
|
from shard_wiki.engine import (
|
|
Extension,
|
|
ExtensionRuntime,
|
|
Hook,
|
|
ProfileContribution,
|
|
StaticProvider,
|
|
build_engine_shard,
|
|
)
|
|
from shard_wiki.engine.activation import ActivationContext
|
|
from shard_wiki.model import Verb
|
|
|
|
|
|
class StructProfileExt(Extension):
|
|
"""Profile-only extension (no body transform → write stays content-preserving)."""
|
|
|
|
id = "ext.struct"
|
|
|
|
def hooks(self):
|
|
return {
|
|
Hook.ON_PROFILE: lambda p, c: ProfileContribution(
|
|
verbs_add=frozenset({Verb.STRUCTURED_PAYLOAD})
|
|
)
|
|
}
|
|
|
|
|
|
def _runtime():
|
|
rt = ExtensionRuntime()
|
|
rt.register(StructProfileExt())
|
|
return rt
|
|
|
|
|
|
def test_kernel_only_engine_shard_is_conformant():
|
|
shard = build_engine_shard("eng", ExtensionRuntime(), activate=set())
|
|
shard.write("Home", "hi")
|
|
report = assert_conformant(shard) # read + positive write probe
|
|
assert report.ok
|
|
assert shard.profile().supports(Verb.WRITE)
|
|
assert not shard.profile().supports(Verb.STRUCTURED_PAYLOAD)
|
|
|
|
|
|
def test_profile_reflects_activated_extensions():
|
|
off = build_engine_shard("a", _runtime(), activate=set())
|
|
on = build_engine_shard("b", _runtime(), activate={"ext.struct"})
|
|
assert not off.profile().supports(Verb.STRUCTURED_PAYLOAD)
|
|
assert on.profile().supports(Verb.STRUCTURED_PAYLOAD) # E-5
|
|
assert_conformant(on)
|
|
|
|
|
|
def test_activation_via_provider():
|
|
provider = StaticProvider(flags={"ext.struct": True})
|
|
shard = build_engine_shard("c", _runtime(), provider=provider, context=ActivationContext("c"))
|
|
assert shard.profile().supports(Verb.STRUCTURED_PAYLOAD)
|
|
|
|
|
|
def test_attach_resolve_edit_through_engine_shard(tmp_path):
|
|
space = InformationSpace("team")
|
|
space.attach(build_engine_shard("wikiE", ExtensionRuntime(), activate=set()))
|
|
# seed a page directly via the shard, then read + edit through the orchestrator
|
|
space.union.shard("wikiE").write("Home", "v1")
|
|
assert space.read("Home").body == "v1"
|
|
result = space.edit("Home", "v2") # overlay -> apply-under-drift -> write-through
|
|
assert result.status.value == "applied"
|
|
assert space.read("Home").body == "v2"
|
|
|
|
|
|
def test_on_write_transform_runs():
|
|
class Upper(Extension):
|
|
id = "ext.upper"
|
|
def hooks(self):
|
|
return {Hook.ON_WRITE: lambda body, ctx: body.upper()}
|
|
|
|
rt = ExtensionRuntime()
|
|
rt.register(Upper())
|
|
shard = build_engine_shard("u", rt, activate={"ext.upper"})
|
|
shard.write("P", "quiet")
|
|
assert shard.read("P").body == "QUIET" # extension transformed the write
|