GameCult/CultLib
Watch
9
0
Fork
You've already forked CultLib
 0
Code Issues Pull requests 1 Projects Releases Packages Wiki Activity Actions

Python and TypeScript: tests for the store's publishing rule and the admission stamp; cultnet-py keeps its payload-based resolution

Browse source 
cultnet-py resolves a raw record whose schema id no document lists by the schema version its payload opens with, and three
cultmesh-py tests (foreign UI-generated ids) depend on it, so that resolution is restored pending a ruling. The compatible-id
replication test goes through the real apply_raw_document_record and the reopened store.

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
This commit is contained in:
Meta Vens 2026-09-30 07:28:17 +02:00
commit 27edda82ea
4 changed files with 190 additions and 71 deletions
Download patch file Download diff file Expand all files Collapse all files

29
packages/cultcache-py/tests/test_cultcache.py
View file

@ -287,6 +287,34 @@ class CultCacheTests(unittest.TestCase):
reopened.put(document, "next", {"name": "next"})
self.assertEqual(self._open_foreign_id_store(document, path).get_required(document, "next")["name"], "next")
def test_the_store_publishes_a_records_schema_id_even_when_the_envelopes_catalog_entry_does_not(self) -> None:
import msgpack # type: ignore
from cultcache_py import CultCacheSchemaCatalogEntry
with tempfile.TemporaryDirectory() as tmp:
path = Path(tmp) / "store.msgpack"
store = SingleFileMessagePackBackingStore(path)
store.push(CultCacheEnvelope(
key="k",
type="tests.store-publishes",
payload=msgpack.packb({"name": "k"}, use_bin_type=True),
stored_at="2026-09-30T00:00:00Z",
schema_id="tests.store-publishes.record",
catalog_entry=CultCacheSchemaCatalogEntry(
schema_id="tests.store-publishes.other",
schema_name="tests.store-publishes",
schema_version="tests.store_publishes.v1",
content_hash="tests.store-publishes.other",
canonical_schema_json="",
compatible_schema_ids=("tests.store-publishes.other",),
members=(),
),
))
stored = msgpack.unpackb(path.read_bytes(), raw=False)
self.assertEqual(stored[2][0][1], "tests.store-publishes.record")
self.assertEqual([entry[0] for entry in stored[1]], ["tests.store-publishes.record"])
self.assertEqual(len(SingleFileMessagePackBackingStore(path).pull_all()), 1)
def test_a_record_loaded_under_a_compatible_schema_id_is_written_back_under_the_id_its_catalog_entry_carries(self) -> None:
import msgpack # type: ignore
@ -300,6 +328,7 @@ class CultCacheTests(unittest.TestCase):
[["old", "tests.foreign-id.older", "2026-09-30T00:00:00Z", document.encode_payload({"name": "old"})]],
], use_bin_type=True))
cache = self._open_foreign_id_store(document, path)
self.assertEqual(cache.get_required_envelope(document, "old").schema_id, "tests.foreign-id.current")
cache.put(document, "next", {"name": "next"})
reopened = self._open_foreign_id_store(document, path)
self.assertEqual(reopened.get_required(document, "old")["name"], "old")

24
packages/cultcache-ts/test/cult-cache.test.ts
View file

