diff --git a/packages/cultcache-py/src/cultcache_py/__init__.py b/packages/cultcache-py/src/cultcache_py/__init__.py index 07721ef4..ad00a74b 100644 --- a/packages/cultcache-py/src/cultcache_py/__init__.py +++ b/packages/cultcache-py/src/cultcache_py/__init__.py @@ -4,10 +4,14 @@ from .cache import CultCache, CultCacheBuilder from .documents import ( DatabaseEntryField, DocumentDefinition, + SlotCodec, + UnionArm, database_entry_field, define_database_entry_type, define_document_registry, define_document_type, + define_record_codec, + define_union, ) from .stores import JsonLinesBackingStore, SingleFileMessagePackBackingStore @@ -22,8 +26,12 @@ __all__ = [ "DocumentDefinition", "JsonLinesBackingStore", "SingleFileMessagePackBackingStore", + "SlotCodec", + "UnionArm", "database_entry_field", "define_database_entry_type", "define_document_registry", "define_document_type", + "define_record_codec", + "define_union", ] diff --git a/packages/cultcache-py/src/cultcache_py/documents.py b/packages/cultcache-py/src/cultcache_py/documents.py index e8511249..382ddfd3 100644 --- a/packages/cultcache-py/src/cultcache_py/documents.py +++ b/packages/cultcache-py/src/cultcache_py/documents.py @@ -2,7 +2,7 @@ import json from dataclasses import MISSING, dataclass, field -from typing import Any, Callable, Generic, Iterable, TypeVar +from typing import Any, Callable, Generic, Iterable, Mapping, TypeVar from .backing_store import CultCacheSchemaCatalogEntry, CultCacheSchemaCatalogMember @@ -58,17 +58,46 @@ def _messagepack_payload_decoder(payload: bytes) -> Any: return msgpack.unpackb(payload, raw=False) +@dataclass(frozen=True) +class SlotCodec: + """Encodes and decodes the value of one slot: a nested record or a union.""" + + encode: Callable[[Any], Any] + decode: Callable[[Any], Any] + + @dataclass(frozen=True) class DatabaseEntryField: name: str key: int default: Any = MISSING + codec: SlotCodec | Callable[[], SlotCodec] | None = None + many: bool = False -def database_entry_field(name: str, key: int, *, default: Any = MISSING) -> DatabaseEntryField: +def database_entry_field( + name: str, + key: int, + *, + default: Any = MISSING, + codec: SlotCodec | Callable[[], SlotCodec] | None = None, + many: bool = False, +) -> DatabaseEntryField: + """A slot. `codec` makes the slot a nested record or union (a zero-argument callable + returning the codec lets a union refer to itself); `many` makes it a list of them.""" if key < 0: raise ValueError("DatabaseEntry field keys must be non-negative integers") - return DatabaseEntryField(name=name, key=key, default=default) + if many and codec is None: + raise ValueError("A many DatabaseEntry field requires a codec") + return DatabaseEntryField(name=name, key=key, default=default, codec=codec, many=many) + + +@dataclass(frozen=True) +class UnionArm: + """A decoded union value: the arm's declared name and its decoded record (None for a unit arm).""" + + name: str + value: Any = None @dataclass(frozen=True) @@ -205,14 +234,7 @@ def define_database_entry_type( and newly added fields can use higher keys without rewriting older payloads. """ - normalized = [_normalize_database_entry_field(field) for field in fields] - if not normalized: - raise ValueError("DatabaseEntry document types require at least one field") - keys = [field.key for field in normalized] - if len(keys) != len(set(keys)): - raise ValueError(f"DatabaseEntry document type {type!r} has duplicate field keys") - normalized.sort(key=lambda field: field.key) - max_key = max(keys) + normalized = _normalize_fields(type, fields) member_name_lookup = name if isinstance(name, str) else None index_lookup = indexes or {} members = tuple( @@ -229,36 +251,12 @@ def define_database_entry_type( for field in normalized ) - def encode(value: Any) -> list[Any]: - slots = [None] * (max_key + 1) - for field in normalized: - slots[field.key] = extract_value(value, field.name) - return slots - - def decode(raw: Any) -> Any: - if not isinstance(raw, list): - raise ValueError(f"DatabaseEntry document type {type!r} expected a MessagePack array") - values: dict[str, Any] = {} - for field in normalized: - if field.key < len(raw): - value = raw[field.key] - elif field.default is not MISSING: - value = field.default() if callable(field.default) else field.default - else: - raise ValueError( - f"DatabaseEntry document type {type!r} is missing required slot {field.key}" - ) - if value is None and field.default is not MISSING: - value = field.default() if callable(field.default) else field.default - values[field.name] = value - if cls is None: - return values - return cls(**values) + record = _record_codec(type, normalized, cls) return define_document_type( type, - encode=encode, - decode=decode, + encode=record.encode, + decode=record.decode, validate=validate, global_document=global_document, name=name, @@ -275,6 +273,60 @@ def define_database_entry_type( ) +def define_record_codec( + name: str, + fields: Iterable[DatabaseEntryField | tuple[str, int] | tuple[str, int, Any]], + *, + cls: type[Any] | None = None, +) -> SlotCodec: + """A nested record: the same slot array as a document payload, usable as a field's codec or a union arm.""" + return _record_codec(name, _normalize_fields(name, fields), cls) + + +def define_union( + name: str, + arms: Mapping[int, tuple[str, SlotCodec | Callable[[], SlotCodec] | None]], +) -> SlotCodec: + """A union: nil, or `[armKey, [armSlots...]]`. Each arm is `(arm name, record codec)`; a unit arm has + no codec and is written `[armKey, []]`. A key the union does not declare is refused, never carried.""" + by_key = dict(arms) + by_name = {arm_name: (key, codec) for key, (arm_name, codec) in by_key.items()} + if not by_key: + raise ValueError(f"Union {name!r} requires at least one arm") + if len(by_name) != len(by_key): + raise ValueError(f"Union {name!r} has duplicate arm names") + if any(not _is_key(key) or key < 0 for key in by_key): + raise ValueError(f"Union {name!r} arm keys must be non-negative integers") + + def encode(value: Any) -> Any: + if value is None: + return None + if not isinstance(value, UnionArm) or value.name not in by_name: + raise ValueError(f"Union {name!r} was given a value that is not one of its arms") + key, codec = by_name[value.name] + if codec is None: + if value.value is not None: + raise ValueError(f"Union {name!r} arm {key} is a unit arm and carries no value") + return [key, []] + return [key, _resolve(codec).encode(value.value)] + + def decode(raw: Any) -> Any: + if raw is None: + return None + if not isinstance(raw, list) or len(raw) != 2 or not isinstance(raw[1], list): + raise ValueError(f"Union {name!r} value is malformed: expected [armKey, [armSlots]]") + key = raw[0] + if not _is_key(key): + raise ValueError(f"Union {name!r} value is malformed: the arm key is not an integer") + if key not in by_key: + declared = ", ".join(str(declared_key) for declared_key in sorted(by_key)) + raise ValueError(f"Union {name!r} does not declare arm key {key} (declared: {declared})") + arm_name, codec = by_key[key] + return UnionArm(arm_name, None if codec is None else _resolve(codec).decode(raw[1])) + + return SlotCodec(encode=encode, decode=decode) + + def define_document_registry(*documents: DocumentDefinition[Any]) -> tuple[DocumentDefinition[Any], ...]: return tuple(documents) @@ -287,6 +339,77 @@ def extract_value(value: Any, extractor: str | Extractor[Any]) -> str | int | fl return getattr(value, extractor, None) +def _is_key(value: Any) -> bool: + return isinstance(value, int) and not isinstance(value, bool) + + +def _resolve(codec: SlotCodec | Callable[[], SlotCodec]) -> SlotCodec: + return codec if isinstance(codec, SlotCodec) else codec() + + +def _normalize_fields( + type: str, + fields: Iterable[DatabaseEntryField | tuple[str, int] | tuple[str, int, Any]], +) -> list[DatabaseEntryField]: + normalized = [_normalize_database_entry_field(field) for field in fields] + if not normalized: + raise ValueError("DatabaseEntry document types require at least one field") + keys = [field.key for field in normalized] + if len(keys) != len(set(keys)): + raise ValueError(f"DatabaseEntry document type {type!r} has duplicate field keys") + return sorted(normalized, key=lambda field: field.key) + + +def _record_codec(type: str, normalized: list[DatabaseEntryField], cls: type[Any] | None) -> SlotCodec: + max_key = normalized[-1].key + + def encode_slot(field: DatabaseEntryField, value: Any) -> Any: + if field.codec is None or value is None: + return value + codec = _resolve(field.codec) + if field.many: + return [None if item is None else codec.encode(item) for item in value] + return codec.encode(value) + + def decode_slot(field: DatabaseEntryField, value: Any) -> Any: + if field.codec is None or value is None: + return value + codec = _resolve(field.codec) + if field.many: + if not isinstance(value, list): + raise ValueError(f"DatabaseEntry document type {type!r} slot {field.key} expected a MessagePack array") + return [None if item is None else codec.decode(item) for item in value] + return codec.decode(value) + + def encode(value: Any) -> list[Any]: + slots = [None] * (max_key + 1) + for field in normalized: + slots[field.key] = encode_slot(field, extract_value(value, field.name)) + return slots + + def decode(raw: Any) -> Any: + if not isinstance(raw, list): + raise ValueError(f"DatabaseEntry document type {type!r} expected a MessagePack array") + values: dict[str, Any] = {} + for field in normalized: + if field.key < len(raw): + value = decode_slot(field, raw[field.key]) + elif field.default is not MISSING: + value = field.default() if callable(field.default) else field.default + else: + raise ValueError( + f"DatabaseEntry document type {type!r} is missing required slot {field.key}" + ) + if value is None and field.default is not MISSING: + value = field.default() if callable(field.default) else field.default + values[field.name] = value + if cls is None: + return values + return cls(**values) + + return SlotCodec(encode=encode, decode=decode) + + def _normalize_database_entry_field( field: DatabaseEntryField | tuple[str, int] | tuple[str, int, Any], ) -> DatabaseEntryField: diff --git a/packages/cultcache-py/tests/test_union_parity.py b/packages/cultcache-py/tests/test_union_parity.py new file mode 100644 index 00000000..a9e88b08 --- /dev/null +++ b/packages/cultcache-py/tests/test_union_parity.py @@ -0,0 +1,267 @@ +from __future__ import annotations + +import shutil +import tempfile +import unittest +from pathlib import Path + +import msgpack + +from cultcache_py import ( + CultCache, + SingleFileMessagePackBackingStore, + UnionArm, + database_entry_field, + define_database_entry_type, + define_record_codec, + define_union, +) + +VECTORS = Path(__file__).resolve().parents[3] / "tests" / "vectors" / "union-parity" + +composition = define_record_codec( + "Composition", [("name", 0), database_entry_field("body", 1, codec=lambda: node)] +) +box = define_record_codec("Box", [("id", 0), database_entry_field("children", 1, codec=lambda: node, many=True)]) +text = define_record_codec("Text", [("id", 0), ("content", 1)]) +use = define_record_codec( + "Use", + [ + ("id", 0), + database_entry_field("composition", 1, codec=composition), + database_entry_field("slots", 2, codec=lambda: node, many=True), + ], +) +image = define_record_codec("Image", [("id", 0), ("uri", 1), ("alt", 3)]) +node = define_union("Node", {0: ("Box", box), 1: ("Text", text), 2: ("Use", use), 5: ("Image", image)}) +outcome = define_union( + "Outcome", + { + 0: ("Pending", define_record_codec("Pending", [("message", 0)])), + 1: ("Completed", define_record_codec("Completed", [("source_version", 0)])), + 2: ("Denied", None), + }, +) +tool = define_record_codec("Tool", [("name", 0), ("uses", 1)]) +part = define_record_codec("Part", [("name", 0), ("count", 1)]) +gem = define_record_codec("Gem", [("name", 0)]) +item = define_union("Item", {0: ("Gem", gem), 1: ("Tool", tool), 2: ("Part", part)}) +crafted = define_union("CraftedItem", {0: ("Tool", tool), 1: ("Part", part)}) + +surface_document = define_database_entry_type( + "tests.union_surface", + [ + ("id", 0), + ("version", 1), + database_entry_field("root", 2, codec=node), + database_entry_field("outcomes", 3, codec=outcome, many=True), + database_entry_field("part", 4, codec=item), + database_entry_field("crafted", 5, codec=crafted), + database_entry_field("missing", 6, codec=item), + ], +) + + +def _surface_value() -> dict: + return { + "id": "surface", + "version": 3, + "root": UnionArm( + "Box", + { + "id": "root", + "children": [ + UnionArm("Text", {"id": "t1", "content": "hello"}), + UnionArm( + "Use", + { + "id": "u1", + "composition": { + "name": "comp", + "body": UnionArm( + "Box", + { + "id": "inner", + "children": [ + UnionArm("Text", {"id": "deep", "content": "x"}), + UnionArm("Image", {"id": "img", "uri": "uri://a", "alt": "alt"}), + ], + }, + ), + }, + "slots": [ + UnionArm("Text", {"id": "s1", "content": "slot"}), + None, + UnionArm("Box", {"id": "b2", "children": []}), + ], + }, + ), + UnionArm("Image", {"id": "i2", "uri": "uri://b", "alt": ""}), + ], + }, + ), + "outcomes": [ + UnionArm("Pending", {"message": "wait"}), + UnionArm("Completed", {"source_version": 7}), + UnionArm("Denied"), + ], + "part": UnionArm("Tool", {"name": "hammer", "uses": 12}), + "crafted": UnionArm("Part", {"name": "gear", "count": 4}), + "missing": None, + } + + +def _vector(name: str) -> bytes: + return (VECTORS / name).read_bytes() + + +class UnionSurfaceVectorTests(unittest.TestCase): + def test_surface_payload_decodes_to_the_value_and_re_encodes_byte_identical(self) -> None: + payload = _vector("union-surface.msgpack") + decoded = surface_document.decode_payload(payload) + self.assertEqual(decoded, _surface_value()) + self.assertEqual(surface_document.encode_payload(decoded), payload) + + def test_the_csharp_written_store_reads_and_the_cache_rewrites_the_same_payload(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + path = Path(tmp) / "union-surface.cc" + shutil.copyfile(VECTORS / "union-surface.cc", path) + cache = ( + CultCache.builder() + .register_document_type(surface_document) + .add_generic_store(SingleFileMessagePackBackingStore(path)) + .build() + ) + cache.pull_all_backing_stores() + self.assertEqual(cache.get(surface_document, "surface"), _surface_value()) + fresh = Path(tmp) / "fresh.cc" + writer = ( + CultCache.builder() + .register_document_type(surface_document) + .add_generic_store(SingleFileMessagePackBackingStore(fresh)) + .build() + ) + writer.pull_all_backing_stores() + writer.put(surface_document, "surface", _surface_value()) + [envelope] = SingleFileMessagePackBackingStore(fresh).pull_all() + self.assertEqual(envelope.payload, _vector("union-surface.msgpack")) + + def test_the_store_vector_payload_equals_the_surface_payload(self) -> None: + [envelope] = SingleFileMessagePackBackingStore(VECTORS / "union-surface.cc").pull_all() + self.assertEqual(envelope.payload, _vector("union-surface.msgpack")) + + def test_every_outcome_in_the_manifest_is_held(self) -> None: + outcomes = msgpack.unpackb(_vector("outcomes.msgpack"), raw=False) + self.assertEqual(len(outcomes), 9) + refused = read_only = 0 + for file, outcome_kind, union, key, detail in outcomes: + payload = _vector(file) + if outcome_kind == "refused": + refused += 1 + with self.assertRaises(ValueError, msg=file) as caught: + surface_document.decode_payload(payload) + message = str(caught.exception) + self.assertIn(union, message, file) + if detail == "unknown-arm": + self.assertIn(f"arm key {key} ", message, file) + else: + self.assertEqual(detail, "arity") + self.assertIn("malformed", message, file) + else: + read_only += 1 + self.assertEqual(outcome_kind, "read-only", file) + self.assertEqual(surface_document.decode_payload(payload), _surface_value(), file) + self.assertEqual((refused, read_only), (5, 4)) + + +class UnionCodecRuleTests(unittest.TestCase): + CANARY = "canary-7f3a91" + + def test_nil_union_is_nil_both_ways(self) -> None: + self.assertIsNone(node.encode(None)) + self.assertIsNone(node.decode(None)) + + def test_unit_arm_is_written_with_empty_slots_and_reads_back(self) -> None: + self.assertEqual(outcome.encode(UnionArm("Denied")), [2, []]) + self.assertEqual(outcome.decode([2, []]), UnionArm("Denied", None)) + + def test_unit_arm_refuses_a_value_to_carry(self) -> None: + with self.assertRaises(ValueError) as caught: + outcome.encode(UnionArm("Denied", {"message": self.CANARY})) + self.assertIn("Outcome", str(caught.exception)) + self.assertNotIn(self.CANARY, str(caught.exception)) + + def test_arm_keys_are_written_as_declared(self) -> None: + self.assertEqual(node.encode(UnionArm("Image", {"id": "a", "uri": "b", "alt": "c"})), [5, ["a", "b", None, "c"]]) + self.assertEqual(node.encode(UnionArm("Text", {"id": "a", "content": "b"})), [1, ["a", "b"]]) + + def test_encode_refuses_an_arm_the_union_does_not_declare_without_echoing_it(self) -> None: + with self.assertRaises(ValueError) as caught: + crafted.encode(UnionArm("Gem", {"name": self.CANARY})) + self.assertIn("CraftedItem", str(caught.exception)) + self.assertNotIn(self.CANARY, str(caught.exception)) + with self.assertRaises(ValueError): + node.encode({"id": "a"}) + + def test_unknown_key_names_the_union_the_key_and_the_declared_keys_never_the_payload(self) -> None: + with self.assertRaises(ValueError) as caught: + node.decode([4, [self.CANARY]]) + message = str(caught.exception) + self.assertIn("'Node'", message) + self.assertIn("arm key 4 ", message) + self.assertIn("declared: 0, 1, 2, 5", message) + self.assertNotIn(self.CANARY, message) + + def test_a_malformed_union_value_is_refused_without_echoing_it(self) -> None: + for raw in ( + [1], + [1, ["a"], "extra"], + [1, self.CANARY], + self.CANARY, + {"k": self.CANARY}, + [self.CANARY, []], + [True, []], + [1.0, []], + ): + with self.subTest(raw=type(raw).__name__): + with self.assertRaises(ValueError) as caught: + node.decode(raw) + self.assertIn("'Node'", str(caught.exception)) + self.assertNotIn(self.CANARY, str(caught.exception)) + + def test_extra_trailing_arm_slots_are_read(self) -> None: + self.assertEqual( + node.decode([1, ["a", "b", "extra"]]), + UnionArm("Text", {"id": "a", "content": "b"}), + ) + + def test_many_slot_requires_a_list_and_keeps_nil_elements(self) -> None: + slots = [UnionArm("Text", {"id": "a", "content": "b"}), None] + encoded = use.encode({"id": "u", "composition": None, "slots": slots}) + self.assertEqual(encoded, ["u", None, [[1, ["a", "b"]], None]]) + self.assertEqual(use.decode(encoded)["slots"], slots) + with self.assertRaises(ValueError) as caught: + use.decode(["u", None, self.CANARY]) + self.assertNotIn(self.CANARY, str(caught.exception)) + + def test_a_nested_record_with_a_missing_required_slot_is_refused(self) -> None: + with self.assertRaises(ValueError) as caught: + node.decode([1, ["only-id"]]) + self.assertIn("Text", str(caught.exception)) + self.assertIn("slot 1", str(caught.exception)) + + def test_definition_rules(self) -> None: + with self.assertRaises(ValueError): + define_union("Empty", {}) + with self.assertRaises(ValueError): + define_union("Dup", {0: ("A", None), 1: ("A", None)}) + with self.assertRaises(ValueError): + define_union("Neg", {-1: ("A", None)}) + with self.assertRaises(ValueError): + define_union("Flag", {True: ("A", None), "x": ("B", None)}) + with self.assertRaises(ValueError): + database_entry_field("loose", 1, many=True) + + +if __name__ == "__main__": + unittest.main()