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

cultcache-py: slot keys capped at 65535 at definition; a short record reads with nil as the reference does; bool arm key pinned

Browse source 
DatabaseEntryField validates its key in one place (0..MAX_SLOT_KEY, no bool), so
database_entry_field, direct construction and field tuples share the rule. A record
shorter than its declared slots no longer refuses a missing slot: the slot is nil
and its default applies, as the C# reference reads it. define_union refuses True
and False as arm keys, each pinned alone. Every commit builds.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Meta Vens 2026-10-11 10:12:24 +02:00
commit 5d4753e2f0
3 changed files with 37 additions and 22 deletions
Download patch file Download diff file Expand all files Collapse all files

2
packages/cultcache-py/src/cultcache_py/__init__.py
View file

@ -3,6 +3,7 @@ from .backing_store import CultCacheSchemaCatalogEntry, CultCacheSchemaCatalogMe
from .cache import CultCache, CultCacheBuilder
from .documents import (
MAX_PAYLOAD_DEPTH,
MAX_SLOT_KEY,
CultCachePayloadDepthError,
DatabaseEntryField,
DocumentDefinition,
@ -29,6 +30,7 @@ __all__ = [
"DocumentDefinition",
"JsonLinesBackingStore",
"MAX_PAYLOAD_DEPTH",
"MAX_SLOT_KEY",
"SingleFileMessagePackBackingStore",
"SlotCodec",
"UnionArm",

18
packages/cultcache-py/src/cultcache_py/documents.py
View file

@ -18,6 +18,9 @@ MAX_PAYLOAD_DEPTH = 256
"""The payload nesting limit, equal to the C# reference's `MaximumPayloadDepth`: every array and map counts, read
or skipped, the outermost one is depth 1; depth 256 is accepted and 257 refused, on read and on write."""
MAX_SLOT_KEY = 65535
"""The largest slot key a record may define: the largest MessagePack array16 length."""
try:
from msgpack.exceptions import StackError as _MsgpackStackError
@ -111,6 +114,10 @@ class DatabaseEntryField:
codec: SlotCodec | Callable[[], SlotCodec] | None = None
many: bool = False
def __post_init__(self) -> None:
if not _is_key(self.key) or not 0 <= self.key <= MAX_SLOT_KEY:
raise ValueError(f"DatabaseEntry field keys must be integers within 0..{MAX_SLOT_KEY}")
def database_entry_field(
name: str,
@ -122,8 +129,6 @@ def database_entry_field(
) -> 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")
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)
@ -452,14 +457,7 @@ def _record_codec(type: str, normalized: list[DatabaseEntryField], cls: type[Any
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 MISSING:
raise ValueError(
f"DatabaseEntry document type {type!r} is missing required slot {field.key}"
)
else:
value = None
value = decode_slot(field, raw[field.key]) if field.key < len(raw) else None
if value is None and field.default is not MISSING:
value = field.default() if callable(field.default) else field.default
values[field.name] = value

39
packages/cultcache-py/tests/test_union_parity.py
View file

@ -10,6 +10,7 @@ import msgpack
from cultcache_py import (
MAX_PAYLOAD_DEPTH,
MAX_SLOT_KEY,
CultCache,
CultCachePayloadDepthError,
DatabaseEntryField,
@ -278,10 +279,9 @@ class UnionCodecRuleTests(unittest.TestCase):
arm.name = "Pending" # type: ignore[misc]
self.assertEqual(len({UnionArm("Denied"), UnionArm("Denied"), UnionArm("Pending", "wait")}), 2)
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.assertEqual(str(caught.exception), "DatabaseEntry document type 'Text' is missing required slot 1")
def test_a_short_record_is_read_with_nil_for_the_slots_it_lacks_as_the_reference_reads_it(self) -> None:
self.assertEqual(node.decode([1, ["only-id"]]), UnionArm("Text", {"id": "only-id", "content": None}))
self.assertEqual(node.decode([0, ["sapphire"]]), UnionArm("Box", {"id": "sapphire", "children": None}))
def test_definition_rules(self) -> None:
cases = [
@ -296,6 +296,10 @@ class UnionCodecRuleTests(unittest.TestCase):
with self.assertRaises(ValueError) as caught:
define_union(union_name, arms)
self.assertEqual(str(caught.exception), expected)
for union_name, key in (("TrueKey", True), ("FalseKey", False)):
with self.assertRaises(ValueError) as caught:
define_union(union_name, {key: ("A", None)})
self.assertEqual(str(caught.exception), f"Union {union_name!r} arm keys must be integers within Int32")
edge = define_union("Edge", {-(2**31): ("Low", None), -1: ("Negative", None), 2**31 - 1: ("High", None)})
for arm_name, key in (("Low", -(2**31)), ("Negative", -1), ("High", 2**31 - 1)):
self.assertEqual(edge.encode(UnionArm(arm_name)), [key, []])
@ -362,12 +366,10 @@ class RecordCodecTests(unittest.TestCase):
define_record_codec("Nothing", [])
self.assertEqual(str(empty.exception), "DatabaseEntry document types require at least one field")
def test_a_plain_field_helper_makes_a_required_slot_with_no_codec(self) -> None:
def test_a_plain_field_helper_makes_a_slot_with_no_codec(self) -> None:
plain = define_record_codec("Plain", [database_entry_field("id", 0), database_entry_field("label", 1)])
self.assertEqual(plain.decode(plain.encode({"id": "a", "label": "b"})), {"id": "a", "label": "b"})
with self.assertRaises(ValueError) as caught:
plain.decode(["a"])
self.assertEqual(str(caught.exception), "DatabaseEntry document type 'Plain' is missing required slot 1")
self.assertEqual(plain.decode(["a"]), {"id": "a", "label": None})
field = database_entry_field("id", 0)
self.assertIsNone(field.codec)
self.assertFalse(field.many)
@ -385,11 +387,24 @@ class RecordCodecTests(unittest.TestCase):
with self.assertRaises(dataclasses.FrozenInstanceError):
database_entry_field("id", 0).key = 1 # type: ignore[misc]
def test_a_negative_slot_key_is_refused_but_zero_is_not(self) -> None:
def test_slot_keys_are_integers_within_zero_and_the_cap_however_the_field_is_made(self) -> None:
message = "DatabaseEntry field keys must be integers within 0..65535"
self.assertEqual(MAX_SLOT_KEY, 65535)
database_entry_field("zero", 0)
with self.assertRaises(ValueError) as caught:
database_entry_field("negative", -1)
self.assertEqual(str(caught.exception), "DatabaseEntry field keys must be non-negative integers")
top = define_record_codec("Top", [("last", MAX_SLOT_KEY)])
self.assertEqual(len(top.encode({"last": "x"})), MAX_SLOT_KEY + 1)
self.assertEqual(top.decode(top.encode({"last": "x"})), {"last": "x"})
for make in (
lambda key: database_entry_field("bad", key),
lambda key: DatabaseEntryField(name="bad", key=key, default=dataclasses.MISSING),
lambda key: define_record_codec("Bad", [("bad", key)]),
lambda key: define_record_codec("Bad", [("bad", key, "d")]),
):
for key in (-1, MAX_SLOT_KEY + 1, 2**31, True, 1.0, "1"):
with self.subTest(key=key):
with self.assertRaises(ValueError) as caught:
make(key)
self.assertEqual(str(caught.exception), message)
def test_slot_order_follows_keys_not_declaration_order(self) -> None:
record = define_record_codec("Ordered", [("b", 2), ("a", 0)])