Merge hands/union-bounds-py: one 256 nesting bound on read and write in cultcache-py, with the arm-key, slot-key and nil-slot refusals, held to the C# bounds vectors
Soul passes cultlib-gaps:verdict:cut-union-bounds-py.s1 and .s2. Low residuals deferred to follow_ups py-depth-bound-residuals and py-default-payload-json. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Meta Vens
commit
5618d8dde7
3 changed files with 357 additions and 29 deletions
|
|
@ -2,6 +2,9 @@
|
|||
from .backing_store import CultCacheSchemaCatalogEntry, CultCacheSchemaCatalogMember
|
||||
from .cache import CultCache, CultCacheBuilder
|
||||
from .documents import (
|
||||
MAX_PAYLOAD_DEPTH,
|
||||
MAX_SLOT_KEY,
|
||||
CultCachePayloadDepthError,
|
||||
DatabaseEntryField,
|
||||
DocumentDefinition,
|
||||
SlotCodec,
|
||||
|
|
@ -19,12 +22,15 @@ __all__ = [
|
|||
"BackingStore",
|
||||
"CultCache",
|
||||
"CultCacheBuilder",
|
||||
"CultCachePayloadDepthError",
|
||||
"CultCacheEnvelope",
|
||||
"CultCacheSchemaCatalogEntry",
|
||||
"CultCacheSchemaCatalogMember",
|
||||
"DatabaseEntryField",
|
||||
"DocumentDefinition",
|
||||
"JsonLinesBackingStore",
|
||||
"MAX_PAYLOAD_DEPTH",
|
||||
"MAX_SLOT_KEY",
|
||||
"SingleFileMessagePackBackingStore",
|
||||
"SlotCodec",
|
||||
"UnionArm",
|
||||
|
|
|
|||
|
|
@ -2,7 +2,9 @@
|
|||
|
||||
import json
|
||||
from dataclasses import MISSING, dataclass, field
|
||||
from typing import Any, Callable, Generic, Iterable, Mapping, TypeVar
|
||||
from collections.abc import Mapping, Sequence, Set
|
||||
from itertools import chain
|
||||
from typing import Any, Callable, Generic, Iterable, TypeVar
|
||||
|
||||
from .backing_store import CultCacheSchemaCatalogEntry, CultCacheSchemaCatalogMember
|
||||
|
||||
|
|
@ -13,6 +15,51 @@ Decoder = Callable[[Any], T]
|
|||
Validator = Callable[[T], None]
|
||||
Extractor = Callable[[T], str | int | float | bool | None]
|
||||
|
||||
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
|
||||
|
||||
_STACK_FAULTS: tuple[type[BaseException], ...] = (_MsgpackStackError,)
|
||||
except ModuleNotFoundError:
|
||||
_STACK_FAULTS = ()
|
||||
|
||||
|
||||
_CONTAINERS = (Mapping, Sequence, Set)
|
||||
_SCALAR_SEQUENCES = (str, bytes, bytearray, memoryview)
|
||||
|
||||
|
||||
class CultCachePayloadDepthError(ValueError):
|
||||
"""A payload nests deeper than `MAX_PAYLOAD_DEPTH`. Names the type, the depth reached and the limit; never the payload."""
|
||||
|
||||
def __init__(self, type_name: str, direction: str) -> None:
|
||||
self.depth = MAX_PAYLOAD_DEPTH + 1
|
||||
self.limit = MAX_PAYLOAD_DEPTH
|
||||
super().__init__(
|
||||
f"{type_name} payload nests {self.depth} containers deep; the limit is {self.limit} (refused on {direction})."
|
||||
)
|
||||
|
||||
|
||||
def _refuse_past_depth_limit(tree: Any, type_name: str, direction: str) -> None:
|
||||
"""The one nesting count. Walks a payload tree iteratively (every mapping, sequence and set that is not text or
|
||||
bytes is a container), refusing the first container past `MAX_PAYLOAD_DEPTH`."""
|
||||
stack = [(tree, 1)] if _is_container(tree) else []
|
||||
while stack:
|
||||
node, depth = stack.pop()
|
||||
if depth > MAX_PAYLOAD_DEPTH:
|
||||
raise CultCachePayloadDepthError(type_name, direction)
|
||||
children = chain(node.keys(), node.values()) if isinstance(node, Mapping) else node
|
||||
stack.extend((child, depth + 1) for child in children if _is_container(child))
|
||||
|
||||
|
||||
def _is_container(node: Any) -> bool:
|
||||
return isinstance(node, _CONTAINERS) and not isinstance(node, _SCALAR_SEQUENCES)
|
||||
|
||||
|
||||
def _default_encode(value: Any) -> Any:
|
||||
if hasattr(value, "model_dump"):
|
||||
|
|
@ -74,6 +121,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,
|
||||
|
|
@ -85,8 +136,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)
|
||||
|
|
@ -123,10 +172,17 @@ class DocumentDefinition(Generic[T]):
|
|||
def encode_payload(self, value: T) -> bytes:
|
||||
if self.validate:
|
||||
self.validate(value)
|
||||
return self.payload_encoder(self.encode(value))
|
||||
tree = self.encode(value)
|
||||
_refuse_past_depth_limit(tree, self.type, "write")
|
||||
return self.payload_encoder(tree)
|
||||
|
||||
def decode_payload(self, payload: bytes) -> T:
|
||||
value = self.decode(self.payload_decoder(payload))
|
||||
try:
|
||||
tree = self.payload_decoder(payload)
|
||||
except _STACK_FAULTS:
|
||||
raise CultCachePayloadDepthError(self.type, "read") from None
|
||||
_refuse_past_depth_limit(tree, self.type, "read")
|
||||
value = self.decode(tree)
|
||||
if self.validate:
|
||||
self.validate(value)
|
||||
return value
|
||||
|
|
@ -313,14 +369,20 @@ def define_union(
|
|||
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):
|
||||
if not isinstance(raw, list) or len(raw) != 2:
|
||||
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")
|
||||
raise ValueError(f"Union {name!r} value is malformed: the arm key is non-integer")
|
||||
if not _INT32_MIN <= key <= _INT32_MAX:
|
||||
raise ValueError(f"Union {name!r} value is malformed: the arm key is outside Int32")
|
||||
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})")
|
||||
if raw[1] is None:
|
||||
raise ValueError(f"Union {name!r} arm {key} has nil slots")
|
||||
if not isinstance(raw[1], list):
|
||||
raise ValueError(f"Union {name!r} value is malformed: expected [armKey, [armSlots]]")
|
||||
arm_name, codec = by_key[key]
|
||||
return UnionArm(arm_name, None if codec is None else _resolve(codec).decode(raw[1]))
|
||||
|
||||
|
|
@ -396,14 +458,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
|
||||
|
|
|
|||
|
|
@ -1,20 +1,27 @@
|
|||
from __future__ import annotations
|
||||
|
||||
import collections
|
||||
import dataclasses
|
||||
import shutil
|
||||
import tempfile
|
||||
import traceback
|
||||
import types
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
|
||||
import msgpack
|
||||
|
||||
from cultcache_py import (
|
||||
MAX_PAYLOAD_DEPTH,
|
||||
MAX_SLOT_KEY,
|
||||
CultCache,
|
||||
CultCachePayloadDepthError,
|
||||
DatabaseEntryField,
|
||||
SingleFileMessagePackBackingStore,
|
||||
UnionArm,
|
||||
database_entry_field,
|
||||
define_database_entry_type,
|
||||
define_document_type,
|
||||
define_record_codec,
|
||||
define_union,
|
||||
)
|
||||
|
|
@ -35,13 +42,18 @@ use = define_record_codec(
|
|||
],
|
||||
)
|
||||
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)})
|
||||
wrap = define_record_codec("Wrap", [database_entry_field("inner", 0, codec=lambda: node)])
|
||||
node = define_union(
|
||||
"Node", {0: ("Box", box), 1: ("Text", text), 2: ("Use", use), 5: ("Image", image), 6: ("Wrap", wrap)}
|
||||
)
|
||||
outcome = define_union(
|
||||
"Outcome",
|
||||
{
|
||||
0: ("Pending", define_record_codec("Pending", [("message", 0)])),
|
||||
1: ("Completed", define_record_codec("Completed", [("source_version", 0)])),
|
||||
2: ("Denied", None),
|
||||
-2147483648: ("Floor", None),
|
||||
2147483647: ("Ceiling", None),
|
||||
},
|
||||
)
|
||||
tool = define_record_codec("Tool", [("name", 0), ("uses", 1)])
|
||||
|
|
@ -219,7 +231,7 @@ class UnionCodecRuleTests(unittest.TestCase):
|
|||
with self.assertRaises(ValueError) as caught:
|
||||
node.decode([4, [self.CANARY]])
|
||||
message = str(caught.exception)
|
||||
self.assertEqual(message, "Union 'Node' does not declare arm key 4 (declared: 0, 1, 2, 5)")
|
||||
self.assertEqual(message, "Union 'Node' does not declare arm key 4 (declared: 0, 1, 2, 5, 6)")
|
||||
|
||||
def test_a_malformed_union_value_is_refused_without_echoing_it(self) -> None:
|
||||
for raw in (
|
||||
|
|
@ -270,10 +282,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 = [
|
||||
|
|
@ -288,6 +299,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, []])
|
||||
|
|
@ -354,12 +369,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)
|
||||
|
|
@ -377,16 +390,270 @@ 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)])
|
||||
self.assertEqual(record.encode({"a": "x", "b": "y"}), ["x", None, "y"])
|
||||
|
||||
|
||||
CANARY = "CANARY-union-golden-4c2a"
|
||||
|
||||
|
||||
def _wrap_chain(wraps: int, leaf: UnionArm) -> UnionArm:
|
||||
value = leaf
|
||||
for _ in range(wraps):
|
||||
value = UnionArm("Wrap", {"inner": value})
|
||||
return value
|
||||
|
||||
|
||||
class BoundsVectorTests(unittest.TestCase):
|
||||
def test_every_bounds_outcome_is_held(self) -> None:
|
||||
entries = msgpack.unpackb(_vector("bounds-outcomes.msgpack"), raw=False)
|
||||
self.assertEqual(entries[0], ["limit", MAX_PAYLOAD_DEPTH])
|
||||
counts = {"accepted": 0, "refused": 0, "read-only": 0}
|
||||
for file, outcome_kind, union, key, detail in entries[1:]:
|
||||
counts[outcome_kind] += 1
|
||||
payload = _vector(file)
|
||||
if outcome_kind == "accepted":
|
||||
self.assertEqual(
|
||||
surface_document.encode_payload(surface_document.decode_payload(payload)), _vector(detail), file
|
||||
)
|
||||
elif outcome_kind == "read-only":
|
||||
self.assertEqual(surface_document.decode_payload(payload), _surface_value(), file)
|
||||
else:
|
||||
with self.assertRaises(ValueError, msg=file) as caught:
|
||||
surface_document.decode_payload(payload)
|
||||
message = str(caught.exception)
|
||||
self.assertNotIn(CANARY, message, file)
|
||||
if detail == "depth":
|
||||
self.assertIsInstance(caught.exception, CultCachePayloadDepthError, file)
|
||||
self.assertIn(surface_document.type, message, file)
|
||||
self.assertIn("nests 257 containers deep", message, file)
|
||||
self.assertIn("the limit is 256", message, file)
|
||||
self.assertIn("refused on read", message, file)
|
||||
continue
|
||||
self.assertNotIsInstance(caught.exception, CultCachePayloadDepthError, file)
|
||||
self.assertIn(f"Union '{union}' ", message, file)
|
||||
if detail == "key-out-of-range":
|
||||
self.assertIn("outside Int32", message, file)
|
||||
self.assertNotIn("does not declare", message, file)
|
||||
elif detail == "key-not-integer":
|
||||
self.assertIn("non-integer", message, file)
|
||||
else:
|
||||
self.assertEqual(detail, "nil-slots")
|
||||
self.assertIn(f"arm {key} has nil slots", message, file)
|
||||
self.assertEqual(counts, {"accepted": 4, "refused": 7, "read-only": 1})
|
||||
|
||||
def test_the_value_at_the_limit_writes_the_vector_and_the_value_past_it_is_refused_on_write(self) -> None:
|
||||
at_limit = {**_surface_value(), "root": _wrap_chain(126, UnionArm("Box", {"id": "leaf", "children": []}))}
|
||||
self.assertEqual(surface_document.encode_payload(at_limit), _vector("depth-at-limit.msgpack"))
|
||||
past = {**_surface_value(), "root": _wrap_chain(127, UnionArm("Text", {"id": "leaf", "content": CANARY}))}
|
||||
with self.assertRaises(CultCachePayloadDepthError) as caught:
|
||||
surface_document.encode_payload(past)
|
||||
message = str(caught.exception)
|
||||
self.assertEqual((caught.exception.depth, caught.exception.limit), (257, 256))
|
||||
self.assertIn("the limit is 256", message)
|
||||
self.assertIn("refused on write", message)
|
||||
self.assertNotIn(CANARY, message)
|
||||
|
||||
def test_a_cache_put_of_the_value_past_the_limit_refuses_and_stores_nothing(self) -> None:
|
||||
past = {**_surface_value(), "root": _wrap_chain(127, UnionArm("Box", {"id": "leaf", "children": []}))}
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
path = Path(tmp) / "store.cc"
|
||||
cache = (
|
||||
CultCache.builder()
|
||||
.register_document_type(surface_document)
|
||||
.add_generic_store(SingleFileMessagePackBackingStore(path))
|
||||
.build()
|
||||
)
|
||||
with self.assertRaises(CultCachePayloadDepthError):
|
||||
cache.put(surface_document, "surface", past)
|
||||
self.assertIsNone(cache.get(surface_document, "surface"))
|
||||
|
||||
def test_a_payload_beyond_the_c_unpackers_own_stack_is_the_depth_refusal(self) -> None:
|
||||
deep = b"\x91" * 2000 + b"\xc0"
|
||||
with self.assertRaises(CultCachePayloadDepthError) as caught:
|
||||
surface_document.decode_payload(deep)
|
||||
self.assertEqual(str(caught.exception).count("refused on read"), 1)
|
||||
huge = b"\x98" + b"\xc0" * 7 + b"\x91" * 100000 + b"\xc0"
|
||||
with self.assertRaises(CultCachePayloadDepthError):
|
||||
surface_document.decode_payload(huge)
|
||||
|
||||
def test_a_deep_value_is_refused_by_the_depth_error_not_a_recursion_error(self) -> None:
|
||||
tree = define_document_type("tests.deep_tree", encode=lambda value: value)
|
||||
with self.assertRaises(CultCachePayloadDepthError) as caught:
|
||||
tree.encode_payload(_nest(5000))
|
||||
self.assertIn("refused on write", str(caught.exception))
|
||||
document = define_document_type("tests.json_tree")
|
||||
with self.assertRaises(CultCachePayloadDepthError) as read:
|
||||
document.decode_payload(b"[" * 300 + b"]" * 300)
|
||||
self.assertIn("refused on read", str(read.exception))
|
||||
|
||||
def test_a_codec_that_exhausts_the_interpreter_stack_keeps_its_own_recursion_error_on_each_side(self) -> None:
|
||||
def boom(value: object) -> object:
|
||||
return boom(value)
|
||||
|
||||
document = define_document_type("tests.boom", encode=boom, decode=boom)
|
||||
for run in (lambda: document.encode_payload("x"), lambda: document.decode_payload(b'"x"')):
|
||||
try:
|
||||
run()
|
||||
except RecursionError as caught:
|
||||
self.assertNotIsInstance(caught, CultCachePayloadDepthError)
|
||||
self.assertIn("boom", [frame.name for frame in traceback.extract_tb(caught.__traceback__)])
|
||||
else:
|
||||
self.fail("the codec's RecursionError was swallowed")
|
||||
with self.assertRaises(CultCachePayloadDepthError) as stack:
|
||||
surface_document.decode_payload(bytes([0x91]) * 2000 + bytes([0xC0]))
|
||||
self.assertEqual(
|
||||
str(stack.exception),
|
||||
"tests.union_surface payload nests 257 containers deep; the limit is 256 (refused on read).",
|
||||
)
|
||||
|
||||
def test_a_json_payload_past_the_interpreters_own_stack_keeps_the_decoders_recursion_error(self) -> None:
|
||||
document = define_document_type("tests.json_tree")
|
||||
with self.assertRaises(RecursionError) as caught:
|
||||
document.decode_payload(b"[" * 100000)
|
||||
self.assertNotIsInstance(caught.exception, CultCachePayloadDepthError)
|
||||
|
||||
def test_a_refusal_reports_the_depth_where_the_walk_stopped_whatever_the_payloads_real_depth(self) -> None:
|
||||
document = define_document_type("tests.json_tree", encode=lambda value: value)
|
||||
for depth in (257, 300, 900):
|
||||
with self.assertRaises(CultCachePayloadDepthError) as write:
|
||||
document.encode_payload(_nest(depth))
|
||||
self.assertEqual((write.exception.depth, write.exception.limit), (257, 256), depth)
|
||||
with self.assertRaises(CultCachePayloadDepthError) as read:
|
||||
document.decode_payload(b"[" * depth + b"]" * depth)
|
||||
self.assertEqual((read.exception.depth, read.exception.limit), (257, 256), depth)
|
||||
|
||||
def test_the_validator_sees_the_value_on_write_and_the_decoded_value_on_read(self) -> None:
|
||||
seen: list[object] = []
|
||||
document = define_document_type("tests.validated", validate=seen.append)
|
||||
document.decode_payload(document.encode_payload({"a": 1}))
|
||||
self.assertEqual(seen, [{"a": 1}, {"a": 1}])
|
||||
|
||||
def test_a_cycle_is_refused_as_depth(self) -> None:
|
||||
loop: list = []
|
||||
loop.append(loop)
|
||||
document = define_document_type("tests.cycle", encode=lambda value: value)
|
||||
with self.assertRaises(CultCachePayloadDepthError):
|
||||
document.encode_payload(loop)
|
||||
|
||||
|
||||
def _nest(depth: int, mixed: bool = False) -> object:
|
||||
value: object = "leaf"
|
||||
for level in range(depth):
|
||||
value = {"k": value} if mixed and level % 2 == 0 else [value]
|
||||
return value
|
||||
|
||||
|
||||
def _as_tuple(tree: object) -> object:
|
||||
return tuple(_as_tuple(child) for child in tree) if isinstance(tree, list) else tree
|
||||
|
||||
|
||||
class PayloadDepthRuleTests(unittest.TestCase):
|
||||
"""The nesting rule at its seam, on definitions that are not unions: the same routine serves any payload codec."""
|
||||
|
||||
def test_the_limit_is_256_and_depth_256_is_accepted_on_read_and_write_while_257_is_refused(self) -> None:
|
||||
self.assertEqual(MAX_PAYLOAD_DEPTH, 256)
|
||||
document = define_document_type("tests.json_tree", encode=lambda value: value)
|
||||
for mixed in (False, True):
|
||||
at_limit = _nest(256, mixed)
|
||||
self.assertEqual(document.decode_payload(document.encode_payload(at_limit)), at_limit)
|
||||
past = _nest(257, mixed)
|
||||
with self.assertRaises(CultCachePayloadDepthError):
|
||||
document.encode_payload(past)
|
||||
with self.assertRaises(CultCachePayloadDepthError):
|
||||
document.decode_payload(document.payload_encoder(past))
|
||||
|
||||
def test_a_tuple_and_a_container_dict_key_count_as_containers(self) -> None:
|
||||
document = define_document_type("tests.tuple_tree", encode=lambda value: value, payload_encoder=lambda tree: b"ok")
|
||||
self.assertEqual(document.encode_payload(_as_tuple(_nest(256))), b"ok")
|
||||
with self.assertRaises(CultCachePayloadDepthError):
|
||||
document.encode_payload(_as_tuple(_nest(257)))
|
||||
self.assertEqual(document.encode_payload({_as_tuple(_nest(255)): 1}), b"ok")
|
||||
with self.assertRaises(CultCachePayloadDepthError):
|
||||
document.encode_payload({_as_tuple(_nest(256)): 1})
|
||||
|
||||
def test_every_mapping_sequence_and_set_counts_not_only_list_tuple_and_dict(self) -> None:
|
||||
wrappers = {
|
||||
"deque": lambda inner: collections.deque([inner]),
|
||||
"mappingproxy": lambda inner: types.MappingProxyType({"k": inner}),
|
||||
"frozenset": lambda inner: frozenset([inner]),
|
||||
"userlist": lambda inner: collections.UserList([inner]),
|
||||
}
|
||||
for name, wrap in wrappers.items():
|
||||
document = define_document_type(f"tests.{name}", encode=lambda value: value, payload_encoder=lambda tree: b"ok")
|
||||
at_limit: object = "leaf"
|
||||
for _ in range(256):
|
||||
at_limit = wrap(at_limit)
|
||||
self.assertEqual(document.encode_payload(at_limit), b"ok", name)
|
||||
with self.assertRaises(CultCachePayloadDepthError, msg=name):
|
||||
document.encode_payload(wrap(at_limit))
|
||||
decoding = define_document_type(f"tests.{name}.read", payload_decoder=lambda payload, w=wrap: w(at_limit))
|
||||
with self.assertRaises(CultCachePayloadDepthError, msg=name):
|
||||
decoding.decode_payload(b"x")
|
||||
|
||||
def test_text_and_bytes_are_leaves_however_long(self) -> None:
|
||||
document = define_document_type("tests.leaves", encode=lambda value: value, payload_encoder=lambda tree: b"ok")
|
||||
self.assertEqual(document.encode_payload([["x" * 100000, b"y" * 100000, bytearray(5), memoryview(b"z")]]), b"ok")
|
||||
|
||||
def test_a_scalar_payload_has_no_depth(self) -> None:
|
||||
document = define_document_type("tests.scalar", encode=lambda value: value)
|
||||
for scalar in ("x", 5, None):
|
||||
self.assertEqual(document.decode_payload(document.encode_payload(scalar)), scalar)
|
||||
|
||||
def test_the_refusal_names_the_type_the_depth_and_the_limit_and_never_the_payload(self) -> None:
|
||||
document = define_document_type("tests.json_tree", encode=lambda value: value)
|
||||
value: object = CANARY
|
||||
for _ in range(257):
|
||||
value = [value]
|
||||
with self.assertRaises(CultCachePayloadDepthError) as caught:
|
||||
document.encode_payload(value)
|
||||
self.assertEqual(
|
||||
str(caught.exception),
|
||||
"tests.json_tree payload nests 257 containers deep; the limit is 256 (refused on write).",
|
||||
)
|
||||
self.assertIsInstance(caught.exception, ValueError)
|
||||
|
||||
|
||||
class UnionDecodeBoundsTests(unittest.TestCase):
|
||||
def test_the_arm_key_refusals_name_the_union_and_say_what_is_wrong(self) -> None:
|
||||
cases = [
|
||||
([2**31, []], "Union 'Node' value is malformed: the arm key is outside Int32"),
|
||||
([-(2**31) - 1, []], "Union 'Node' value is malformed: the arm key is outside Int32"),
|
||||
([1.0, []], "Union 'Node' value is malformed: the arm key is non-integer"),
|
||||
([True, []], "Union 'Node' value is malformed: the arm key is non-integer"),
|
||||
([1, None], "Union 'Node' arm 1 has nil slots"),
|
||||
([1, CANARY], "Union 'Node' value is malformed: expected [armKey, [armSlots]]"),
|
||||
]
|
||||
for raw, expected in cases:
|
||||
with self.subTest(key=raw[0], slots=type(raw[1]).__name__):
|
||||
with self.assertRaises(ValueError) as caught:
|
||||
node.decode(raw)
|
||||
self.assertEqual(str(caught.exception), expected)
|
||||
|
||||
def test_the_ends_of_the_int32_key_domain_are_unit_arms_that_read_and_write(self) -> None:
|
||||
self.assertEqual(outcome.decode([-(2**31), []]), UnionArm("Floor"))
|
||||
self.assertEqual(outcome.encode(UnionArm("Ceiling")), [2**31 - 1, []])
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
|
|
|
|||
Loading…
Reference in a new issue