Python: a record with unread slots is held read-only (slot-readonly-py) (#126)
This commit is contained in:
parent
5e10909fe9
commit
892765ba97
8 changed files with 705 additions and 31 deletions
|
|
@ -9,6 +9,8 @@
|
|||
|
||||
### Breaking
|
||||
|
||||
- A record whose payload carries a slot past its reader's declared slots, at any depth (the document, a nested record, a list element, a union arm, a unit arm), is read, held read-only, and every write or removal of its key (`put`, `put_global`, `update`, `put_envelope`, `put_envelopes`, `delete`, `delete_global`) is refused with `SchemaConflictError` before any store is touched (`cultlib-gaps:ruling:extra-trailing-slots`). Reading dropped those slots silently and the next write shed them. `SchemaConflictError.unread_slots` is an `UnreadSlots(type_name, slot_count, declared_count, occurrences, depth0_only)` and the message names the type and both counts, never a slot value. A depth-0 mark is consented, and the record writable, only when the record is stored under an id the document lists in `compatible_schema_ids` and that id's catalog entry has slots past the document's, covering the mark's slot count (`cultlib-gaps:ruling:slot-mark-consent`); a nested mark, or a mark under the document's own id, keeps the record read-only whatever is declared. Missing trailing slots keep their declared defaults. An untouched marked record keeps its stored bytes through every other write. Added `CultCache.is_writable(document, key)` (advisory; the write still decides: it also answers False for a foreign-carried key, for a key held as a type the document does not own, and for a global document at a non-global key; `delete` asks the same judges, so removing a key held as another type is refused with `SchemaConflictError` unless the document owns that record's schema id (its own or one it lists compatible), in which case the removal removes that record, its file row and its lookups, as a write of the key would replace it; it returned silently and removed nothing), `DocumentDefinition.decode_payload_marked`, and `UnreadSlots`. A document with a custom `decode` (no slot codec) never marks. A store holding a variant record is still unreadable here. Implies a minor bump while the version is 0.x (0.3.0 to 0.4.0): code that wrote back a record from a newer writer now gets `SchemaConflictError`.
|
||||
|
||||
- A cache holds one record per key whatever its type (`variants:ruling:cache-refuses-retype`). A `put`, `put_envelope`, `put_envelopes` or `put_global` of a second type under a key the cache holds, or under a key a store carries foreign, is refused with `SchemaConflictError` naming the key, even with no store, and changes nothing; retype a key by deleting it, then putting the new type. A write may replace another registered type's record only when the written type owns its schema id (declares it or lists it compatible) and both types share one home store. A load that reads one key from two stores, whatever the types, is refused with `CultCacheError` naming the key and both types, and admits nothing. A store that holds a record no registered type claims now loads and is carried: the cache does not serve it, the store keeps its bytes through every write, `delete` or `put` at its key is refused with `SchemaConflictError`, and registering a type that declares its schema id claims it at the next load. It raised `CultCacheError` "Unknown persisted document type". Two global types share the key `__global__`, so a second global type in one cache is now refused, and a store that already holds two global types at `__global__` (a JSONL store written by caches that never pulled could produce one) no longer loads: the pull raises `CultCacheError` and admits nothing. C# refuses a second persisted global on pull the same way (`cultcache-store-composition.md`, "Globals never invented"); repair such a file by deleting one of the two records with the runtime that wrote it.
|
||||
- Added `foreign_records()` and the frozen `ForeignRecord(key, schema_id, schema_name, stored_at)`, sorted by key in UTF-16 code-unit order. `stores.retyped` and `stores.written_owns` lose their underscore: the cache judge and the single-file store's commit guard ask one predicate.
|
||||
|
||||
|
|
|
|||
|
|
@ -97,6 +97,7 @@ store.
|
|||
- `update_global(...)`
|
||||
- `delete(...)`
|
||||
- `delete_global(...)`
|
||||
- `is_writable(document, key)` (advisory: would a write or removal of the key be refused; see below)
|
||||
- `snapshot()`
|
||||
|
||||
## Indexes And Roles
|
||||
|
|
@ -180,6 +181,16 @@ The payload is a MessagePack array. Field keys are durable slot indexes:
|
|||
- newly added fields should use new keys
|
||||
- fields with defaults tolerate missing or nil slots when older payloads are read
|
||||
|
||||
A payload carrying a slot past the slots a reader declares, at any depth (the document, a nested record, a list element, a
|
||||
union arm, a unit arm), was written by a newer build. It is read, and the record is held read-only: `put`, `update`,
|
||||
`put_envelope(s)`, `delete` and their global forms refuse its key with `SchemaConflictError` before any store is touched, and
|
||||
`SchemaConflictError.unread_slots` is an `UnreadSlots(type_name, slot_count, declared_count, occurrences, depth0_only)` naming the
|
||||
first type read with extra slots (never a slot value). `CultCache.is_writable(document, key)` answers the same question before
|
||||
the write: False for such a record, for a key a store carries foreign, and for a key held as another type's record; the write
|
||||
still decides. `DocumentDefinition.decode_payload_marked(payload)` returns the value and its `UnreadSlots` (or None). Declaring the
|
||||
stored id in `compatible_schema_ids` consents to the loss of depth-0 slots that id's catalog entry reports; a nested mark never is.
|
||||
A store keeps a read-only record's bytes through every other write. A document with a custom `decode` never marks.
|
||||
|
||||
That matches the Rust `#[derive(DatabaseEntry)]` formatter shape and the C#
|
||||
`[Key(n)]` intent: schema evolution comes from stable field slots, not source
|
||||
member order.
|
||||
|
|
|
|||
|
|
@ -9,6 +9,7 @@ from .documents import (
|
|||
DatabaseEntryField,
|
||||
DocumentDefinition,
|
||||
UnionArm,
|
||||
UnreadSlots,
|
||||
database_entry_field,
|
||||
define_database_entry_type,
|
||||
define_document_registry,
|
||||
|
|
@ -38,6 +39,7 @@ __all__ = [
|
|||
"SlotCodec",
|
||||
"StoreUnreadableError",
|
||||
"UnionArm",
|
||||
"UnreadSlots",
|
||||
"leaves",
|
||||
"database_entry_field",
|
||||
"define_database_entry_type",
|
||||
|
|
|
|||
|
|
@ -8,7 +8,7 @@ from dataclasses import dataclass, field, replace
|
|||
from typing import Any, Callable, Generic, Iterable, TypeVar
|
||||
|
||||
from .backing_store import BackingStore, CultCacheEnvelope
|
||||
from .documents import DocumentDefinition, extract_value
|
||||
from .documents import DocumentDefinition, UnreadSlots, extract_value
|
||||
from .stores import SchemaConflictError, retyped, written_owns
|
||||
|
||||
T = TypeVar("T")
|
||||
|
|
@ -123,6 +123,10 @@ class _State:
|
|||
# The records the last pull read that no registered type claims, sorted by key. Derived from the pull and swapped with the
|
||||
# rest of the state, written nowhere else; it decides only refusals at its keys. A store keeps their bytes.
|
||||
foreign: list["ForeignRecord"] = field(default_factory=list)
|
||||
# The records held read-only, by key (a key holds one record) with the type that holds it: the payload carried slots past the
|
||||
# declared ones and nothing consents to the loss. Derived from the load or the write that admitted the record and swapped with
|
||||
# the rest of the state; it decides only refusals at its keys, and any other key is absent.
|
||||
unread: dict[str, tuple[str, UnreadSlots]] = field(default_factory=dict)
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
|
|
@ -337,6 +341,7 @@ class CultCache:
|
|||
routes, generic = self._state.stores_by_type, self._state.generic_store
|
||||
seen_globals: set[str] = set()
|
||||
foreign: list[ForeignRecord] = []
|
||||
loaded_unread: dict[str, tuple[str, UnreadSlots]] = {}
|
||||
type_at_key: dict[str, tuple[str, int]] = {} # one record per key across every store: its type (or schema name) and store
|
||||
stores: list[BackingStore] = []
|
||||
for candidate in [*self._state.stores_by_type.values(), self._state.generic_store]:
|
||||
|
|
@ -376,8 +381,10 @@ class CultCache:
|
|||
loaded_legacy_global_keys[envelope.type] = envelope.key
|
||||
envelope = replace(envelope, key=GLOBAL_KEY)
|
||||
self._claim_key(type_at_key, envelope.key, envelope.type, position)
|
||||
envelope = self._stamped(document, envelope)
|
||||
value = document.decode_payload(envelope.payload)
|
||||
arrived, envelope = envelope, self._stamped(document, envelope)
|
||||
value, unread = self._decode_marked(document, arrived)
|
||||
if unread is not None:
|
||||
loaded_unread[envelope.key] = (envelope.type, unread)
|
||||
loaded_values.setdefault(envelope.type, {})[envelope.key] = value
|
||||
read_values[(envelope.type, envelope.key)] = value
|
||||
loaded_envelopes.setdefault(envelope.type, {})[envelope.key] = envelope
|
||||
|
|
@ -393,6 +400,7 @@ class CultCache:
|
|||
self._state.lookup_keys,
|
||||
self._state.legacy_global_keys,
|
||||
self._state.foreign,
|
||||
self._state.unread,
|
||||
) = (
|
||||
loaded_values,
|
||||
loaded_envelopes,
|
||||
|
|
@ -401,6 +409,7 @@ class CultCache:
|
|||
loaded_lookup_keys,
|
||||
loaded_legacy_global_keys,
|
||||
sorted(foreign, key=lambda record: _utf16_order(record.key)),
|
||||
loaded_unread,
|
||||
)
|
||||
|
||||
@staticmethod
|
||||
|
|
@ -491,9 +500,9 @@ class CultCache:
|
|||
def put_envelope(self, document: DocumentDefinition[T], envelope: CultCacheEnvelope) -> T:
|
||||
self._assert_registered(document)
|
||||
self._check_envelope(document, envelope)
|
||||
envelope = self._stamped(document, envelope)
|
||||
value = document.decode_payload(envelope.payload)
|
||||
self._write(document, [(envelope, value)], batch=False)
|
||||
arrived, envelope = envelope, self._stamped(document, envelope)
|
||||
value, unread = self._decode_marked(document, envelope)
|
||||
self._write(document, [(envelope, value)], batch=False, unread={envelope.key: unread})
|
||||
return value
|
||||
|
||||
@_mutating
|
||||
|
|
@ -501,13 +510,33 @@ class CultCache:
|
|||
self._assert_registered(document)
|
||||
values: list[T] = []
|
||||
stamped: list[CultCacheEnvelope] = []
|
||||
unread: dict[str, UnreadSlots | None] = {}
|
||||
for envelope in envelopes:
|
||||
self._check_envelope(document, envelope)
|
||||
stamped.append(self._stamped(document, envelope))
|
||||
values.append(document.decode_payload(envelope.payload))
|
||||
self._write(document, list(zip(stamped, values)), batch=True)
|
||||
value, unread[envelope.key] = self._decode_marked(document, stamped[-1])
|
||||
values.append(value)
|
||||
self._write(document, list(zip(stamped, values)), batch=True, unread=unread)
|
||||
return values
|
||||
|
||||
@staticmethod
|
||||
def _decode_marked(document: DocumentDefinition[Any], held_as: CultCacheEnvelope) -> tuple[Any, UnreadSlots | None]:
|
||||
"""The payload decoded under the document, with the mark of any slot it carried past the declared ones, or no mark when the
|
||||
loss is consented. `held_as` is the envelope under the id the record is stored under: a load's arrival, a write's stamped
|
||||
envelope, so a key is writable in memory exactly when it is after a reload. A depth-0 mark is consented only when that id is
|
||||
one the document lists compatible and its catalog entry has slots past the document's that cover the mark's slot count (the
|
||||
loss a compatible-id declaration consents to, cultlib-gaps:ruling:slot-mark-consent). Any other mark keeps the record read-only."""
|
||||
value, slots = document.decode_payload_marked(held_as.payload)
|
||||
if slots is None:
|
||||
return value, None
|
||||
entry, stored = document.catalog_entry(), held_as.catalog_entry
|
||||
if slots.depth0_only and stored is not None and held_as.schema_id in entry.compatible_schema_ids and held_as.schema_id != entry.schema_id:
|
||||
declared = {member.slot for member in entry.members}
|
||||
reported = [member.slot for member in stored.members if member.slot not in declared]
|
||||
if reported and slots.slot_count <= max(reported) + 1:
|
||||
return value, None
|
||||
return value, slots
|
||||
|
||||
@staticmethod
|
||||
def _stamped(document: DocumentDefinition[Any], envelope: CultCacheEnvelope) -> CultCacheEnvelope:
|
||||
"""The record under its registered schema's id, whatever id the envelope arrived under, so the catalog entry a store
|
||||
|
|
@ -527,7 +556,12 @@ class CultCache:
|
|||
raise CultCacheError(f"Global document {document.type} must use key {GLOBAL_KEY}")
|
||||
|
||||
def _write(
|
||||
self, document: DocumentDefinition[Any], records: list[tuple[CultCacheEnvelope, Any]], *, batch: bool
|
||||
self,
|
||||
document: DocumentDefinition[Any],
|
||||
records: list[tuple[CultCacheEnvelope, Any]],
|
||||
*,
|
||||
batch: bool,
|
||||
unread: dict[str, UnreadSlots | None] | None = None,
|
||||
) -> None:
|
||||
"""Validates every record (home and name and index extractors included) before the store is
|
||||
touched; once the store accepts, applying to the cache has no fallible step."""
|
||||
|
|
@ -535,6 +569,7 @@ class CultCache:
|
|||
# A batch naming a key twice lands the last record for it, as C# CultCacheBatch stages one operation per key. The
|
||||
# judge, the store and the apply all read this one list.
|
||||
records = list({envelope.key: (envelope, value) for envelope, value in records}.values())
|
||||
self._refuse_unread([envelope.key for envelope, _ in records])
|
||||
name_extractor = self._state.name_extractors.get(type)
|
||||
index_extractors = self._state.index_extractors.get(type, {})
|
||||
admitted = [
|
||||
|
|
@ -566,6 +601,9 @@ class CultCache:
|
|||
self._remove_lookups(type, envelope.key)
|
||||
values[envelope.key] = value
|
||||
envelopes[envelope.key] = envelope
|
||||
# A held mark refused this write above, so a key reaching here carries none to clear.
|
||||
if (mark := (unread or {}).get(envelope.key)) is not None:
|
||||
self._state.unread[envelope.key] = (type, mark)
|
||||
lookup_keys[envelope.key] = keys
|
||||
if keys.name is not None:
|
||||
self._state.names[(type, keys.name)] = envelope.key
|
||||
|
|
@ -592,6 +630,51 @@ class CultCache:
|
|||
replaced.append((other, envelope.key))
|
||||
return replaced
|
||||
|
||||
@_locked
|
||||
def is_writable(self, document: DocumentDefinition[Any], key: str) -> bool:
|
||||
"""Whether a write or removal of the key by this document would pass the judges that run before any store is touched: False for
|
||||
a record held read-only (its payload carried slots past the declared ones, at any depth, and nothing consents to the loss, so
|
||||
writing it would drop data a newer writer stored), for a key a store carries foreign, and for a key held as another type's
|
||||
record this document does not own. Advisory (the write still decides): it answers at call time, and a key nothing holds is
|
||||
writable."""
|
||||
self._assert_registered(document)
|
||||
try:
|
||||
self._refuse_at(document, key)
|
||||
except (SchemaConflictError, CultCacheError):
|
||||
return False
|
||||
return True
|
||||
|
||||
def _refuse_at(self, document: DocumentDefinition[Any], key: str) -> list[tuple[str, str]]:
|
||||
"""The judges a write or removal of the key by this document passes before any store is touched, asked by `is_writable` and
|
||||
by `delete`: a global document writes only its global key, a record held read-only is never replaced, and a key held as
|
||||
another type's record or carried foreign is refused (`_refuse_retype`)."""
|
||||
if document.global_document and key != GLOBAL_KEY:
|
||||
raise CultCacheError(f"Global document {document.type} must use key {GLOBAL_KEY}")
|
||||
self._refuse_unread([key])
|
||||
entry = document.catalog_entry()
|
||||
probe = CultCacheEnvelope.create(key=key, type=document.type, payload=b"", schema_id=entry.schema_id, catalog_entry=entry)
|
||||
return self._refuse_retype(document, [probe])
|
||||
|
||||
def _refuse_unread(self, keys: list[str]) -> None:
|
||||
"""The one check every write and removal asks before any store is touched."""
|
||||
for key in keys:
|
||||
held = self._state.unread.get(key)
|
||||
if held is None:
|
||||
continue
|
||||
type, slots = held
|
||||
raise SchemaConflictError(
|
||||
self._state.envelopes[type][key].schema_id or type,
|
||||
[self._state.documents[type].catalog_entry().schema_name],
|
||||
key,
|
||||
message=(
|
||||
f"{self._state.documents[type].catalog_entry().schema_name} record {key} is read-only: {slots.type_name} was read with "
|
||||
f"{slots.slot_count} slots where this build declares {slots.declared_count} ({slots.occurrences} object(s) in the "
|
||||
"record). Writing it would drop data a newer writer stored; read it with types that declare those slots, or "
|
||||
"write it from that build."
|
||||
),
|
||||
unread_slots=slots,
|
||||
)
|
||||
|
||||
@staticmethod
|
||||
def _foreign_refusal(record: ForeignRecord) -> SchemaConflictError:
|
||||
return SchemaConflictError(
|
||||
|
|
@ -644,16 +727,18 @@ class CultCache:
|
|||
|
||||
def _delete(self, document: DocumentDefinition[Any], key: str) -> None:
|
||||
self._assert_registered(document)
|
||||
if key not in self._state.envelopes.get(document.type, {}):
|
||||
carried = next((record for record in self._state.foreign if record.key == key), None)
|
||||
if carried is not None:
|
||||
raise self._foreign_refusal(carried)
|
||||
# A key held as another type's record this document owns is removed, as a write of the key would replace it (C# Remove is by key).
|
||||
owned = self._refuse_at(document, key)
|
||||
store = self._store_for_type(document.type)
|
||||
legacy_key = self._state.legacy_global_keys.get(document.type) if key == GLOBAL_KEY else None
|
||||
if store is not None:
|
||||
store.delete(document.type, key if legacy_key is None else legacy_key)
|
||||
# One record per key: the owned record when this document does not hold the key, else its own, in one store call.
|
||||
holder = owned[0][0] if owned else document.type
|
||||
store.delete(holder, key if legacy_key is None else legacy_key)
|
||||
if legacy_key is not None:
|
||||
del self._state.legacy_global_keys[document.type]
|
||||
for other, _ in owned:
|
||||
self._drop(other, key)
|
||||
self._drop(document.type, key)
|
||||
|
||||
def _drop(self, type: str, key: str) -> None:
|
||||
|
|
|
|||
|
|
@ -1,6 +1,7 @@
|
|||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import threading
|
||||
from dataclasses import MISSING, dataclass, field
|
||||
from collections.abc import Mapping, Sequence, Set
|
||||
from itertools import chain
|
||||
|
|
@ -30,6 +31,49 @@ MAX_SLOT_KEY = 65535
|
|||
_STACK_FAULTS: tuple[type[BaseException], ...] = (_MsgpackStackError,)
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class UnreadSlots:
|
||||
"""What a payload decode found past the slots this runtime declares, at any depth: the first type read with extra slots, how
|
||||
many slots it carried against how many it declares, and how many objects in the payload did. It holds no slot value, so a
|
||||
refusal can name it. `depth0_only` is True when every mark sat in the document's own record (the half of consent the payload
|
||||
decides). A record carrying it is read-only unless the schema comparison explains every mark (see `CultCache.is_writable`)."""
|
||||
|
||||
type_name: str
|
||||
slot_count: int
|
||||
declared_count: int
|
||||
occurrences: int
|
||||
depth0_only: bool
|
||||
|
||||
|
||||
class _SlotScope:
|
||||
"""Opened by `DocumentDefinition.decode_payload_marked`; the record codec and the union codec note into it. It is thread-local,
|
||||
so no codec signature carries it, and a codec decoded outside a payload decode notes nothing."""
|
||||
|
||||
def __init__(self) -> None:
|
||||
self.depth = 0 # records entered: 0 while the document's own record is read
|
||||
self.first: tuple[str, int, int] | None = None
|
||||
self.occurrences = 0
|
||||
self.depth0_only = True
|
||||
|
||||
def note(self, type_name: str, slot_count: int, declared_count: int, depth0: bool) -> None:
|
||||
if self.first is None:
|
||||
self.first = (type_name, slot_count, declared_count)
|
||||
self.occurrences += 1
|
||||
self.depth0_only = self.depth0_only and depth0
|
||||
|
||||
def result(self) -> UnreadSlots | None:
|
||||
if self.first is None:
|
||||
return None
|
||||
return UnreadSlots(*self.first, self.occurrences, self.depth0_only)
|
||||
|
||||
|
||||
_scope = threading.local()
|
||||
|
||||
|
||||
def _current_scope() -> _SlotScope | None:
|
||||
return getattr(_scope, "open", None)
|
||||
|
||||
|
||||
_CONTAINERS = (Mapping, Sequence, Set)
|
||||
_SCALAR_SEQUENCES = (str, bytes, bytearray, memoryview)
|
||||
|
||||
|
|
@ -185,15 +229,25 @@ class DocumentDefinition(Generic[T]):
|
|||
return self.payload_encoder(tree)
|
||||
|
||||
def decode_payload(self, payload: bytes) -> T:
|
||||
return self.decode_payload_marked(payload)[0]
|
||||
|
||||
def decode_payload_marked(self, payload: bytes) -> tuple[T, UnreadSlots | None]:
|
||||
"""The value and what the decode found past the declared slots, at any depth. The payload depth check runs first: a
|
||||
payload past the limit is refused, never marked."""
|
||||
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)
|
||||
outer, opened = _current_scope(), _SlotScope()
|
||||
_scope.open = opened
|
||||
try:
|
||||
value = self.decode(tree)
|
||||
finally:
|
||||
_scope.open = outer
|
||||
if self.validate:
|
||||
self.validate(value)
|
||||
return value
|
||||
return value, opened.result()
|
||||
|
||||
def __post_init__(self) -> None:
|
||||
# Judged here so every constructor passes it: the factories, the dataclass constructor and dataclasses.replace.
|
||||
|
|
@ -418,6 +472,9 @@ def define_union(
|
|||
raise ValueError(f"Union {name!r} value is malformed: expected [armKey, [armSlots]]")
|
||||
arm_name, codec = by_key[key]
|
||||
if codec is None:
|
||||
scope = _current_scope()
|
||||
if scope is not None and raw[1]:
|
||||
scope.note(arm_name, len(raw[1]), 0, False)
|
||||
return UnionArm(arm_name, None)
|
||||
try:
|
||||
return UnionArm(arm_name, codec.decode(raw[1]))
|
||||
|
|
@ -479,17 +536,26 @@ def _record_codec(
|
|||
def decode(raw: Any) -> Any:
|
||||
if not isinstance(raw, list):
|
||||
raise ValueError(f"DatabaseEntry document type {type!r} expected a MessagePack array")
|
||||
scope = _current_scope()
|
||||
if scope is not None and len(raw) > max_key + 1:
|
||||
scope.note(type, len(raw), max_key + 1, scope.depth == 0)
|
||||
values: dict[str, Any] = {}
|
||||
for field in normalized:
|
||||
if field.key < len(raw):
|
||||
try:
|
||||
values[field.name] = field.codec.decode(raw[field.key])
|
||||
except ValueError as error:
|
||||
raise ValueError(f"{type} slot {field.key} ({field.name}): {error}") from None
|
||||
elif field.default is MISSING:
|
||||
values[field.name] = None
|
||||
else:
|
||||
values[field.name] = field.default() if callable(field.default) else field.default
|
||||
if scope is not None:
|
||||
scope.depth += 1
|
||||
try:
|
||||
for field in normalized:
|
||||
if field.key < len(raw):
|
||||
try:
|
||||
values[field.name] = field.codec.decode(raw[field.key])
|
||||
except ValueError as error:
|
||||
raise ValueError(f"{type} slot {field.key} ({field.name}): {error}") from None
|
||||
elif field.default is MISSING:
|
||||
values[field.name] = None
|
||||
else:
|
||||
values[field.name] = field.default() if callable(field.default) else field.default
|
||||
finally:
|
||||
if scope is not None:
|
||||
scope.depth -= 1
|
||||
if cls is None:
|
||||
return values
|
||||
return cls(**values)
|
||||
|
|
|
|||
|
|
@ -13,6 +13,7 @@ from typing import Any, Callable, Iterator, TypeVar
|
|||
|
||||
import msgpack
|
||||
|
||||
from .documents import UnreadSlots
|
||||
from .backing_store import (
|
||||
CultCacheEnvelope,
|
||||
CultCacheSchemaCatalogEntry,
|
||||
|
|
@ -46,9 +47,19 @@ class SchemaConflictError(ValueError):
|
|||
"""A write the catalog cannot describe: records of different types under one schema id, two arrived schemas that share an id
|
||||
and disagree on the schema name, or a record no chosen entry publishes. Nothing is written. It names the id, the names
|
||||
involved (`schema_names`) and a record key (`record_key`). A cache raises it too, with an empty record key, when a
|
||||
document registers a schema id another registered document already carries."""
|
||||
document registers a schema id another registered document already carries. `unread_slots` is None except for the refusal to
|
||||
write or remove a record held read-only because its payload carries slots past the declared ones; its message names the type
|
||||
and both slot counts and never holds a slot value."""
|
||||
|
||||
def __init__(self, schema_id: str, schema_names: list[str], record_key: str, *, message: str | None = None) -> None:
|
||||
def __init__(
|
||||
self,
|
||||
schema_id: str,
|
||||
schema_names: list[str],
|
||||
record_key: str,
|
||||
*,
|
||||
message: str | None = None,
|
||||
unread_slots: UnreadSlots | None = None,
|
||||
) -> None:
|
||||
super().__init__(
|
||||
message
|
||||
or f"Schema id {schema_id!r} cannot describe record {record_key!r}: {', '.join(repr(name) for name in schema_names)}"
|
||||
|
|
@ -56,6 +67,7 @@ class SchemaConflictError(ValueError):
|
|||
self.schema_id = schema_id
|
||||
self.schema_names = schema_names
|
||||
self.record_key = record_key
|
||||
self.unread_slots = unread_slots
|
||||
|
||||
|
||||
class JsonLinesBackingStore:
|
||||
|
|
|
|||
492
packages/cultcache-py/tests/test_slot_readonly.py
Normal file
492
packages/cultcache-py/tests/test_slot_readonly.py
Normal file
|
|
@ -0,0 +1,492 @@
|
|||
from __future__ import annotations
|
||||
|
||||
import dataclasses
|
||||
import shutil
|
||||
import threading
|
||||
import tempfile
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
|
||||
import msgpack
|
||||
from cultcache_py.cache import CultCacheError
|
||||
|
||||
from cultcache_py import (
|
||||
CultCache,
|
||||
CultCachePayloadDepthError,
|
||||
SlotCodec,
|
||||
SchemaConflictError,
|
||||
SingleFileMessagePackBackingStore,
|
||||
UnionArm,
|
||||
UnreadSlots,
|
||||
database_entry_field,
|
||||
define_database_entry_type,
|
||||
define_record_codec,
|
||||
leaves as t,
|
||||
)
|
||||
import test_union_parity as _parity
|
||||
from test_union_parity import (
|
||||
VECTORS,
|
||||
_surface_value,
|
||||
_vector,
|
||||
surface_document,
|
||||
unit_outcome,
|
||||
)
|
||||
|
||||
SLOT = "extra-slot" # the canary: the value of every extra slot in the vectors; no refusal may carry it
|
||||
|
||||
# The vectors name the C# types; this runtime names a type by its codec's name, a document by its document type.
|
||||
_NAMES = {"UnionSurface": surface_document.type, "SurfaceDeclaring": surface_document.type, "Denied2": "Denied"}
|
||||
|
||||
slot_unit_document = define_database_entry_type(
|
||||
"tests.slot_unit",
|
||||
[database_entry_field("name", 0, codec=t.string), database_entry_field("outcome", 1, codec=unit_outcome)],
|
||||
name="name",
|
||||
)
|
||||
# The reader of extra-slot-consent.cc: the surface document's slots under a new schema id that lists the old one compatible.
|
||||
declaring_document = define_database_entry_type(
|
||||
"tests.union_surface",
|
||||
[
|
||||
database_entry_field("id", 0, codec=t.string),
|
||||
database_entry_field("version", 1, codec=t.int32),
|
||||
database_entry_field("root", 2, codec=_parity._node),
|
||||
database_entry_field("outcomes", 3, codec=t.list(_parity.outcome)),
|
||||
database_entry_field("part", 4, codec=t.nullable(_parity.item)),
|
||||
database_entry_field("crafted", 5, codec=t.nullable(_parity.crafted)),
|
||||
database_entry_field("missing", 6, codec=_parity._node),
|
||||
],
|
||||
name="id",
|
||||
schema_id="tests.union_surface.v2",
|
||||
schema_name="tests.union_surface",
|
||||
compatible_schema_ids=("tests.union_surface.wide",),
|
||||
)
|
||||
|
||||
# file -> (type, slots carried, slots declared, every mark at depth 0): the README table.
|
||||
_PAYLOADS = {
|
||||
"extra-slot-record.msgpack": ("UnionSurface", 8, 7, True),
|
||||
"extra-slot-nested-record.msgpack": ("Composition", 3, 2, False),
|
||||
"extra-slot-arm.msgpack": ("Box", 3, 2, False),
|
||||
"extra-slot-list-element.msgpack": ("Text", 3, 2, False),
|
||||
"depth-skipped-at-limit.msgpack": ("UnionSurface", 8, 7, True),
|
||||
}
|
||||
|
||||
|
||||
def _copy_store_without_variants(source: Path, target: Path) -> None:
|
||||
"""This runtime holds no variant records (a five-slot record makes a store unreadable here), so the store vector is
|
||||
walked with its variant row removed; every other byte is the C# writer's."""
|
||||
[snapshot] = list(msgpack.Unpacker(open(source, "rb"), raw=False))
|
||||
snapshot[2] = [record for record in snapshot[2] if len(record) <= 4]
|
||||
target.write_bytes(msgpack.packb(snapshot, use_bin_type=True))
|
||||
|
||||
|
||||
def _open(path: Path, *documents) -> CultCache:
|
||||
builder = CultCache.builder()
|
||||
for document in documents:
|
||||
builder.register_document_type(document)
|
||||
cache = builder.add_generic_store(SingleFileMessagePackBackingStore(path)).build()
|
||||
cache.pull_all_backing_stores()
|
||||
return cache
|
||||
|
||||
|
||||
def _rows():
|
||||
return msgpack.unpackb(_vector("slot-outcomes.msgpack"), raw=False)
|
||||
|
||||
|
||||
class PayloadMarkTests(unittest.TestCase):
|
||||
def test_every_read_only_payload_decodes_and_names_its_mark(self) -> None:
|
||||
for file, (type_name, carried, declared, depth0) in _PAYLOADS.items():
|
||||
with self.subTest(file=file):
|
||||
value, mark = surface_document.decode_payload_marked(_vector(file))
|
||||
self.assertEqual(value, _surface_value())
|
||||
self.assertEqual(mark, UnreadSlots(_NAMES.get(type_name, type_name), carried, declared, 1, depth0))
|
||||
self.assertEqual(surface_document.decode_payload(_vector(file)), _surface_value())
|
||||
|
||||
def test_the_outcomes_manifest_still_counts_nine_rows_and_four_read_only(self) -> None:
|
||||
rows = msgpack.unpackb(_vector("outcomes.msgpack"), raw=False)
|
||||
self.assertEqual(len(rows), 9)
|
||||
self.assertEqual([row[0] for row in rows if row[1] == "read-only"], [f for f in _PAYLOADS if f != "depth-skipped-at-limit.msgpack"])
|
||||
|
||||
def test_a_clean_payload_and_a_short_one_carry_no_mark(self) -> None:
|
||||
self.assertEqual(surface_document.decode_payload_marked(_vector("union-surface.msgpack")), (_surface_value(), None))
|
||||
slots = msgpack.unpackb(_vector("union-surface.msgpack"), raw=False)
|
||||
slots[2] = [1, ["only-id"]] # a Text arm with a missing trailing slot: defaults, no mark
|
||||
_, mark = surface_document.decode_payload_marked(msgpack.packb(slots, use_bin_type=True))
|
||||
self.assertIsNone(mark)
|
||||
|
||||
def test_a_retired_slot_below_the_highest_key_does_not_mark(self) -> None:
|
||||
slots = msgpack.unpackb(_vector("union-surface.msgpack"), raw=False)
|
||||
# Image's declared slots are 0, 1, 3 (2 retired): four carried, four declared.
|
||||
image = slots[2][1][1][2]
|
||||
self.assertEqual(image[0], 5)
|
||||
image[1][2] = "retired-slot-value"
|
||||
_, mark = surface_document.decode_payload_marked(msgpack.packb(slots, use_bin_type=True))
|
||||
self.assertIsNone(mark)
|
||||
|
||||
def test_the_depth_check_runs_before_the_mark(self) -> None:
|
||||
with self.assertRaises(CultCachePayloadDepthError):
|
||||
surface_document.decode_payload_marked(_vector("depth-skipped-past-limit.msgpack"))
|
||||
|
||||
def test_the_unit_arm_with_extra_slots_is_read_past_and_marked(self) -> None:
|
||||
payload = msgpack.packb(["unit-arm", [2, [SLOT]]], use_bin_type=True)
|
||||
value, mark = slot_unit_document.decode_payload_marked(payload)
|
||||
self.assertEqual(value, {"name": "unit-arm", "outcome": UnionArm("Denied")})
|
||||
self.assertEqual(mark, UnreadSlots("Denied", 1, 0, 1, False))
|
||||
self.assertEqual(slot_unit_document.decode_payload_marked(msgpack.packb(["ok", [2, []]]))[1], None)
|
||||
|
||||
def test_a_codec_decoded_outside_a_payload_decode_notes_nothing(self) -> None:
|
||||
self.assertEqual(unit_outcome.decode([2, [SLOT]]), UnionArm("Denied"))
|
||||
|
||||
def test_the_scope_belongs_to_its_thread(self) -> None:
|
||||
def decode_elsewhere(raw):
|
||||
thread = threading.Thread(target=lambda: unit_outcome.decode([2, [SLOT]]))
|
||||
thread.start()
|
||||
thread.join()
|
||||
return raw
|
||||
|
||||
outer = define_database_entry_type("tests.threaded", [database_entry_field("a", 0, codec=SlotCodec(lambda v: v, decode_elsewhere))])
|
||||
self.assertIsNone(outer.decode_payload_marked(msgpack.packb(["x"]))[1])
|
||||
|
||||
def test_a_nested_payload_decode_restores_the_outer_scope(self) -> None:
|
||||
def decode_clean_inside(raw):
|
||||
surface_document.decode_payload_marked(_vector("union-surface.msgpack"))
|
||||
return raw
|
||||
|
||||
inner = define_record_codec("Inner", [database_entry_field("x", 0, codec=t.string)])
|
||||
outer = define_database_entry_type(
|
||||
"tests.nested",
|
||||
[
|
||||
database_entry_field("a", 0, codec=SlotCodec(lambda v: v, decode_clean_inside)),
|
||||
database_entry_field("b", 1, codec=inner),
|
||||
],
|
||||
)
|
||||
_, mark = outer.decode_payload_marked(msgpack.packb(["x", ["v", SLOT]]))
|
||||
self.assertEqual(mark, UnreadSlots("Inner", 2, 1, 1, False))
|
||||
|
||||
|
||||
class StoreRowTests(unittest.TestCase):
|
||||
def setUp(self) -> None:
|
||||
self.tmp = Path(tempfile.mkdtemp())
|
||||
self.addCleanup(shutil.rmtree, self.tmp, True)
|
||||
|
||||
def _store(self, name: str) -> tuple[Path, CultCache]:
|
||||
path = self.tmp / name
|
||||
_copy_store_without_variants(VECTORS / name, path)
|
||||
documents = (declaring_document,) if name == "extra-slot-consent.cc" else (surface_document, slot_unit_document)
|
||||
return path, _open(path, *documents)
|
||||
|
||||
def _document(self, store: str, key: str):
|
||||
if key == "unit-arm":
|
||||
return slot_unit_document
|
||||
return declaring_document if store == "extra-slot-consent.cc" else surface_document
|
||||
|
||||
def test_every_row_is_writable_exactly_where_the_vector_says_and_refuses_with_its_mark(self) -> None:
|
||||
walked = 0
|
||||
for store, key, writable, explained, type_name, carried, declared, occurrences, flatten in _rows():
|
||||
if flatten is not None:
|
||||
continue # a variant: this runtime holds none (deviation)
|
||||
walked += 1
|
||||
with self.subTest(store=store, key=key):
|
||||
path, cache = self._store(store)
|
||||
document = self._document(store, key)
|
||||
before = path.read_bytes()
|
||||
self.assertEqual(cache.is_writable(document, key), writable)
|
||||
held = cache.get(document, key)
|
||||
self.assertIsNotNone(held)
|
||||
if writable:
|
||||
continue
|
||||
self.assertEqual(explained, False)
|
||||
for attempt in (
|
||||
lambda: cache.put(document, key, held),
|
||||
lambda: cache.delete(document, key),
|
||||
lambda: cache.put_envelope(document, cache.get_required_envelope(document, key)),
|
||||
lambda: cache.put_envelopes(document, [cache.get_required_envelope(document, key)]),
|
||||
lambda: cache.update(document, key, lambda value: value),
|
||||
):
|
||||
with self.assertRaises(SchemaConflictError) as caught:
|
||||
attempt()
|
||||
refusal = caught.exception
|
||||
self.assertEqual(
|
||||
refusal.unread_slots,
|
||||
UnreadSlots(_NAMES.get(type_name, type_name), carried, declared, occurrences, refusal.unread_slots.depth0_only),
|
||||
)
|
||||
self.assertEqual(refusal.record_key, key)
|
||||
message = str(refusal)
|
||||
self.assertNotIn(SLOT, message)
|
||||
self.assertIn(refusal.unread_slots.type_name, message)
|
||||
self.assertIn(f"{carried} slots", message)
|
||||
self.assertIn(f"declares {declared}", message)
|
||||
self.assertEqual(path.read_bytes(), before)
|
||||
self.assertTrue(cache.get(document, key) is held)
|
||||
self.assertEqual(walked, 7) # eight rows, the variant not walked
|
||||
|
||||
def test_a_key_no_store_holds_and_another_key_are_writable_while_a_marked_one_is_held(self) -> None:
|
||||
path, cache = self._store("extra-slot-store.cc")
|
||||
self.assertFalse(cache.is_writable(surface_document, "marked"))
|
||||
self.assertTrue(cache.is_writable(surface_document, "nothing-here"))
|
||||
cache.put(surface_document, "other", cache.get_required(surface_document, "clean"))
|
||||
self.assertEqual(cache.get(surface_document, "other"), _surface_value())
|
||||
|
||||
def test_consent_lands_an_explained_mark_and_refuses_the_others(self) -> None:
|
||||
path, cache = self._store("extra-slot-consent.cc")
|
||||
self.assertTrue(cache.is_writable(declaring_document, "consented"))
|
||||
cache.put(declaring_document, "consented", cache.get_required(declaring_document, "consented"))
|
||||
[stored] = [e for e in SingleFileMessagePackBackingStore(path).pull_all() if e.key == "consented"]
|
||||
self.assertEqual(stored.schema_id, "tests.union_surface.v2")
|
||||
self.assertEqual(len(msgpack.unpackb(stored.payload, raw=False)), 7) # the consented slot is gone
|
||||
self.assertTrue(cache.is_writable(declaring_document, "consented"))
|
||||
for key in ("over-consent", "nested"):
|
||||
self.assertFalse(cache.is_writable(declaring_document, key), key)
|
||||
with self.assertRaises(SchemaConflictError):
|
||||
cache.put(declaring_document, key, cache.get_required(declaring_document, key))
|
||||
|
||||
def test_a_reload_marks_again_and_a_write_of_a_clean_envelope_over_a_marked_key_is_refused(self) -> None:
|
||||
path, cache = self._store("extra-slot-store.cc")
|
||||
envelope = cache.get_required_envelope(surface_document, "clean")
|
||||
with self.assertRaises(SchemaConflictError):
|
||||
cache.put_envelope(surface_document, type(envelope)(**{**envelope.__dict__, "key": "marked"}))
|
||||
cache.pull_all_backing_stores()
|
||||
self.assertFalse(cache.is_writable(surface_document, "marked"))
|
||||
|
||||
def test_a_written_envelope_that_carries_extra_slots_is_held_read_only(self) -> None:
|
||||
path, cache = self._store("extra-slot-store.cc")
|
||||
envelope = cache.get_required_envelope(surface_document, "clean")
|
||||
carrying = type(envelope)(**{**envelope.__dict__, "key": "arrived", "payload": _vector("extra-slot-arm.msgpack")})
|
||||
cache.put_envelope(surface_document, carrying)
|
||||
self.assertFalse(cache.is_writable(surface_document, "arrived"))
|
||||
with self.assertRaises(SchemaConflictError):
|
||||
cache.delete(surface_document, "arrived")
|
||||
cache.pull_all_backing_stores()
|
||||
self.assertFalse(cache.is_writable(surface_document, "arrived"))
|
||||
|
||||
def test_a_batch_of_envelopes_that_carry_extra_slots_is_held_read_only(self) -> None:
|
||||
path, cache = self._store("extra-slot-store.cc")
|
||||
envelope = cache.get_required_envelope(surface_document, "clean")
|
||||
batch = [
|
||||
type(envelope)(**{**envelope.__dict__, "key": key, "payload": _vector(file)})
|
||||
for key, file in (("b1", "extra-slot-arm.msgpack"), ("b2", "union-surface.msgpack"))
|
||||
]
|
||||
cache.put_envelopes(surface_document, batch)
|
||||
self.assertEqual([cache.is_writable(surface_document, key) for key in ("b1", "b2")], [False, True])
|
||||
|
||||
def test_a_mark_under_an_id_the_reader_does_not_list_compatible_is_not_consented(self) -> None:
|
||||
unlisted = define_database_entry_type(
|
||||
"tests.union_surface",
|
||||
[database_entry_field("id", 0, codec=t.string)] + [database_entry_field(n, i, codec=c) for n, i, c in (
|
||||
("version", 1, t.int32), ("root", 2, _parity._node), ("outcomes", 3, t.list(_parity.outcome)),
|
||||
("part", 4, t.nullable(_parity.item)), ("crafted", 5, t.nullable(_parity.crafted)), ("missing", 6, _parity._node))],
|
||||
name="id",
|
||||
schema_id="tests.union_surface.v2",
|
||||
schema_name="tests.union_surface",
|
||||
)
|
||||
path = self.tmp / "unlisted.cc"
|
||||
_copy_store_without_variants(VECTORS / "extra-slot-consent.cc", path)
|
||||
cache = _open(path, unlisted)
|
||||
self.assertFalse(cache.is_writable(unlisted, "consented"))
|
||||
|
||||
def test_a_written_envelope_is_judged_on_the_id_it_is_stored_under_so_a_reload_agrees(self) -> None:
|
||||
path, cache = self._store("extra-slot-consent.cc")
|
||||
arrived = next(e for e in SingleFileMessagePackBackingStore(path).pull_all() if e.key == "consented")
|
||||
self.assertEqual(arrived.schema_id, "tests.union_surface.wide")
|
||||
cache.put_envelope(declaring_document, dataclasses.replace(arrived, key="via-envelope"))
|
||||
cache.put_envelopes(declaring_document, [dataclasses.replace(arrived, key="via-batch")])
|
||||
in_memory = [cache.is_writable(declaring_document, k) for k in ("via-envelope", "via-batch")]
|
||||
cache.pull_all_backing_stores()
|
||||
self.assertEqual([cache.is_writable(declaring_document, k) for k in ("via-envelope", "via-batch")], in_memory)
|
||||
self.assertEqual(in_memory, [False, False])
|
||||
|
||||
def test_is_writable_agrees_with_the_write_for_a_key_a_store_carries_foreign(self) -> None:
|
||||
path = self.tmp / "foreign.cc"
|
||||
_copy_store_without_variants(VECTORS / "extra-slot-store.cc", path)
|
||||
cache = _open(path, surface_document) # tests.slot_unit is not registered: its record is carried foreign
|
||||
self.assertEqual([r.key for r in cache.foreign_records()], ["unit-arm"])
|
||||
self.assertFalse(cache.is_writable(surface_document, "unit-arm"))
|
||||
with self.assertRaises(SchemaConflictError):
|
||||
cache.put(surface_document, "unit-arm", _surface_value())
|
||||
with self.assertRaises(SchemaConflictError):
|
||||
cache.delete(surface_document, "unit-arm")
|
||||
|
||||
def test_every_key_of_a_batch_is_checked_wherever_the_marked_one_stands(self) -> None:
|
||||
path, cache = self._store("extra-slot-store.cc")
|
||||
before = path.read_bytes()
|
||||
envelope = cache.get_required_envelope(surface_document, "clean")
|
||||
fresh = dataclasses.replace(envelope, key="fresh")
|
||||
marked = dataclasses.replace(envelope, key="marked")
|
||||
for batch in ([fresh, marked], [marked, fresh], [fresh, dataclasses.replace(envelope, key="f2"), marked]):
|
||||
with self.assertRaises(SchemaConflictError) as caught:
|
||||
cache.put_envelopes(surface_document, batch)
|
||||
self.assertEqual(caught.exception.record_key, "marked")
|
||||
self.assertEqual(path.read_bytes(), before)
|
||||
self.assertIsNone(cache.get(surface_document, "fresh"))
|
||||
|
||||
def test_only_the_marks_that_hold_a_record_read_only_are_kept_by_key(self) -> None:
|
||||
_, store = self._store("extra-slot-store.cc")
|
||||
self.assertEqual({k: v[0] for k, v in store._state.unread.items()}, {"marked": surface_document.type, "unit-arm": "tests.slot_unit"})
|
||||
_, consent = self._store("extra-slot-consent.cc")
|
||||
self.assertEqual(sorted(consent._state.unread), ["nested", "over-consent"]) # the consented mark is no state
|
||||
|
||||
def test_an_arrival_with_no_catalog_entry_under_a_listed_id_is_not_consented(self) -> None:
|
||||
wide = next(e for e in SingleFileMessagePackBackingStore(VECTORS / "extra-slot-consent.cc").pull_all() if e.key == "consented")
|
||||
bare = dataclasses.replace(wide, catalog_entry=None)
|
||||
|
||||
class Held:
|
||||
def pull_all(self):
|
||||
return [bare]
|
||||
|
||||
def push(self, envelope): ...
|
||||
|
||||
def push_all(self, envelopes): ...
|
||||
|
||||
def delete(self, type, key): ...
|
||||
|
||||
cache = CultCache.builder().register_document_type(declaring_document).add_generic_store(Held()).build()
|
||||
cache.pull_all_backing_stores()
|
||||
self.assertFalse(cache.is_writable(declaring_document, "consented"))
|
||||
|
||||
def test_a_removal_is_refused_where_is_writable_says_no(self) -> None:
|
||||
path, cache = self._store("extra-slot-store.cc")
|
||||
before = path.read_bytes()
|
||||
other = define_database_entry_type("tests.other", [database_entry_field("a", 0, codec=t.string)], name="a")
|
||||
cache2 = _open(self.tmp / "extra-slot-store.cc", surface_document, slot_unit_document, other)
|
||||
self.assertFalse(cache2.is_writable(other, "clean"))
|
||||
with self.assertRaises(SchemaConflictError):
|
||||
cache2.delete(other, "clean")
|
||||
self.assertEqual(path.read_bytes(), before)
|
||||
self.assertIsNotNone(cache2.get(surface_document, "clean"))
|
||||
self.assertTrue(cache2.is_writable(other, "nothing-here"))
|
||||
|
||||
def test_a_global_document_is_not_writable_at_a_key_other_than_the_global_one(self) -> None:
|
||||
solo = define_database_entry_type("tests.solo", [database_entry_field("a", 0, codec=t.string)], global_document=True)
|
||||
cache = CultCache.builder().register_document_type(solo).build()
|
||||
self.assertFalse(cache.is_writable(solo, "elsewhere"))
|
||||
with self.assertRaises(CultCacheError):
|
||||
cache.put(solo, "elsewhere", {"a": "x"})
|
||||
with self.assertRaises(CultCacheError):
|
||||
cache.delete(solo, "elsewhere")
|
||||
self.assertTrue(cache.is_writable(solo, CultCache.GLOBAL_KEY))
|
||||
|
||||
def test_a_reload_replaces_the_marks_wholesale(self) -> None:
|
||||
path, cache = self._store("extra-slot-store.cc")
|
||||
self.assertFalse(cache.is_writable(surface_document, "marked"))
|
||||
# The store is repaired by its owner: the marked record is rewritten clean. The reload must forget the old mark.
|
||||
[snapshot] = list(msgpack.Unpacker(open(path, "rb"), raw=False))
|
||||
clean = next(r for r in snapshot[2] if r[0] == "clean")
|
||||
for record in snapshot[2]:
|
||||
if record[0] in ("marked", "unit-arm"):
|
||||
record[3] = clean[3] if record[0] == "marked" else record[3]
|
||||
snapshot[2] = [r for r in snapshot[2] if r[0] != "unit-arm"]
|
||||
path.write_bytes(msgpack.packb(snapshot, use_bin_type=True))
|
||||
cache.pull_all_backing_stores()
|
||||
self.assertTrue(cache.is_writable(surface_document, "marked"))
|
||||
self.assertEqual(cache._state.unread, {})
|
||||
|
||||
def test_a_listed_id_whose_catalog_reports_no_extra_slot_does_not_consent(self) -> None:
|
||||
wide = next(e for e in SingleFileMessagePackBackingStore(VECTORS / "extra-slot-consent.cc").pull_all() if e.key == "consented")
|
||||
same = dataclasses.replace(wide, catalog_entry=declaring_document.catalog_entry()) # the stored entry has no slot past the reader's
|
||||
|
||||
class Held:
|
||||
def pull_all(self):
|
||||
return [same]
|
||||
|
||||
def push(self, envelope): ...
|
||||
|
||||
def push_all(self, envelopes): ...
|
||||
|
||||
def delete(self, type, key): ...
|
||||
|
||||
cache = CultCache.builder().register_document_type(declaring_document).add_generic_store(Held()).build()
|
||||
cache.pull_all_backing_stores()
|
||||
self.assertFalse(cache.is_writable(declaring_document, "consented"))
|
||||
|
||||
def test_a_record_stored_under_the_readers_own_id_is_never_consented(self) -> None:
|
||||
own = dataclasses.replace(
|
||||
next(e for e in SingleFileMessagePackBackingStore(VECTORS / "extra-slot-consent.cc").pull_all() if e.key == "consented"),
|
||||
schema_id="tests.union_surface.v2",
|
||||
)
|
||||
wider = dataclasses.replace(own.catalog_entry, schema_id="tests.union_surface.v2", compatible_schema_ids=("tests.union_surface.v2",))
|
||||
listing = dataclasses.replace(
|
||||
declaring_document, schema_id="tests.union_surface.v2", compatible_schema_ids=("tests.union_surface.wide", "tests.union_surface.v2")
|
||||
)
|
||||
|
||||
class Held:
|
||||
def pull_all(self):
|
||||
return [dataclasses.replace(own, catalog_entry=wider)]
|
||||
|
||||
def push(self, envelope): ...
|
||||
|
||||
def push_all(self, envelopes): ...
|
||||
|
||||
def delete(self, type, key): ...
|
||||
|
||||
cache = CultCache.builder().register_document_type(listing).add_generic_store(Held()).build()
|
||||
cache.pull_all_backing_stores()
|
||||
self.assertFalse(cache.is_writable(listing, "consented"))
|
||||
|
||||
def test_a_removal_by_the_owner_of_a_key_held_as_another_type_removes_it_as_a_write_would_replace_it(self) -> None:
|
||||
a = define_database_entry_type(
|
||||
"tests.owned_a", [database_entry_field("a", 0, codec=t.string)], name="a", indexes={"by_a": "a"}
|
||||
)
|
||||
b = define_database_entry_type(
|
||||
"tests.owned_b",
|
||||
[database_entry_field("a", 0, codec=t.string)],
|
||||
compatible_schema_ids=(a.catalog_entry().schema_id,),
|
||||
)
|
||||
path = self.tmp / "owned.cc"
|
||||
writer = _open(path, a)
|
||||
writer.put(a, "k", {"a": "x"})
|
||||
writer.put(a, "other", {"a": "y"})
|
||||
cache = _open(path, a, b)
|
||||
self.assertTrue(cache.is_writable(b, "k"))
|
||||
cache.delete(b, "k")
|
||||
self.assertIsNone(cache.get(a, "k"))
|
||||
self.assertIsNone(cache.get_by_index(a, "by_a", "x"))
|
||||
self.assertEqual([e.key for e in SingleFileMessagePackBackingStore(path).pull_all()], ["other"])
|
||||
self.assertTrue(cache.is_writable(b, "k"))
|
||||
|
||||
def test_an_owners_removal_is_one_store_call_so_the_cache_and_the_store_agree_when_it_fails(self) -> None:
|
||||
a = define_database_entry_type("tests.one_a", [database_entry_field("a", 0, codec=t.string)], name="a")
|
||||
b = define_database_entry_type(
|
||||
"tests.one_b", [database_entry_field("a", 0, codec=t.string)], compatible_schema_ids=(a.catalog_entry().schema_id,)
|
||||
)
|
||||
path = self.tmp / "one.cc"
|
||||
_open(path, a).put(a, "k", {"a": "x"})
|
||||
|
||||
class SecondDeleteFails(SingleFileMessagePackBackingStore):
|
||||
calls = 0
|
||||
|
||||
def delete(self, type, key):
|
||||
self.calls += 1
|
||||
if self.calls > 1:
|
||||
raise OSError("second delete")
|
||||
super().delete(type, key)
|
||||
|
||||
store = SecondDeleteFails(path)
|
||||
cache = CultCache.builder().register_document_type(a).register_document_type(b).add_generic_store(store).build()
|
||||
cache.pull_all_backing_stores()
|
||||
cache.delete(b, "k")
|
||||
self.assertEqual(store.calls, 1)
|
||||
self.assertIsNone(cache.get(a, "k"))
|
||||
self.assertEqual(SingleFileMessagePackBackingStore(path).pull_all(), [])
|
||||
|
||||
def test_writing_another_record_and_flushing_carries_the_marked_one_unchanged(self) -> None:
|
||||
path, cache = self._store("extra-slot-store.cc")
|
||||
original = {e.key: e for e in SingleFileMessagePackBackingStore(path).pull_all()}
|
||||
cache.put(surface_document, "clean", {**_surface_value(), "version": 9})
|
||||
after = {e.key: e for e in SingleFileMessagePackBackingStore(path).pull_all()}
|
||||
for key in ("marked", "unit-arm"):
|
||||
self.assertEqual(after[key].payload, original[key].payload, key)
|
||||
self.assertEqual(after[key].schema_id, original[key].schema_id, key)
|
||||
self.assertEqual(after[key].stored_at, original[key].stored_at, key)
|
||||
self.assertNotEqual(after["clean"].payload, original["clean"].payload)
|
||||
self.assertEqual(after["marked"].catalog_entry, original["marked"].catalog_entry)
|
||||
|
||||
def test_the_vector_store_loads_under_the_reference_documents_and_marks_nothing_else(self) -> None:
|
||||
path, cache = self._store("extra-slot-store.cc")
|
||||
self.assertTrue(cache.is_writable(surface_document, "clean"))
|
||||
self.assertFalse(cache.is_writable(surface_document, "marked"))
|
||||
self.assertFalse(cache.is_writable(slot_unit_document, "unit-arm"))
|
||||
self.assertEqual(
|
||||
cache.get(slot_unit_document, "unit-arm"), {"name": "unit-arm", "outcome": UnionArm("Denied")}
|
||||
)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
|
|
@ -261,7 +261,9 @@ class UnionSurfaceVectorTests(unittest.TestCase):
|
|||
else:
|
||||
read_only += 1
|
||||
self.assertEqual(outcome_kind, "read-only", file)
|
||||
self.assertEqual(surface_document.decode_payload(payload), _surface_value(), file)
|
||||
value, mark = surface_document.decode_payload_marked(payload)
|
||||
self.assertEqual(value, _surface_value(), file)
|
||||
self.assertIsNotNone(mark, file) # read past and marked: tests/test_slot_readonly.py pins what the mark holds
|
||||
self.assertEqual((refused, read_only), (5, 4))
|
||||
|
||||
|
||||
|
|
@ -526,7 +528,9 @@ class BoundsVectorTests(unittest.TestCase):
|
|||
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)
|
||||
value, mark = surface_document.decode_payload_marked(payload)
|
||||
self.assertEqual(value, _surface_value(), file)
|
||||
self.assertIsNotNone(mark, file)
|
||||
else:
|
||||
with self.assertRaises(ValueError, msg=file) as caught:
|
||||
surface_document.decode_payload(payload)
|
||||
|
|
|
|||
Loading…
Reference in a new issue