@ -1321,6 +1321,29 @@ test("putEnvelope under a compatible schema id writes a store that reopens and t
assert.deepEqual((await reopenForeignIdStore(file)).getRequired(foreignIdDocument, "next"), { name: "next" });
});
test("the store publishes a record's schema id even when the envelope's catalog entry does not", async () => {
const file = join(await mkdtemp(join(tmpdir(), "cultcache-store-publishes-")), "store.msgpack");
const store = new SingleFileMessagePackBackingStore(file);
await store.push({
key: "k",
type: "tests.store-publishes",
schemaId: "tests.store-publishes.record",
payload: encode({ name: "k" }),
storedAt: "2026-09-30T00:00:00.0000000Z",
catalogEntry: {
schemaId: "tests.store-publishes.other",
schemaName: "tests.store-publishes",
schemaVersion: "tests.store_publishes.v1",
contentHash: "tests.store-publishes.other",
canonicalSchemaJson: "",
compatibleSchemaIds: ["tests.store-publishes.other"],
members: [],
},
});
assert.deepEqual(schemaIdsOf(await readFile(file)), { record: "tests.store-publishes.record", catalog: ["tests.store-publishes.record"] });
assert.equal((await new SingleFileMessagePackBackingStore(file).pullAll()).length, 1);
});
test("a record loaded under a compatible schema id is written back under the id its catalog entry carries", async () => {
const file = join(await mkdtemp(join(tmpdir(), "cultcache-foreign-id-")), "store.msgpack");
const payload = encode({ name: "old" });
@ -1342,6 +1365,7 @@ test("a record loaded under a compatible schema id is written back under the id
);
const cache = await reopenForeignIdStore(file);
assert.equal(cache.getRequiredEnvelope(foreignIdDocument, "old").schemaId, "tests.foreign-id.current");
await cache.put(foreignIdDocument, "next", { name: "next" });
const reopened = await reopenForeignIdStore(file);
assert.deepEqual(reopened.getRequired(foreignIdDocument, "old"), { name: "old" });

29
packages/cultnet-py/src/cultnet_py/replication.py
View file

@ -175,4 +175,33 @@ def resolve_document_and_schema_id_for_raw_record(
document = documents_by_schema_id.get(schema_id)
if document is not None:
return document, schema_id
schema_version = _infer_schema_version_from_payload(bytes(record["payload"]))
schema_name = _infer_schema_name(schema_version) if schema_version is not None else None
if schema_name is not None and schema_name in documents_by_schema_id:
document = documents_by_schema_id[schema_name]
return document, document.catalog_entry().schema_id
raise KeyError(schema_id)
def _infer_schema_version_from_payload(payload: bytes) -> str | None:
try:
import msgpack # type: ignore
decoded = msgpack.unpackb(payload, raw=False)
except Exception:
return None
if isinstance(decoded, list) and decoded and isinstance(decoded[0], str):
return decoded[0]
if isinstance(decoded, dict):
value = decoded.get("schemaVersion", decoded.get("schema_version"))
return value if isinstance(value, str) else None
return None
def _infer_schema_name(schema_version: str) -> str | None:
marker = schema_version.rfind(".v")
if marker <= 0:
return None
version = schema_version[marker + 2:]
if not version or not version.isdigit():
return None
return schema_version[:marker]

177
packages/cultnet-py/tests/test_cultnet.py
View file

@ -1682,12 +1682,7 @@ class CultNetTests(unittest.TestCase):
self.assertEqual([change.change_kind for change in typed_applied_log], ["updated", "removed"])
self.assertIsNone(cache.get(document, "item:1"))
def test_cultnet_replication_refuses_an_unknown_schema_id_and_stamps_a_compatible_one(self) -> None:
import tempfile
from pathlib import Path
from cultcache_py import SingleFileMessagePackBackingStore
def test_cultnet_replication_helpers_recover_schema_stamped_raw_records(self) -> None:
document = define_database_entry_type(
"replica.runtime-policy",
[
@ -1698,52 +1693,120 @@ class CultNetTests(unittest.TestCase):
schema_id="replica.runtime-policy.current",
schema_name="replica.runtime_policy",
schema_version="replica.runtime_policy.v1",
compatible_schema_ids=["replica.runtime-policy.current", "replica.runtime-policy.older"],
)
cache = CultCache()
cache.register_document_type(document)
stale_schema_id = "sha256:stale-replica-runtime-policy"
payload = document.encode_payload({
"schema_version": "replica.runtime_policy.v1",
"name": "policy",
"value": "synced",
"value": "still synced",
})
record = {
"schemaId": stale_schema_id,
"recordKey": "policy:1",
"storedAt": "2026-06-25T00:00:00Z",
"payloadEncoding": "messagepack",
"payload": payload,
}
snapshot = {
"schemaVersion": "cultnet.snapshot_response_raw.v0",
"messageId": "snapshot-stale",
"documents": [record],
}
def raw(schema_id: str, key: str) -> dict:
return {
"schemaId": schema_id,
"recordKey": key,
"storedAt": "2026-06-25T00:00:00Z",
"payloadEncoding": "messagepack",
"payload": payload,
}
applied_snapshot = apply_raw_snapshot(cache, [document], snapshot)
local_schema_id = document.catalog_entry().schema_id
self.assertEqual(applied_snapshot[0].schema_id, local_schema_id)
self.assertEqual(cache.get_required(document, "policy:1")["value"], "still synced")
self.assertEqual(cache.get_required_envelope(document, "policy:1").schema_id, local_schema_id)
direct_applied = apply_raw_document_record(cache, schema_document_map([document]), record)
self.assertEqual(direct_applied.schema_id, local_schema_id)
self.assertEqual(cache.get_required_envelope(document, "policy:1").schema_id, local_schema_id)
documents = schema_document_map([document])
# A schema id no binding lists is refused whatever the payload opens with; the payload does not name a schema.
stale = raw("sha256:stale-replica-runtime-policy", "policy:stale")
with self.assertRaises(KeyError):
apply_raw_document_record(CultCache(), documents, stale)
shard_log = {
"schemaVersion": "cultnet.shard_log_response.v0",
"messageId": "log-stale",
"shardId": "interop",
"shardEpoch": 1,
"resyncRequired": False,
"entries": [{
"sequence": 0,
"changeKind": "updated",
"put": {
"schemaVersion": "cultnet.document_put_raw.v0",
"messageId": "put-stale",
"document": stale,
"shardId": "interop",
"shardEpoch": 1,
},
}],
"entries": [
{
"sequence": 0,
"changeKind": "updated",
"put": {
"schemaVersion": "cultnet.document_put_raw.v0",
"messageId": "put-stale",
"document": {
**record,
"payload": document.encode_payload({
"schema_version": "replica.runtime_policy.v1",
"name": "policy",
"value": "updated through log",
}),
},
"shardId": "interop",
"shardEpoch": 1,
},
}
],
}
stale_cache = CultCache()
stale_cache.register_document_type(document)
self.assertEqual(apply_shard_log_response(stale_cache, [document], shard_log), [])
# A compatible id applies, the record is stamped with the registered id, and the store it leaves reopens.
local_schema_id = document.catalog_entry().schema_id
applied_log = apply_shard_log_response(cache, [document], shard_log)
self.assertEqual(applied_log[0].schema_id, local_schema_id)
self.assertEqual(cache.get_required(document, "policy:1")["value"], "updated through log")
self.assertEqual(cache.get_required_envelope(document, "policy:1").schema_id, local_schema_id)
delete_log = {
"schemaVersion": "cultnet.shard_log_response.v0",
"messageId": "log-delete-alias",
"shardId": "interop",
"shardEpoch": 1,
"resyncRequired": False,
"entries": [
{
"sequence": 1,
"changeKind": "removed",
"delete": {
"schemaVersion": "cultnet.document_delete.v0",
"messageId": "delete-version-alias",
"schemaId": "replica.runtime_policy.v1",
"recordKey": "policy:1",
"shardId": "interop",
"shardEpoch": 1,
},
}
],
}
applied_delete = apply_shard_log_response(cache, [document], delete_log)
self.assertEqual(applied_delete[0].schema_id, local_schema_id)
self.assertEqual(applied_delete[0].change_kind, "removed")
self.assertIsNone(cache.get(document, "policy:1"))
# A compatible schema id applies through the real replication function, the record is stamped with the registered id, and
# the store it leaves reopens and takes the next write.
def test_cultnet_raw_replication_under_a_compatible_schema_id_leaves_a_store_that_reopens(self) -> None:
import tempfile
from pathlib import Path
from cultcache_py import SingleFileMessagePackBackingStore
document = define_database_entry_type(
"replica.compatible-policy",
[("schema_version", 0), ("name", 1)],
schema_id="replica.compatible-policy.current",
schema_name="replica.compatible_policy",
schema_version="replica.compatible_policy.v1",
compatible_schema_ids=["replica.compatible-policy.current", "replica.compatible-policy.older"],
)
value = {"schema_version": "replica.compatible_policy.v1", "name": "synced"}
record = {
"schemaId": "replica.compatible-policy.older",
"recordKey": "policy:1",
"storedAt": "2026-06-25T00:00:00Z",
"payloadEncoding": "messagepack",
"payload": document.encode_payload(value),
}
with tempfile.TemporaryDirectory() as tmp:
path = Path(tmp) / "store.msgpack"
@ -1755,40 +1818,14 @@ class CultNetTests(unittest.TestCase):
return cache
cache = open_cache()
applied = apply_raw_document_record(cache, documents, raw("replica.runtime-policy.older", "policy:1"))
self.assertEqual(applied.record_key, "policy:1")
self.assertEqual(cache.get_required_envelope(document, "policy:1").schema_id, local_schema_id)
apply_raw_document_record(cache, schema_document_map([document]), record)
self.assertEqual(
cache.get_required_envelope(document, "policy:1").schema_id, document.catalog_entry().schema_id
)
reopened = open_cache()
self.assertEqual(reopened.get_required(document, "policy:1")["value"], "synced")
reopened.put(document, "policy:2", {"schema_version": "replica.runtime_policy.v1", "name": "n", "value": "next"})
self.assertEqual(open_cache().get_required(document, "policy:2")["value"], "next")
delete_log = {
"schemaVersion": "cultnet.shard_log_response.v0",
"messageId": "log-delete-alias",
"shardId": "interop",
"shardEpoch": 1,
"resyncRequired": False,
"entries": [{
"sequence": 1,
"changeKind": "removed",
"delete": {
"schemaVersion": "cultnet.document_delete.v0",
"messageId": "delete-version-alias",
"schemaId": "replica.runtime_policy.v1",
"recordKey": "policy:1",
"shardId": "interop",
"shardEpoch": 1,
},
}],
}
live = CultCache()
live.register_document_type(document)
live.put(document, "policy:1", {"schema_version": "replica.runtime_policy.v1", "name": "n", "value": "v"})
applied_delete = apply_shard_log_response(live, [document], delete_log)
self.assertEqual(applied_delete[0].schema_id, local_schema_id)
self.assertEqual(applied_delete[0].change_kind, "removed")
self.assertIsNone(live.get(document, "policy:1"))
self.assertEqual(reopened.get_required(document, "policy:1")["name"], "synced")
reopened.put(document, "policy:2", {**value, "name": "next"})
self.assertEqual(open_cache().get_required(document, "policy:2")["name"], "next")
def test_cultnet_document_delete_helper_matches_schema_v0_shape(self) -> None:
message = document_delete(