diff --git a/apps/predbat/lattice_complete_inventory_fragment.py b/apps/predbat/lattice_complete_inventory_fragment.py new file mode 100644 index 000000000..7c09d1ccd --- /dev/null +++ b/apps/predbat/lattice_complete_inventory_fragment.py @@ -0,0 +1,721 @@ +# ----------------------------------------------------------------------------- +# Predbat Home Battery System - Complete inventory Lattice fragment adapter +# Copyright Trefor Southwell 2026 - All Rights Reserved +# This application maybe used for personal use only and not for commercial use +# ----------------------------------------------------------------------------- +"""Pure, default-off publisher for complete provider inventory snapshots. + +The caller owns discovery and supplies a complete immutable site/device view +at a monotonically increasing discovery generation. A newer accepted view +replaces the provider's previous view, so an omitted site or device is no +longer present in that provider-local fragment. + +Inventory observations are REFERENCE-only. Provider site/device identifiers +remain local aliases and never correlate nodes globally. Only an explicitly +API-verified hardware serial can produce a strong ``ProviderIdentityAlias``. +No capability, configuration projection, role assignment, endpoint, or +control authority is inferred. + +This module composes the existing durable fragment contracts. It imports no +live integration and performs no discovery or registration by itself. +""" + +# cspell:ignore autoconfig + +import re +import threading +import unicodedata +from collections.abc import Mapping +from dataclasses import dataclass +from itertools import islice +from typing import Optional + +from lattice_autoconfig import ( + AliasRole, + ProviderAlias, + ProviderHealth, + ProviderIdentityAlias, + ProviderSnapshot, + _plain, +) +from lattice_fragment_adapters import ( + DurableFragmentAdapter, + FragmentAdapterConflict, + FragmentAdapterReadError, + FragmentAdapterRemoved, +) + + +INVENTORY_HEALTH_UNCHANGED = object() +MAX_COMPLETE_INVENTORY_SITES = 1024 +MAX_COMPLETE_INVENTORY_DEVICES = 16384 +_IDENTIFIER_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]*$") +_PROVIDER_ID_RE = re.compile(r"^[a-z0-9][a-z0-9._-]*$") +_DEVICE_KINDS = frozenset( + ( + "battery", + "energy-management-system", + "ev-charger", + "gateway", + "inverter", + "meter", + "sensor", + ) +) + + +def _bounded_text(value, name, maximum): + """Return normalized non-empty text with an explicit size bound.""" + if type(value) is not str or not value.strip(): + raise ValueError("{} must be a non-empty string".format(name)) + if any(unicodedata.category(character) in ("Cc", "Cf", "Cs") for character in value): + raise ValueError( + "{} must not contain control characters or malformed Unicode".format( + name, + ), + ) + value = value.strip() + if len(value) > maximum: + raise ValueError("{} must be at most {} characters".format(name, maximum)) + return value + + +def _provider_id(value): + """Normalize one bounded canonical integration provider ID.""" + value = _bounded_text(value, "provider_id", 128) + if _PROVIDER_ID_RE.fullmatch(value) is None: + raise ValueError( + "provider_id must use lowercase identifier characters", + ) + return value + + +def _provider_identifier(value, name): + """Normalize one bounded provider-local string or integer identifier.""" + if type(value) is int: + value = str(value) + value = _bounded_text(value, name, 128) + if _IDENTIFIER_RE.fullmatch(value) is None: + raise ValueError("{} must be a provider identifier".format(name)) + return value + + +def _optional_text(value, name, maximum=128): + """Return one normalized optional bounded text field.""" + if value is None: + return None + return _bounded_text(value, name, maximum) + + +def _optional_bool(value, name): + """Accept only explicit boolean or unknown state.""" + if value is None or isinstance(value, bool): + return value + raise ValueError("{} must be True, False, or None".format(name)) + + +def _discovery_generation(value): + """Validate one caller-owned monotonic discovery generation.""" + if type(value) is not int or value < 0: + raise ValueError( + "discovery_generation must be a non-negative integer", + ) + return value + + +def _provider_health(value): + """Normalize explicit integration liveness into compiler health.""" + if isinstance(value, ProviderHealth): + return value + if value is True: + return ProviderHealth.HEALTHY + if value is False: + return ProviderHealth.OFFLINE + if value is None: + return ProviderHealth.DEGRADED + raise ValueError("health must be ProviderHealth, True, False, or None") + + +def _bounded_observations(values, name, observation_type, maximum): + """Collect at most max+1 typed observations without retaining source faults.""" + expected = observation_type.__name__ + if isinstance(values, (str, bytes, bytearray, Mapping)): + raise ValueError( + "{} must be an iterable of {}".format(name, expected), + ) + invalid_iterable = False + try: + observations = tuple( + islice( + iter(values), + maximum + 1, + ) + ) + except Exception: + invalid_iterable = True + if invalid_iterable: + raise ValueError( + "{} must be an iterable of {}".format(name, expected), + ) + if len(observations) > maximum: + raise ValueError( + "{} must contain at most {} observations".format( + name, + maximum, + ), + ) + if any(type(observation) is not observation_type for observation in observations): + raise ValueError( + "{} must contain only {} values".format( + name, + expected, + ), + ) + return observations + + +@dataclass(frozen=True) +class InventorySiteObservation: + """One provider-local site/station observation.""" + + site_id: object + online: Optional[bool] = None + + def __post_init__(self): + """Normalize bounded site state without creating strong identity.""" + object.__setattr__( + self, + "site_id", + _provider_identifier(self.site_id, "site_id"), + ) + object.__setattr__( + self, + "online", + _optional_bool(self.online, "online"), + ) + + def node_id(self, provider_id): + """Return one stable provider-qualified site node.""" + return "inventory:{}:site:{}".format(provider_id, self.site_id) + + +@dataclass(frozen=True) +class InventoryDeviceObservation: + """One provider-local device with optional verified hardware identity.""" + + device_id: object + site_id: object + kind: str + model: Optional[str] = None + online: Optional[bool] = None + verified_hardware_serial: Optional[str] = None + + def __post_init__(self): + """Normalize metadata and retain only an explicitly verified serial.""" + device_id = _provider_identifier(self.device_id, "device_id") + site_id = _provider_identifier(self.site_id, "site_id") + kind = _bounded_text(self.kind, "kind", 64).lower().replace("_", "-") + if kind not in _DEVICE_KINDS: + raise ValueError( + "kind must be one of {}".format( + ", ".join(sorted(_DEVICE_KINDS)), + ) + ) + serial = self.verified_hardware_serial + if serial is not None: + serial = _provider_identifier( + serial, + "verified_hardware_serial", + ).upper() + object.__setattr__(self, "device_id", device_id) + object.__setattr__(self, "site_id", site_id) + object.__setattr__(self, "kind", kind) + object.__setattr__( + self, + "model", + _optional_text(self.model, "model"), + ) + object.__setattr__( + self, + "online", + _optional_bool(self.online, "online"), + ) + object.__setattr__( + self, + "verified_hardware_serial", + serial, + ) + + def node_id(self, provider_id): + """Return one stable provider-qualified device node.""" + return "inventory:{}:device:{}".format( + provider_id, + self.device_id, + ) + + +def _normalize_sites(sites): + """Freeze, sort, and collision-check a complete site set.""" + sites = _bounded_observations( + sites, + "sites", + InventorySiteObservation, + MAX_COMPLETE_INVENTORY_SITES, + ) + canonical_sites = [] + for site in sites: + raw_site_id = site.site_id + raw_online = site.online + clean = InventorySiteObservation( + site_id=raw_site_id, + online=raw_online, + ) + if (raw_site_id, raw_online) != ( + clean.site_id, + clean.online, + ): + raise ValueError( + "complete inventory site observation is not canonical", + ) + canonical_sites.append(clean) + sites = tuple(canonical_sites) + seen = set() + for site in sites: + if site.site_id in seen: + raise FragmentAdapterConflict( + "complete inventory contains a duplicate site_id", + ) + seen.add(site.site_id) + return tuple(sorted(sites, key=lambda item: item.site_id)) + + +def _normalize_devices(devices, site_ids): + """Freeze, sort, and reject ambiguous device or serial ownership.""" + devices = _bounded_observations( + devices, + "devices", + InventoryDeviceObservation, + MAX_COMPLETE_INVENTORY_DEVICES, + ) + canonical_devices = [] + for device in devices: + raw_device_id = device.device_id + raw_site_id = device.site_id + raw_kind = device.kind + raw_model = device.model + raw_online = device.online + raw_serial = device.verified_hardware_serial + clean = InventoryDeviceObservation( + device_id=raw_device_id, + site_id=raw_site_id, + kind=raw_kind, + model=raw_model, + online=raw_online, + verified_hardware_serial=raw_serial, + ) + if ( + raw_device_id, + raw_site_id, + raw_kind, + raw_model, + raw_online, + raw_serial, + ) != ( + clean.device_id, + clean.site_id, + clean.kind, + clean.model, + clean.online, + clean.verified_hardware_serial, + ): + raise ValueError( + "complete inventory device observation is not canonical", + ) + canonical_devices.append(clean) + devices = tuple(canonical_devices) + + device_owners = {} + serial_owners = {} + for device in devices: + if device.site_id not in site_ids: + raise ValueError( + "complete inventory device references an unknown site_id", + ) + owner = device_owners.get(device.device_id) + if owner is not None: + raise FragmentAdapterConflict( + "complete inventory contains a duplicate device_id", + ) + device_owners[device.device_id] = device.site_id + serial = device.verified_hardware_serial + if serial is not None: + owner = serial_owners.get(serial) + if owner is not None and owner != device.device_id: + raise FragmentAdapterConflict( + "complete inventory contains duplicate " "verified_hardware_serial ownership", + ) + serial_owners[serial] = device.device_id + return tuple( + sorted( + devices, + key=lambda item: ( + item.site_id, + item.device_id, + item.kind, + ), + ) + ) + + +def _site_node(site, provider_id): + """Project one provider-local site without authority.""" + attributes = { + "inventorySiteId": site.site_id, + "identityScope": "provider-local", + } + if site.online is not None: + attributes["online"] = site.online + return { + "id": site.node_id(provider_id), + "kind": "site", + "attributes": attributes, + "accessPaths": [ + { + "id": "inventory-read", + "provider": provider_id, + "preference": 0, + } + ], + "capabilities": [], + } + + +def _device_node(device, provider_id): + """Project bounded device observations without deriving capabilities.""" + attributes = { + "inventoryDeviceId": device.device_id, + "inventorySiteId": device.site_id, + "identityScope": "provider-local", + } + if device.model is not None: + attributes["model"] = device.model + if device.online is not None: + attributes["online"] = device.online + if device.verified_hardware_serial is not None: + attributes["serial"] = device.verified_hardware_serial + attributes["serialEvidence"] = "api-verified" + return { + "id": device.node_id(provider_id), + "kind": device.kind, + "attributes": attributes, + "accessPaths": [ + { + "id": "inventory-read", + "provider": provider_id, + "preference": 0, + } + ], + "capabilities": [], + } + + +def _topology_document( + provider_id, + producer_name, + discovery_generation, + sites, + devices, +): + """Build one deterministic complete provider inventory fragment.""" + nodes = [_site_node(site, provider_id) for site in sites] + nodes.extend(_device_node(device, provider_id) for device in devices) + relationships = [ + { + "from": "inventory:{}:site:{}".format( + provider_id, + device.site_id, + ), + "to": device.node_id(provider_id), + "type": "contains", + } + for device in devices + ] + document = { + "topologyVersion": "0.3.0", + "scope": "fragment", + "docVersion": discovery_generation, + "producer": { + "name": producer_name, + "provider": provider_id, + "authority": 0, + }, + "nodes": nodes, + } + if relationships: + document["relationships"] = relationships + return document + + +def _reference_aliases(provider_id, sites, devices): + """Publish provider-local names with REFERENCE role only.""" + aliases = [ + ProviderAlias( + name="site:{}".format(site.site_id), + node_id=site.node_id(provider_id), + roles=frozenset((AliasRole.REFERENCE,)), + ) + for site in sites + ] + aliases.extend( + ProviderAlias( + name="device:{}".format(device.device_id), + node_id=device.node_id(provider_id), + roles=frozenset((AliasRole.REFERENCE,)), + ) + for device in devices + ) + return tuple( + sorted( + aliases, + key=lambda item: (item.name, item.node_id), + ) + ) + + +def _identity_aliases(provider_id, devices): + """Publish only explicitly API-verified hardware serial assertions.""" + return tuple( + sorted( + ( + ProviderIdentityAlias( + kind="serial", + value=device.verified_hardware_serial, + node_id=device.node_id(provider_id), + ) + for device in devices + if device.verified_hardware_serial is not None + ), + key=lambda item: (item.kind, item.value, item.node_id), + ) + ) + + +class CompleteInventoryFragmentPublisher: + """Publish complete immutable site/device inventory generations.""" + + def __init__( + self, + provider_id, + producer_name, + state_store, + enabled=False, + ): + """Create an unwired publisher; disabled is the safe default.""" + if not isinstance(enabled, bool): + raise ValueError("enabled must be a boolean") + provider_id = _provider_id(provider_id) + self._producer_name = _bounded_text( + producer_name, + "producer_name", + 128, + ) + self._enabled = enabled + self._adapter = DurableFragmentAdapter(provider_id, state_store) + self.provider_id = self._adapter.provider_id + self._lock = threading.RLock() + try: + self._adapter.read_state() + except FragmentAdapterReadError as exc: + expected = "provider {} has no durable fragment".format( + self.provider_id, + ) + if str(exc) != expected: + raise + self._seeded = False + else: + self._seeded = True + + @property + def enabled(self): + """Return whether this explicitly constructed publisher accepts input.""" + return self._enabled + + @property + def generation(self): + """Fresh-read the current durable adapter generation.""" + return self.read_state().generation + + @property + def semantic_fingerprint(self): + """Fresh-read the generation-bound semantic fingerprint.""" + return self.read_state().semantic_fingerprint + + @property + def discovery_generation(self): + """Fresh-read the current caller-owned discovery generation.""" + return _plain( + self.read_state().snapshot.topology_fragment, + )["docVersion"] + + def lattice_fragment_adapter(self): + """Expose structural discovery only when enabled and seeded.""" + if not self._enabled or not self._seeded: + return None + self.read_state() + return self + + def read_state(self): + """Fresh-read the complete durable adapter state.""" + return self._adapter.read_state() + + def read_snapshot(self): + """Fresh-read the current immutable provider snapshot.""" + return self._adapter.read_snapshot() + + def subscribe_invalidation(self, listener): + """Subscribe to inventory, liveness, and removal changes.""" + return self._adapter.subscribe_invalidation(listener) + + def _current_state(self): + """Return current state, treating an unseeded store as empty.""" + if not self._seeded: + return None + return self._adapter.read_state() + + def ingest_inventory( + self, + discovery_generation, + sites, + devices, + health=INVENTORY_HEALTH_UNCHANGED, + feedback_token=None, + ): + """Publish one complete site/device inventory replacement.""" + if not self._enabled: + return False + discovery_generation = _discovery_generation( + discovery_generation, + ) + sites = _normalize_sites(sites) + site_ids = frozenset(site.site_id for site in sites) + devices = _normalize_devices(devices, site_ids) + document = _topology_document( + self.provider_id, + self._producer_name, + discovery_generation, + sites, + devices, + ) + + with self._lock: + current = self._current_state() + if current is not None and current.removed: + raise FragmentAdapterRemoved( + "provider {} was removed at generation {}; complete " + "inventory cannot re-enrol it".format( + self.provider_id, + current.generation, + ) + ) + if current is None: + next_health = ProviderHealth.DEGRADED if health is INVENTORY_HEALTH_UNCHANGED else _provider_health(health) + else: + previous = _plain(current.snapshot.topology_fragment) + previous_generation = previous["docVersion"] + if discovery_generation < previous_generation: + return False + if discovery_generation == previous_generation and document != previous: + raise FragmentAdapterConflict( + "provider {} reused complete-inventory discovery " + "generation {} for different content".format( + self.provider_id, + discovery_generation, + ) + ) + next_health = current.snapshot.health if health is INVENTORY_HEALTH_UNCHANGED else _provider_health(health) + if discovery_generation == previous_generation and next_health is current.snapshot.health: + return False + + generation = 1 if current is None else current.generation + 1 + snapshot = ProviderSnapshot( + provider_id=self.provider_id, + generation=generation, + health=next_health, + topology_fragment=document, + aliases=_reference_aliases( + self.provider_id, + sites, + devices, + ), + identity_aliases=_identity_aliases( + self.provider_id, + devices, + ), + role_assignments=(), + config_projections=(), + ) + published = self._adapter.publish( + snapshot, + "complete inventory changed to discovery generation " "{}".format(discovery_generation), + feedback_token=feedback_token, + ) + if published: + self._seeded = True + return published + + def set_liveness(self, health, feedback_token=None): + """Publish provider liveness without changing inventory contents.""" + if not self._enabled: + return False + health = _provider_health(health) + with self._lock: + current = self._current_state() + if current is None: + raise FragmentAdapterReadError( + "complete inventory must be seeded before liveness", + ) + if current.removed: + raise FragmentAdapterRemoved( + "provider {} was removed at generation {}".format( + self.provider_id, + current.generation, + ) + ) + if current.snapshot.health is health: + return False + previous = current.snapshot + snapshot = ProviderSnapshot( + provider_id=self.provider_id, + generation=current.generation + 1, + health=health, + topology_fragment=_plain(previous.topology_fragment), + aliases=previous.aliases, + identity_aliases=previous.identity_aliases, + role_assignments=(), + config_projections=(), + ) + return self._adapter.publish( + snapshot, + "complete inventory liveness changed to {}".format( + health.value, + ), + feedback_token=feedback_token, + ) + + def remove(self, feedback_token=None): + """Publish an irreversible whole-provider removal tombstone.""" + if not self._enabled: + return False + with self._lock: + current = self._current_state() + if current is None: + raise FragmentAdapterReadError( + "complete inventory must be seeded before removal", + ) + if current.removed: + return False + return self._adapter.remove( + current.generation + 1, + "complete inventory integration removed", + feedback_token=feedback_token, + ) diff --git a/apps/predbat/lattice_deye_inventory_fragment.py b/apps/predbat/lattice_deye_inventory_fragment.py new file mode 100644 index 000000000..8c9775c7b --- /dev/null +++ b/apps/predbat/lattice_deye_inventory_fragment.py @@ -0,0 +1,204 @@ +# ----------------------------------------------------------------------------- +# Predbat Home Battery System - Deye complete-inventory Lattice adapter +# Copyright Trefor Southwell 2026 - All Rights Reserved +# This application maybe used for personal use only and not for commercial use +# ----------------------------------------------------------------------------- +"""Pure Deye station/inverter inventory normalization. + +The live Deye component is intentionally not imported. A future gated seam +may pass explicit station/device observations here only after a successful +cloud discovery. This module publishes no TOU, scheduler, write, endpoint, +feature, authentication, or control claims. + +Station and device IDs stay provider-local. The separate +``api_verified_serial`` field is the only input allowed to become a strong +cross-provider hardware identity. +""" + +from dataclasses import dataclass +from typing import Optional + +from lattice_complete_inventory_fragment import ( + INVENTORY_HEALTH_UNCHANGED, + MAX_COMPLETE_INVENTORY_DEVICES, + MAX_COMPLETE_INVENTORY_SITES, + CompleteInventoryFragmentPublisher, + InventoryDeviceObservation, + InventorySiteObservation, + _bounded_observations, + _optional_bool, + _optional_text, + _provider_identifier, +) + + +@dataclass(frozen=True) +class DeyeStationObservation: + """One explicit Deye station returned by cloud discovery.""" + + station_id: object + online: Optional[bool] = None + + def __post_init__(self): + """Normalize only provider-local station state.""" + site = InventorySiteObservation( + site_id=self.station_id, + online=self.online, + ) + object.__setattr__(self, "station_id", site.site_id) + object.__setattr__(self, "online", site.online) + + def inventory_site(self): + """Return the generic immutable site observation.""" + return InventorySiteObservation( + site_id=self.station_id, + online=self.online, + ) + + +@dataclass(frozen=True) +class DeyeInverterObservation: + """One explicit Deye inverter assigned to a discovered station.""" + + device_id: object + station_id: object + api_verified_serial: Optional[str] = None + model: Optional[str] = None + online: Optional[bool] = None + + def __post_init__(self): + """Normalize bounded metadata without inferring any capabilities.""" + device_id = _provider_identifier(self.device_id, "device_id") + station_id = _provider_identifier(self.station_id, "station_id") + serial = self.api_verified_serial + if serial is not None: + serial = _provider_identifier( + serial, + "api_verified_serial", + ).upper() + object.__setattr__(self, "device_id", device_id) + object.__setattr__(self, "station_id", station_id) + object.__setattr__(self, "api_verified_serial", serial) + object.__setattr__( + self, + "model", + _optional_text(self.model, "model"), + ) + object.__setattr__( + self, + "online", + _optional_bool(self.online, "online"), + ) + + def inventory_device(self): + """Return the generic immutable inverter observation.""" + return InventoryDeviceObservation( + device_id=self.device_id, + site_id=self.station_id, + kind="inverter", + model=self.model, + online=self.online, + verified_hardware_serial=self.api_verified_serial, + ) + + +def _validated_deye_station(station): + """Return one exact canonical Deye station reconstruction.""" + if type(station) is not DeyeStationObservation: + raise ValueError("Deye station observation has an invalid type") + raw_station_id = station.station_id + raw_online = station.online + clean = DeyeStationObservation( + station_id=raw_station_id, + online=raw_online, + ) + if (raw_station_id, raw_online) != ( + clean.station_id, + clean.online, + ): + raise ValueError("Deye station observation is not canonical") + return clean + + +def _validated_deye_inverter(inverter): + """Return one exact canonical Deye inverter reconstruction.""" + if type(inverter) is not DeyeInverterObservation: + raise ValueError("Deye inverter observation has an invalid type") + raw_device_id = inverter.device_id + raw_station_id = inverter.station_id + raw_serial = inverter.api_verified_serial + raw_model = inverter.model + raw_online = inverter.online + clean = DeyeInverterObservation( + device_id=raw_device_id, + station_id=raw_station_id, + api_verified_serial=raw_serial, + model=raw_model, + online=raw_online, + ) + if ( + raw_device_id, + raw_station_id, + raw_serial, + raw_model, + raw_online, + ) != ( + clean.device_id, + clean.station_id, + clean.api_verified_serial, + clean.model, + clean.online, + ): + raise ValueError("Deye inverter observation is not canonical") + return clean + + +class DeyeInventoryFragmentPublisher(CompleteInventoryFragmentPublisher): + """Default-off publisher for explicit complete Deye inventories.""" + + def __init__(self, provider_id, state_store, enabled=False): + """Create an unwired Deye inventory publisher.""" + super().__init__( + provider_id, + "Deye Cloud Inventory", + state_store, + enabled=enabled, + ) + + def ingest_snapshot( + self, + discovery_generation, + stations, + inverters, + health=INVENTORY_HEALTH_UNCHANGED, + feedback_token=None, + ): + """Normalize and publish one complete Deye discovery generation.""" + if not self.enabled: + return False + stations = _bounded_observations( + stations, + "stations", + DeyeStationObservation, + MAX_COMPLETE_INVENTORY_SITES, + ) + inverters = _bounded_observations( + inverters, + "inverters", + DeyeInverterObservation, + MAX_COMPLETE_INVENTORY_DEVICES, + ) + stations = tuple(_validated_deye_station(station) for station in stations) + inverters = tuple(_validated_deye_inverter(inverter) for inverter in inverters) + return self.ingest_inventory( + discovery_generation, + (DeyeStationObservation.inventory_site(station) for station in stations), + ( + DeyeInverterObservation.inventory_device( + inverter, + ) + for inverter in inverters + ), + health=health, + feedback_token=feedback_token, + ) diff --git a/apps/predbat/lattice_fragment_adapters.py b/apps/predbat/lattice_fragment_adapters.py index 22f21e1d5..38a6ddb42 100644 --- a/apps/predbat/lattice_fragment_adapters.py +++ b/apps/predbat/lattice_fragment_adapters.py @@ -18,14 +18,25 @@ # cspell:ignore autoconfig idempotently unsubscribers import hashlib +import math import threading from dataclasses import dataclass from functools import partial +from itertools import islice from types import MappingProxyType from typing import Optional, Protocol from lattice_autoconfig import ( + AliasRole, + ProjectionCardinality, + ProjectionRouting, + ProjectionValueKind, + ProviderAlias, + ProviderConfigProjection, ProviderHealth, + ProviderIdentityAlias, + ProviderProjectionValue, + ProviderRoleAssignment, ProviderSnapshot, _fingerprint_snapshot, _plain, @@ -49,22 +60,33 @@ class FragmentAdapterRemoved(FragmentAdapterReadError): """A registered integration has published a durable removal tombstone.""" +_MAX_SNAPSHOT_DEPTH = 32 +_MAX_SNAPSHOT_CONTAINER_ITEMS = 65536 +_MAX_SNAPSHOT_TOTAL_ITEMS = 1000000 +_MAX_SNAPSHOT_SEQUENCE_ITEMS = 32768 +_MAX_SNAPSHOT_STRING_CHARS = 16384 +_MAX_SNAPSHOT_INTEGER = (1 << 63) - 1 +_MAX_SNAPSHOT_FLOAT = 1.0e308 +_MAX_ALIAS_ROLES = 3 +_MAX_PROJECTION_TRANSFORMS = 64 + + def _validate_provider_id(provider_id): """Normalize one provider-owned stable identifier.""" - if not isinstance(provider_id, str) or not provider_id.strip(): + if type(provider_id) is not str or not provider_id.strip(): raise ValueError("provider_id must be a non-empty string") return provider_id.strip() def _validate_generation(generation): """Validate one monotonically increasing integration generation.""" - if not isinstance(generation, int) or isinstance(generation, bool) or generation < 0: + if type(generation) is not int or generation < 0: raise ValueError("generation must be a non-negative integer") def _validate_reason(reason): """Normalize one auditable invalidation reason.""" - if not isinstance(reason, str) or not reason.strip(): + if type(reason) is not str or not reason.strip(): raise ValueError("reason must be a non-empty string") return reason.strip() @@ -77,6 +99,472 @@ def _semantic_fingerprint(snapshot, removed=False): return fingerprint +def _bounded_snapshot_string(value, *, non_empty=False): + """Return one exact bounded string without invoking subclass hooks.""" + if type(value) is not str: + raise ValueError("snapshot text must be a string") + if len(value) > _MAX_SNAPSHOT_STRING_CHARS: + raise ValueError("snapshot text exceeds its size bound") + if non_empty and not value.strip(): + raise ValueError("snapshot text must not be empty") + return value + + +def _bounded_snapshot_integer(value, *, non_negative=False): + """Return one exact bounded integer.""" + if type(value) is not int: + raise ValueError("snapshot integer must be an integer") + if abs(value) > _MAX_SNAPSHOT_INTEGER: + raise ValueError("snapshot integer exceeds its size bound") + if non_negative and value < 0: + raise ValueError("snapshot integer must not be negative") + return value + + +def _bounded_snapshot_float(value): + """Return one exact finite bounded float.""" + if type(value) is not float: + raise ValueError("snapshot float must be a float") + if not math.isfinite(value) or abs(value) > _MAX_SNAPSHOT_FLOAT: + raise ValueError("snapshot float must be finite and bounded") + return value + + +def _bounded_exact_tuple(value, maximum, budget=None): + """Collect one exact tuple through a max+1 bounded iterator.""" + if type(value) is not tuple: + raise ValueError("snapshot sequence must be a tuple") + items = tuple(islice(iter(value), maximum + 1)) + if len(items) > maximum: + raise ValueError("snapshot sequence exceeds its item bound") + if budget is not None: + _consume_snapshot_budget( + budget, + len(items), + ) + return items + + +def _consume_snapshot_budget(budget, count=1): + """Account for recursively reconstructed snapshot values.""" + budget[0] += count + if budget[0] > _MAX_SNAPSHOT_TOTAL_ITEMS: + raise ValueError("snapshot exceeds its total item bound") + + +def _copy_bounded_snapshot_json(value, budget, depth=0): + """Copy exact frozen JSON into safe built-in containers with hard bounds.""" + if depth > _MAX_SNAPSHOT_DEPTH: + raise ValueError("snapshot exceeds its nesting bound") + _consume_snapshot_budget(budget) + + if type(value) is MappingProxyType: + entries = tuple( + islice( + iter(value.items()), + _MAX_SNAPSHOT_CONTAINER_ITEMS + 1, + ) + ) + if len(entries) > _MAX_SNAPSHOT_CONTAINER_ITEMS: + raise ValueError("snapshot mapping exceeds its item bound") + copied = {} + for key, item in entries: + key = _bounded_snapshot_string(key) + _consume_snapshot_budget(budget) + copied[key] = _copy_bounded_snapshot_json( + item, + budget, + depth + 1, + ) + return copied + + if type(value) is tuple: + items = tuple( + islice( + iter(value), + _MAX_SNAPSHOT_CONTAINER_ITEMS + 1, + ) + ) + if len(items) > _MAX_SNAPSHOT_CONTAINER_ITEMS: + raise ValueError("snapshot sequence exceeds its item bound") + return tuple( + _copy_bounded_snapshot_json( + item, + budget, + depth + 1, + ) + for item in items + ) + + if type(value) is str: + return _bounded_snapshot_string(value) + if type(value) is bool or value is None: + return value + if type(value) is int: + return _bounded_snapshot_integer(value) + if type(value) is float: + return _bounded_snapshot_float(value) + raise ValueError("snapshot contains an unsupported JSON value") + + +def _validated_alias(alias, budget): + """Reconstruct one exact provider alias.""" + if type(alias) is not ProviderAlias: + raise ValueError("snapshot alias has an invalid type") + name = alias.name + node_id = alias.node_id + roles_value = alias.roles + name = _bounded_snapshot_string(name, non_empty=True) + node_id = _bounded_snapshot_string(node_id, non_empty=True) + if type(roles_value) is not frozenset: + raise ValueError("snapshot alias roles must be a frozenset") + roles = tuple( + islice( + iter(roles_value), + _MAX_ALIAS_ROLES + 1, + ) + ) + if not roles or len(roles) > _MAX_ALIAS_ROLES or any(type(role) is not AliasRole for role in roles): + raise ValueError("snapshot alias roles are invalid") + _consume_snapshot_budget( + budget, + len(roles), + ) + clean = ProviderAlias( + name=name, + node_id=node_id, + roles=frozenset(roles), + ) + if (name, node_id, frozenset(roles)) != ( + clean.name, + clean.node_id, + clean.roles, + ): + raise ValueError("snapshot alias is not canonical") + return clean + + +def _validated_identity_alias(alias, _budget): + """Reconstruct one exact provider identity assertion.""" + if type(alias) is not ProviderIdentityAlias: + raise ValueError("snapshot identity alias has an invalid type") + kind = alias.kind + value = alias.value + node_id = alias.node_id + clean = ProviderIdentityAlias( + kind=_bounded_snapshot_string( + kind, + non_empty=True, + ), + value=_bounded_snapshot_string( + value, + non_empty=True, + ), + node_id=_bounded_snapshot_string( + node_id, + non_empty=True, + ), + ) + if (kind, value, node_id) != ( + clean.kind, + clean.value, + clean.node_id, + ): + raise ValueError("snapshot identity alias is not canonical") + return clean + + +def _validated_role_assignment(assignment, _budget): + """Reconstruct one exact provider role assignment.""" + if type(assignment) is not ProviderRoleAssignment: + raise ValueError("snapshot role assignment has an invalid type") + role = assignment.role + group = assignment.group + index = assignment.index + node_id = assignment.node_id + if type(role) is not AliasRole: + raise ValueError("snapshot role assignment has an invalid role") + clean = ProviderRoleAssignment( + role=role, + group=_bounded_snapshot_string( + group, + non_empty=True, + ), + index=_bounded_snapshot_integer( + index, + non_negative=True, + ), + node_id=_bounded_snapshot_string( + node_id, + non_empty=True, + ), + ) + if (role, group, index, node_id) != ( + clean.role, + clean.group, + clean.index, + clean.node_id, + ): + raise ValueError("snapshot role assignment is not canonical") + return clean + + +def _validated_optional_snapshot_string(value): + """Reconstruct one optional exact non-empty string.""" + if value is None: + return None + return _bounded_snapshot_string(value, non_empty=True) + + +def _validated_projection_constant(value): + """Reconstruct one bounded exact JSON scalar projection value.""" + if type(value) is str: + return _bounded_snapshot_string(value) + if type(value) is bool: + return value + if type(value) is int: + return _bounded_snapshot_integer(value) + if type(value) is float: + return _bounded_snapshot_float(value) + raise ValueError("snapshot projection constant has an invalid type") + + +def _validated_projection_value(value, _budget): + """Reconstruct one exact provider projection value.""" + if type(value) is not ProviderProjectionValue: + raise ValueError("snapshot projection value has an invalid type") + node_id = value.node_id + kind = value.kind + raw_value = value.value + capability = value.capability + identity_kind = value.identity_kind + identity_value = value.identity_value + access_path_id = value.access_path_id + if type(kind) is not ProjectionValueKind: + raise ValueError("snapshot projection value has an invalid kind") + + if kind is ProjectionValueKind.ENTITY: + clean_value = _bounded_snapshot_string( + raw_value, + non_empty=True, + ) + elif kind is ProjectionValueKind.CONSTANT: + clean_value = _validated_projection_constant(raw_value) + else: + if raw_value is not None: + raise ValueError("snapshot None projection carries a value") + clean_value = None + + clean = ProviderProjectionValue( + node_id=_bounded_snapshot_string( + node_id, + non_empty=True, + ), + kind=kind, + value=clean_value, + capability=_validated_optional_snapshot_string( + capability, + ), + identity_kind=_validated_optional_snapshot_string( + identity_kind, + ), + identity_value=_validated_optional_snapshot_string( + identity_value, + ), + access_path_id=_validated_optional_snapshot_string( + access_path_id, + ), + ) + if ( + node_id, + kind, + raw_value, + capability, + identity_kind, + identity_value, + access_path_id, + ) != ( + clean.node_id, + clean.kind, + clean.value, + clean.capability, + clean.identity_kind, + clean.identity_value, + clean.access_path_id, + ): + raise ValueError("snapshot projection value is not canonical") + return clean + + +def _validated_config_projection(projection, budget): + """Reconstruct one exact provider configuration projection.""" + if type(projection) is not ProviderConfigProjection: + raise ValueError("snapshot projection has an invalid type") + argument = projection.argument + role = projection.role + group = projection.group + routing = projection.routing + cardinality = projection.cardinality + raw_values = projection.values + required = projection.required + raw_transforms = projection.transforms + if type(role) is not AliasRole: + raise ValueError("snapshot projection has an invalid role") + if type(routing) is not ProjectionRouting: + raise ValueError("snapshot projection has invalid routing") + if type(cardinality) is not ProjectionCardinality: + raise ValueError("snapshot projection has invalid cardinality") + if type(required) is not bool: + raise ValueError("snapshot projection required must be boolean") + + values = tuple( + _validated_projection_value( + value, + budget, + ) + for value in _bounded_exact_tuple( + raw_values, + _MAX_SNAPSHOT_SEQUENCE_ITEMS, + budget, + ) + ) + transforms = tuple( + _bounded_snapshot_string( + transform, + non_empty=True, + ) + for transform in _bounded_exact_tuple( + raw_transforms, + _MAX_PROJECTION_TRANSFORMS, + budget, + ) + ) + clean = ProviderConfigProjection( + argument=_bounded_snapshot_string( + argument, + non_empty=True, + ), + role=role, + group=_bounded_snapshot_string( + group, + non_empty=True, + ), + routing=routing, + cardinality=cardinality, + values=values, + required=required, + transforms=transforms, + ) + if ( + argument, + role, + group, + routing, + cardinality, + raw_values, + required, + raw_transforms, + ) != ( + clean.argument, + clean.role, + clean.group, + clean.routing, + clean.cardinality, + clean.values, + clean.required, + clean.transforms, + ): + raise ValueError("snapshot projection is not canonical") + return clean + + +def _validated_provider_snapshot(snapshot): + """Return a bounded exact reconstruction of one provider snapshot.""" + if type(snapshot) is not ProviderSnapshot: + raise ValueError("snapshot must be ProviderSnapshot") + raw_provider_id = snapshot.provider_id + raw_generation = snapshot.generation + health = snapshot.health + raw_topology_fragment = snapshot.topology_fragment + raw_aliases = snapshot.aliases + raw_identity_aliases = snapshot.identity_aliases + raw_role_assignments = snapshot.role_assignments + raw_config_projections = snapshot.config_projections + provider_id = _bounded_snapshot_string( + raw_provider_id, + non_empty=True, + ) + generation = _bounded_snapshot_integer( + raw_generation, + non_negative=True, + ) + if type(health) is not ProviderHealth: + raise ValueError("snapshot health must be ProviderHealth") + + budget = [0] + topology_fragment = _copy_bounded_snapshot_json( + raw_topology_fragment, + budget, + ) + aliases = tuple( + _validated_alias( + alias, + budget, + ) + for alias in _bounded_exact_tuple( + raw_aliases, + _MAX_SNAPSHOT_SEQUENCE_ITEMS, + budget, + ) + ) + identity_aliases = tuple( + _validated_identity_alias( + alias, + budget, + ) + for alias in _bounded_exact_tuple( + raw_identity_aliases, + _MAX_SNAPSHOT_SEQUENCE_ITEMS, + budget, + ) + ) + role_assignments = tuple( + _validated_role_assignment( + assignment, + budget, + ) + for assignment in _bounded_exact_tuple( + raw_role_assignments, + _MAX_SNAPSHOT_SEQUENCE_ITEMS, + budget, + ) + ) + config_projections = tuple( + _validated_config_projection( + projection, + budget, + ) + for projection in _bounded_exact_tuple( + raw_config_projections, + _MAX_SNAPSHOT_SEQUENCE_ITEMS, + budget, + ) + ) + clean = ProviderSnapshot( + provider_id=provider_id, + generation=generation, + health=health, + topology_fragment=topology_fragment, + aliases=aliases, + identity_aliases=identity_aliases, + role_assignments=role_assignments, + config_projections=config_projections, + ) + if raw_provider_id != clean.provider_id: + raise ValueError("snapshot provider_id is not canonical") + return clean + + @dataclass(frozen=True) class FragmentAdapterState: """One integration-owned durable fragment cursor and immutable value. @@ -97,55 +585,109 @@ def __post_init__(self): """Validate that the durable cursor exactly binds its snapshot.""" provider_id = _validate_provider_id(self.provider_id) _validate_generation(self.generation) - if not isinstance(self.snapshot, ProviderSnapshot): + if type(self.semantic_fingerprint) is not str: + raise ValueError("semantic_fingerprint must be a string") + if type(self.snapshot) is not ProviderSnapshot: raise ValueError("snapshot must be ProviderSnapshot") + if type(self.snapshot.provider_id) is not str: + raise ValueError("snapshot provider_id must be a string") + if type(self.snapshot.generation) is not int: + raise ValueError("snapshot generation must be an integer") if self.snapshot.provider_id != provider_id: raise ValueError("snapshot provider_id does not match durable state") if self.snapshot.generation != self.generation: raise ValueError("snapshot generation does not match durable state") - if not isinstance(self.removed, bool): + if type(self.removed) is not bool: raise ValueError("removed must be a boolean") - expected = _semantic_fingerprint(self.snapshot, self.removed) + fingerprint_failed = False + try: + expected = _semantic_fingerprint( + self.snapshot, + self.removed, + ) + except Exception: + fingerprint_failed = True + if fingerprint_failed: + raise ValueError("snapshot semantic fingerprint is invalid") if self.semantic_fingerprint != expected: raise ValueError("semantic_fingerprint does not match the immutable snapshot") object.__setattr__(self, "provider_id", provider_id) +def _validated_fragment_state(state): + """Revalidate one exact durable state without subclass dispatch.""" + if type(state) is not FragmentAdapterState: + raise ValueError("state must be FragmentAdapterState") + raw_provider_id = state.provider_id + provider_id = _validate_provider_id(raw_provider_id) + if raw_provider_id != provider_id: + raise ValueError("durable provider_id is not canonical") + _validate_generation(state.generation) + if type(state.semantic_fingerprint) is not str: + raise ValueError("semantic_fingerprint must be a string") + if type(state.removed) is not bool: + raise ValueError("removed must be a boolean") + snapshot = _validated_provider_snapshot(state.snapshot) + if snapshot.provider_id != provider_id: + raise ValueError("snapshot provider_id does not match durable state") + if snapshot.generation != state.generation: + raise ValueError("snapshot generation does not match durable state") + expected = _semantic_fingerprint( + snapshot, + state.removed, + ) + if state.semantic_fingerprint != expected: + raise ValueError("semantic_fingerprint does not match the immutable snapshot") + return FragmentAdapterState( + provider_id=provider_id, + generation=state.generation, + semantic_fingerprint=expected, + snapshot=snapshot, + removed=state.removed, + ) + + def _compiler_fragment_snapshot(adapter): """Fresh-read one compiler input, translating only durable removals.""" - state = adapter.read_state() - if not isinstance(state, FragmentAdapterState): - raise FragmentAdapterReadError("adapter read_state must return FragmentAdapterState") + validation_failed = False try: - state.__post_init__() - except ValueError as exc: - raise FragmentAdapterReadError(str(exc)) from exc - provider_id = _validate_provider_id(getattr(adapter, "provider_id", None)) - if state.provider_id != provider_id: - raise FragmentAdapterReadError("adapter state belongs to provider {}".format(state.provider_id)) - if not state.removed: - return state.snapshot - return ProviderSnapshot( - provider_id=state.provider_id, - generation=state.generation, - health=ProviderHealth.HEALTHY, - topology_fragment={ - "topologyVersion": "0.3.0", - "scope": "fragment", - "docVersion": state.generation, - "producer": { - "name": "PredBat fragment tombstone", - "provider": state.provider_id, - "authority": 0, - }, - "nodes": [], - "relationships": [], - }, - aliases=(), - identity_aliases=(), - role_assignments=(), - config_projections=(), - ) + state = _validated_fragment_state(adapter.read_state()) + provider_id = _validate_provider_id( + getattr(adapter, "provider_id", None), + ) + if state.provider_id != provider_id: + raise ValueError("adapter state provider mismatch") + if not state.removed: + snapshot = state.snapshot + else: + snapshot = ProviderSnapshot( + provider_id=state.provider_id, + generation=state.generation, + health=ProviderHealth.HEALTHY, + topology_fragment={ + "topologyVersion": "0.3.0", + "scope": "fragment", + "docVersion": state.generation, + "producer": { + "name": "PredBat fragment tombstone", + "provider": state.provider_id, + "authority": 0, + }, + "nodes": [], + "relationships": [], + }, + aliases=(), + identity_aliases=(), + role_assignments=(), + config_projections=(), + ) + except Exception: + validation_failed = True + if validation_failed: + raise FragmentAdapterReadError( + "compiler fragment state failed validation", + ) + return snapshot class FragmentAdapterStateStore: @@ -238,28 +780,32 @@ def __init__(self, provider_id, state_store): def _load(self): """Load durable state and convert store faults into adapter faults.""" + load_failed = False try: - return self._state_store.load() - except Exception as exc: + state = self._state_store.load() + except Exception: + load_failed = True + if load_failed: raise FragmentAdapterReadError( - "durable fragment load failed: {}: {}".format( - type(exc).__name__, - exc, - ) - ) from exc + "durable fragment load failed", + ) + return state def _validate_loaded_state(self, state): """Validate one store result and its provider ownership.""" if state is None: return None - if not isinstance(state, FragmentAdapterState): - raise FragmentAdapterReadError("state_store.load must return FragmentAdapterState or None") + validation_failed = False try: - state.__post_init__() - except ValueError as exc: - raise FragmentAdapterReadError(str(exc)) from exc - if state.provider_id != self.provider_id: - raise FragmentAdapterReadError("durable state belongs to provider {}".format(state.provider_id)) + state = _validated_fragment_state(state) + if state.provider_id != self.provider_id: + raise ValueError("durable state provider mismatch") + except Exception: + validation_failed = True + if validation_failed: + raise FragmentAdapterReadError( + "durable fragment state failed validation", + ) return state def read_state(self): @@ -318,20 +864,29 @@ def publish( A subscriber may return ``False`` only to suppress publication-origin feedback; that token then causes neither persistence nor recompilation. """ - if not isinstance(snapshot, ProviderSnapshot): - raise ValueError("snapshot must be ProviderSnapshot") - if snapshot.provider_id != self.provider_id: - raise ValueError("snapshot provider_id does not match adapter") - if not isinstance(removed, bool): - raise ValueError("removed must be a boolean") reason = _validate_reason(reason) - candidate = FragmentAdapterState( - self.provider_id, - snapshot.generation, - _semantic_fingerprint(snapshot, removed), - snapshot, - removed, - ) + candidate_failed = False + try: + snapshot = _validated_provider_snapshot(snapshot) + provider_id = _validate_provider_id(snapshot.provider_id) + _validate_generation(snapshot.generation) + if type(removed) is not bool: + raise ValueError("removed must be a boolean") + if provider_id != self.provider_id: + raise ValueError("snapshot provider_id does not match adapter") + candidate = FragmentAdapterState( + self.provider_id, + snapshot.generation, + _semantic_fingerprint(snapshot, removed), + snapshot, + removed, + ) + except Exception: + candidate_failed = True + if candidate_failed: + raise ValueError( + "fragment snapshot failed validation", + ) with self._lock: current = self._validate_loaded_state(self._load()) @@ -436,19 +991,27 @@ def readers(self): def _validate_adapter(self, adapter): """Validate the common adapter surface and its durable current state.""" - provider_id = _validate_provider_id(getattr(adapter, "provider_id", None)) - for method_name in ( - "read_state", - "read_snapshot", - "subscribe_invalidation", - ): - if not callable(getattr(adapter, method_name, None)): - raise ValueError("fragment adapter must provide {}".format(method_name)) - state = adapter.read_state() - if not isinstance(state, FragmentAdapterState): - raise ValueError("adapter read_state must return FragmentAdapterState") - if state.provider_id != provider_id: - raise ValueError("adapter state belongs to provider {}".format(state.provider_id)) + validation_failed = False + try: + provider_id = _validate_provider_id( + getattr(adapter, "provider_id", None), + ) + for method_name in ( + "read_state", + "read_snapshot", + "subscribe_invalidation", + ): + if not callable(getattr(adapter, method_name, None)): + raise ValueError("fragment adapter method is missing") + state = _validated_fragment_state(adapter.read_state()) + if state.provider_id != provider_id: + raise ValueError("adapter state provider mismatch") + except Exception: + validation_failed = True + if validation_failed: + raise ValueError( + "fragment adapter failed validation", + ) return provider_id def discover(self, components): diff --git a/apps/predbat/tests/test_lattice_complete_inventory_fragment.py b/apps/predbat/tests/test_lattice_complete_inventory_fragment.py new file mode 100644 index 000000000..63bdfeec8 --- /dev/null +++ b/apps/predbat/tests/test_lattice_complete_inventory_fragment.py @@ -0,0 +1,978 @@ +"""Tests for the generic complete-inventory Lattice publisher.""" + +# cspell:ignore autoconfig + +import os +import sys +import unittest + +sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..")) + +from lattice_autoconfig import ( # noqa: E402 + AliasRole, + CompileStatus, + ProviderHealth, + compile_auto_config, +) +from lattice_compiled_publication import ( # noqa: E402 + InMemoryCompiledLatticeStateStore, +) +from lattice_complete_inventory_fragment import ( # noqa: E402 + CompleteInventoryFragmentPublisher, + InventoryDeviceObservation, + InventorySiteObservation, + MAX_COMPLETE_INVENTORY_DEVICES, + MAX_COMPLETE_INVENTORY_SITES, +) +from lattice_fragment_adapters import ( # noqa: E402 + FragmentAdapterConflict, + FragmentAdapterReadError, + FragmentAdapterRegistry, + FragmentAdapterRemoved, + InMemoryFragmentAdapterStateStore, +) + + +def site(site_id="SITE-A", online=True): + """Build one provider-local site observation.""" + return InventorySiteObservation(site_id=site_id, online=online) + + +def device( + device_id="DEVICE-A", + site_id="SITE-A", + kind="inverter", + model="Model A", + online=True, + verified_hardware_serial=None, +): + """Build one provider-local device observation.""" + return InventoryDeviceObservation( + device_id=device_id, + site_id=site_id, + kind=kind, + model=model, + online=online, + verified_hardware_serial=verified_hardware_serial, + ) + + +def publisher(enabled=True, state_store=None): + """Build one generic publisher and durable store.""" + state_store = state_store or InMemoryFragmentAdapterStateStore() + return ( + CompleteInventoryFragmentPublisher( + "inventory-provider", + "Inventory Provider", + state_store, + enabled=enabled, + ), + state_store, + ) + + +def assert_value_free_validation_error(test_case, action, raw_value): + """Assert one public validation error retains no provider source text.""" + try: + action() + except ValueError as error: + test_case.assertIs(type(error), ValueError) + test_case.assertNotIn(raw_value, str(error)) + test_case.assertNotIn(raw_value, repr(error)) + test_case.assertNotIn(raw_value, repr(error.args)) + test_case.assertNotIn(raw_value, error.args) + test_case.assertFalse(hasattr(error, "object")) + test_case.assertIsNone(error.__cause__) + test_case.assertIsNone(error.__context__) + else: + test_case.fail("expected a value-free ValueError") + + +class HostileText(str): + """String subclass whose scalar hooks expose a fake provider secret.""" + + secret = "provider-secret-value" + + def _fail(self, *_args, **_kwargs): + """Fail if validation invokes any subclass-controlled hook.""" + raise RuntimeError(self.secret) + + strip = _fail + lower = _fail + upper = _fail + __format__ = _fail + __iter__ = _fail + __len__ = _fail + __hash__ = _fail + __eq__ = _fail + __lt__ = _fail + + +class HostileInt(int): + """Integer subclass whose scalar hooks expose a fake provider secret.""" + + secret = HostileText.secret + + def _fail(self, *_args, **_kwargs): + """Fail if validation invokes any subclass-controlled hook.""" + raise RuntimeError(self.secret) + + __str__ = _fail + __format__ = _fail + __hash__ = _fail + __eq__ = _fail + __lt__ = _fail + + +class InventorySiteObservationSubclass(InventorySiteObservation): + """Observation subclass that must not cross the public boundary.""" + + +class InventoryDeviceObservationSubclass(InventoryDeviceObservation): + """Observation subclass that must not cross the public boundary.""" + + +class TestInventoryObservations(unittest.TestCase): + """Inventory inputs are closed, bounded, and provider-local.""" + + def test_site_and_device_identifiers_are_canonical(self): + """Integer site IDs normalize without becoming strong identities.""" + observed_site = site(1234, online=None) + observed_device = device( + device_id="dev_1", + site_id=1234, + kind="energy_management_system", + model=None, + online=None, + ) + + self.assertEqual(observed_site.site_id, "1234") + self.assertEqual(observed_device.site_id, "1234") + self.assertEqual( + observed_device.kind, + "energy-management-system", + ) + self.assertIsNone(observed_device.verified_hardware_serial) + + def test_metadata_rejects_controls_and_unbounded_values(self): + """Control characters and unbounded metadata cannot be serialized.""" + with self.assertRaisesRegex(ValueError, "provider identifier"): + site("site with spaces") + with self.assertRaisesRegex(ValueError, "control characters"): + device(model="model\0secret") + with self.assertRaisesRegex(ValueError, "at most 128"): + device(model="m" * 129) + with self.assertRaisesRegex(ValueError, "kind must be"): + device(kind="tou-schedule-controller") + with self.assertRaisesRegex(ValueError, "online"): + device(online=1) + + def test_model_rejects_unicode_controls_and_surrogates(self): + """Invisible controls and malformed Unicode cannot enter metadata.""" + invalid_models = ( + ("bidi override", "Model\u202eA"), + ("zero-width", "Model\u200bA"), + ("unpaired surrogate", "Model\ud800A"), + ) + for label, invalid_model in invalid_models: + with self.subTest(label=label): + with self.assertRaisesRegex( + ValueError, + "control characters or malformed Unicode", + ): + device(model=invalid_model) + + def test_producer_name_rejects_unicode_controls_and_surrogates(self): + """Publisher provenance cannot contain invisible or malformed text.""" + invalid_names = ( + ("bidi override", "Inventory\u202eProvider"), + ("zero-width", "Inventory\u200bProvider"), + ("unpaired surrogate", "Inventory\ud800Provider"), + ) + for label, invalid_name in invalid_names: + with self.subTest(label=label): + with self.assertRaisesRegex( + ValueError, + "control characters or malformed Unicode", + ): + CompleteInventoryFragmentPublisher( + "inventory-provider", + invalid_name, + InMemoryFragmentAdapterStateStore(), + ) + + def test_valid_unicode_metadata_is_preserved_exactly(self): + """Valid provider evidence is not silently Unicode-normalized.""" + decomposed_model = "Cafe\u0301 Model" + + self.assertEqual(device(model=decomposed_model).model, decomposed_model) + + def test_provider_id_is_bounded_and_canonical(self): + """Provider qualification cannot contain paths or controls.""" + store = InMemoryFragmentAdapterStateStore() + with self.assertRaisesRegex(ValueError, "lowercase"): + CompleteInventoryFragmentPublisher( + "Inventory Provider", + "Inventory", + store, + ) + with self.assertRaisesRegex(ValueError, "control"): + CompleteInventoryFragmentPublisher( + "inventory\0provider", + "Inventory", + store, + ) + + def test_scalar_subclasses_are_rejected_before_any_hook_runs(self): + """Provider scalar subclasses cannot execute or retain source text.""" + cases = ( + lambda: site(HostileInt(1)), + lambda: device(device_id=HostileText("DEVICE")), + lambda: device(site_id=HostileText("SITE")), + lambda: device(kind=HostileText("inverter")), + lambda: device(model=HostileText("Model")), + lambda: device( + verified_hardware_serial=HostileText("SERIAL"), + ), + lambda: CompleteInventoryFragmentPublisher( + HostileText("inventory-provider"), + "Inventory Provider", + InMemoryFragmentAdapterStateStore(), + ), + lambda: CompleteInventoryFragmentPublisher( + "inventory-provider", + HostileText("Inventory Provider"), + InMemoryFragmentAdapterStateStore(), + ), + ) + for action in cases: + with self.subTest(action=action): + assert_value_free_validation_error( + self, + action, + HostileText.secret, + ) + + +class TestCompleteInventoryFragmentPublisher(unittest.TestCase): + """Complete inventories replace one durable provider-local view.""" + + def test_default_off_is_unseeded_unregistered_and_write_free(self): + """Disabled input is not validated, persisted, or discoverable.""" + adapter, state_store = publisher(enabled=False) + + self.assertIsNone(adapter.lattice_fragment_adapter()) + self.assertFalse( + adapter.ingest_inventory( + -1, + (object(),), + (object(),), + ) + ) + self.assertFalse(adapter.set_liveness(True)) + self.assertFalse(adapter.remove()) + self.assertEqual(state_store.writes, 0) + with self.assertRaises(FragmentAdapterReadError): + adapter.read_state() + + def test_generation_and_observation_subclasses_fail_before_write(self): + """Subclass hooks cannot bypass complete-inventory normalization.""" + cases = ( + lambda adapter: adapter.ingest_inventory( + HostileInt(1), + (), + (), + ), + lambda adapter: adapter.ingest_inventory( + 1, + (InventorySiteObservationSubclass("SITE-A"),), + (), + ), + lambda adapter: adapter.ingest_inventory( + 1, + (site(),), + ( + InventoryDeviceObservationSubclass( + "DEVICE-A", + "SITE-A", + "inverter", + ), + ), + ), + ) + for action in cases: + with self.subTest(action=action): + adapter, state_store = publisher() + assert_value_free_validation_error( + self, + lambda: action(adapter), + HostileText.secret, + ) + self.assertEqual(state_store.writes, 0) + + def test_post_init_mutation_is_revalidated_before_set_sort_or_format(self): + """Frozen observation mutation cannot retain or execute source data.""" + mutated_site = site() + object.__setattr__( + mutated_site, + "site_id", + HostileText("SITE-A"), + ) + mutated_device = device() + object.__setattr__( + mutated_device, + "model", + HostileText("Model"), + ) + cases = ( + ( + (mutated_site,), + (), + ), + ( + (site(),), + (mutated_device,), + ), + ) + for sites, devices in cases: + with self.subTest( + sites=sites, + devices=devices, + ): + adapter, state_store = publisher() + assert_value_free_validation_error( + self, + lambda: adapter.ingest_inventory( + 1, + sites, + devices, + ), + HostileText.secret, + ) + self.assertEqual(state_store.writes, 0) + + def test_post_init_normalization_cannot_heal_inventory_observations(self): + """Whitespace and case mutations fail before durable publication.""" + + def site_id(): + candidate = site() + object.__setattr__( + candidate, + "site_id", + " SITE-A ", + ) + return (candidate,), (), " SITE-A " + + def device_id(): + candidate = device() + object.__setattr__( + candidate, + "device_id", + " DEVICE-A ", + ) + return (site(),), (candidate,), " DEVICE-A " + + def device_site_id(): + candidate = device() + object.__setattr__( + candidate, + "site_id", + " SITE-A ", + ) + return (site(),), (candidate,), " SITE-A " + + def device_kind(): + candidate = device() + object.__setattr__( + candidate, + "kind", + "INVERTER", + ) + return (site(),), (candidate,), "INVERTER" + + def device_model(): + candidate = device() + object.__setattr__( + candidate, + "model", + " Model A ", + ) + return (site(),), (candidate,), " Model A " + + def device_serial(): + candidate = device( + verified_hardware_serial="SERIAL-A", + ) + object.__setattr__( + candidate, + "verified_hardware_serial", + "serial-a", + ) + return (site(),), (candidate,), "serial-a" + + for build_case in ( + site_id, + device_id, + device_site_id, + device_kind, + device_model, + device_serial, + ): + with self.subTest(build_case=build_case): + sites, devices, raw_value = build_case() + adapter, state_store = publisher() + assert_value_free_validation_error( + self, + lambda: adapter.ingest_inventory( + 1, + sites, + devices, + ), + raw_value, + ) + self.assertEqual(state_store.writes, 0) + + def test_multi_site_inventory_is_deterministic_reference_only(self): + """Order cannot affect topology or manufacture control authority.""" + adapter, state_store = publisher() + sites = [site("SITE-B", online=None), site("SITE-A")] + devices = [ + device( + "METER-B", + "SITE-B", + kind="meter", + model=None, + online=None, + ), + device( + "INV-A", + "SITE-A", + verified_hardware_serial="serial-a", + ), + ] + + self.assertTrue( + adapter.ingest_inventory( + 7, + reversed(sites), + reversed(devices), + health=True, + ) + ) + sites.append(site("MUTATED")) + devices.append(device("MUTATED")) + snapshot = adapter.read_snapshot() + topology = snapshot.topology_fragment + + self.assertIs(adapter.lattice_fragment_adapter(), adapter) + self.assertEqual(adapter.generation, 1) + self.assertEqual(adapter.discovery_generation, 7) + self.assertEqual(state_store.writes, 1) + self.assertEqual(snapshot.health, ProviderHealth.HEALTHY) + self.assertEqual( + tuple(node["id"] for node in topology["nodes"]), + ( + "inventory:inventory-provider:site:SITE-A", + "inventory:inventory-provider:site:SITE-B", + "inventory:inventory-provider:device:INV-A", + "inventory:inventory-provider:device:METER-B", + ), + ) + self.assertEqual( + tuple( + ( + relationship["from"], + relationship["to"], + relationship["type"], + ) + for relationship in topology["relationships"] + ), + ( + ( + "inventory:inventory-provider:site:SITE-A", + "inventory:inventory-provider:device:INV-A", + "contains", + ), + ( + "inventory:inventory-provider:site:SITE-B", + "inventory:inventory-provider:device:METER-B", + "contains", + ), + ), + ) + self.assertEqual(topology["producer"]["authority"], 0) + self.assertTrue(all(node["capabilities"] == () for node in topology["nodes"])) + self.assertTrue(all(alias.roles == frozenset((AliasRole.REFERENCE,)) for alias in snapshot.aliases)) + self.assertEqual(snapshot.role_assignments, ()) + self.assertEqual(snapshot.config_projections, ()) + self.assertEqual( + tuple((alias.kind, alias.value) for alias in snapshot.identity_aliases), + (("serial", "SERIAL-A"),), + ) + + plan = compile_auto_config((snapshot,)) + self.assertEqual(plan.role_assignments, ()) + self.assertEqual(plan.config_arguments, ()) + self.assertFalse(plan.materialization_readiness.ready) + + def test_newer_complete_inventory_removes_omitted_observations(self): + """A disappearing device is absent, not retained from an older view.""" + adapter, _state_store = publisher() + adapter.ingest_inventory( + 1, + (site("SITE-A"), site("SITE-B")), + ( + device("INV-A", "SITE-A", verified_hardware_serial="SER-A"), + device("INV-B", "SITE-B", verified_hardware_serial="SER-B"), + ), + health=True, + ) + registry = FragmentAdapterRegistry(enabled=True) + self.assertEqual( + registry.discover((adapter,)), + ("inventory-provider",), + ) + compiler = registry.create_compiler( + InMemoryCompiledLatticeStateStore(), + ) + self.assertEqual(compiler.drain().status, CompileStatus.FRESH) + + self.assertTrue( + adapter.ingest_inventory( + 2, + (site("SITE-A"),), + ( + device( + "INV-A", + "SITE-A", + verified_hardware_serial="SER-A", + ), + ), + ) + ) + snapshot = adapter.read_snapshot() + serialized = repr(snapshot) + + self.assertNotIn("INV-B", serialized) + self.assertNotIn("SER-B", serialized) + self.assertNotIn("SITE-B", serialized) + self.assertEqual(adapter.generation, 2) + self.assertEqual(adapter.discovery_generation, 2) + compiled = compiler.drain() + self.assertEqual(compiled.status, CompileStatus.FRESH) + self.assertEqual( + compiled.publication.provider_generations, + (("inventory-provider", 2),), + ) + + def test_empty_complete_inventory_removes_every_observation(self): + """An explicit empty view is distinct from provider removal.""" + adapter, _state_store = publisher() + adapter.ingest_inventory( + 1, + (site(),), + (device(),), + health=True, + ) + + self.assertTrue(adapter.ingest_inventory(2, (), ())) + snapshot = adapter.read_snapshot() + self.assertEqual(snapshot.topology_fragment["nodes"], ()) + self.assertNotIn( + "relationships", + snapshot.topology_fragment, + ) + self.assertEqual(snapshot.aliases, ()) + self.assertEqual(snapshot.identity_aliases, ()) + self.assertFalse(adapter.read_state().removed) + + def test_generation_replay_regression_and_conflict(self): + """Replay is inert; stale input is ignored; mutation fails closed.""" + adapter, state_store = publisher() + original_sites = (site(),) + original_devices = (device(),) + adapter.ingest_inventory( + 5, + original_sites, + original_devices, + health=True, + ) + before = adapter.read_state() + + self.assertFalse( + adapter.ingest_inventory( + 5, + reversed(original_sites), + reversed(original_devices), + ) + ) + self.assertFalse( + adapter.ingest_inventory( + 4, + (site(online=False),), + (device(model="Stale"),), + health=False, + ) + ) + with self.assertRaisesRegex( + FragmentAdapterConflict, + "reused complete-inventory discovery generation 5", + ): + adapter.ingest_inventory( + 5, + original_sites, + (device(model="Different"),), + ) + self.assertEqual(adapter.read_state(), before) + self.assertEqual(state_store.writes, 1) + + def test_collisions_and_unknown_membership_fail_before_write(self): + """Ambiguous local and strong identities cannot become durable.""" + adapter, state_store = publisher() + + with self.assertRaisesRegex( + FragmentAdapterConflict, + "^complete inventory contains a duplicate site_id$", + ): + adapter.ingest_inventory( + 1, + (site(), site()), + (), + ) + with self.assertRaisesRegex( + ValueError, + "^complete inventory device references an unknown site_id$", + ): + adapter.ingest_inventory( + 1, + (site(),), + (device(site_id="OTHER"),), + ) + with self.assertRaisesRegex( + FragmentAdapterConflict, + "^complete inventory contains a duplicate device_id$", + ): + adapter.ingest_inventory( + 1, + (site("SITE-A"), site("SITE-B")), + ( + device("DEVICE", "SITE-A"), + device("DEVICE", "SITE-B"), + ), + ) + with self.assertRaisesRegex( + FragmentAdapterConflict, + "^complete inventory contains duplicate " "verified_hardware_serial ownership$", + ): + adapter.ingest_inventory( + 1, + (site(),), + ( + device( + "DEVICE-A", + verified_hardware_serial="shared", + ), + device( + "DEVICE-B", + verified_hardware_serial="SHARED", + ), + ), + ) + self.assertEqual(state_store.writes, 0) + + def test_maximum_inventory_cardinality_is_accepted(self): + """The documented limits still permit unusually large inventories.""" + adapter, state_store = publisher() + sites = tuple(site("SITE-{}".format(index)) for index in range(MAX_COMPLETE_INVENTORY_SITES)) + devices = tuple( + device( + "DEVICE-{}".format(index), + "SITE-0", + ) + for index in range(MAX_COMPLETE_INVENTORY_DEVICES) + ) + + self.assertTrue(adapter.ingest_inventory(1, sites, devices)) + self.assertEqual(state_store.writes, 1) + self.assertEqual( + len(adapter.read_snapshot().topology_fragment["nodes"]), + MAX_COMPLETE_INVENTORY_SITES + MAX_COMPLETE_INVENTORY_DEVICES, + ) + + def test_maximum_plus_one_is_rejected_before_publication(self): + """A finite oversized inventory cannot publish or persist.""" + cases = ( + ( + "sites", + (site("SITE-{}".format(index)) for index in range(MAX_COMPLETE_INVENTORY_SITES + 1)), + (), + MAX_COMPLETE_INVENTORY_SITES, + ), + ( + "devices", + (site(),), + (device("DEVICE-{}".format(index)) for index in range(MAX_COMPLETE_INVENTORY_DEVICES + 1)), + MAX_COMPLETE_INVENTORY_DEVICES, + ), + ) + for label, sites, devices, maximum in cases: + with self.subTest(label=label): + adapter, state_store = publisher() + with self.assertRaisesRegex( + ValueError, + "^{} must contain at most {} observations$".format( + label, + maximum, + ), + ): + adapter.ingest_inventory(1, sites, devices) + self.assertEqual(state_store.writes, 0) + self.assertIsNone(adapter.lattice_fragment_adapter()) + + def test_infinite_inventory_is_bounded_before_validation(self): + """Only max+1 observations are consumed from an infinite source.""" + + def infinite_observations(observation, counter): + while True: + counter.append(None) + yield observation + + cases = ( + ( + "sites", + MAX_COMPLETE_INVENTORY_SITES, + lambda counter: ( + infinite_observations(site(), counter), + (), + ), + ), + ( + "devices", + MAX_COMPLETE_INVENTORY_DEVICES, + lambda counter: ( + (site(),), + infinite_observations(device(), counter), + ), + ), + ) + for label, maximum, inventory in cases: + with self.subTest(label=label): + adapter, state_store = publisher() + counter = [] + sites, devices = inventory(counter) + with self.assertRaisesRegex( + ValueError, + "^{} must contain at most {} observations$".format( + label, + maximum, + ), + ): + adapter.ingest_inventory(1, sites, devices) + self.assertEqual(len(counter), maximum + 1) + self.assertEqual(state_store.writes, 0) + self.assertIsNone(adapter.lattice_fragment_adapter()) + + def test_string_and_mapping_inventories_are_rejected_without_values(self): + """Ambiguous iterable containers fail with stable value-free errors.""" + cases = ( + ("sites", "SITE-SECRET", (), "sites"), + ("sites", {"SITE-SECRET": site()}, (), "sites"), + ("devices", (site(),), "DEVICE-SECRET", "devices"), + ( + "devices", + (site(),), + {"DEVICE-SECRET": device()}, + "devices", + ), + ) + for label, sites, devices, expected_name in cases: + with self.subTest(label=label, container=type(sites).__name__): + adapter, state_store = publisher() + with self.assertRaisesRegex( + ValueError, + "^{} must be an iterable of Inventory".format( + expected_name, + ), + ) as raised: + adapter.ingest_inventory(1, sites, devices) + self.assertNotIn("SECRET", str(raised.exception)) + self.assertEqual(state_store.writes, 0) + + def test_iterator_type_errors_do_not_retain_provider_values(self): + """Malformed iterable faults cannot survive in public error objects.""" + raw_value = "provider-secret-value" + + class FaultingIterable: + def __iter__(self): + raise TypeError(raw_value) + + cases = ( + lambda adapter: adapter.ingest_inventory( + 1, + FaultingIterable(), + (), + ), + lambda adapter: adapter.ingest_inventory( + 1, + (site(),), + FaultingIterable(), + ), + ) + for action in cases: + with self.subTest(action=action): + adapter, state_store = publisher() + assert_value_free_validation_error( + self, + lambda: action(adapter), + raw_value, + ) + self.assertEqual(state_store.writes, 0) + + def test_non_type_iterator_errors_do_not_retain_provider_values(self): + """Acquisition and mid-stream faults are replaced outside handlers.""" + raw_value = "provider-secret-value" + + class FaultOnIter: + def __iter__(self): + raise RuntimeError(raw_value) + + class FaultOnNext: + def __init__(self, first): + self._first = first + + def __iter__(self): + return self + + def __next__(self): + if self._first is not None: + first = self._first + self._first = None + return first + raise ValueError(raw_value) + + cases = ( + lambda adapter: adapter.ingest_inventory( + 1, + FaultOnIter(), + (), + ), + lambda adapter: adapter.ingest_inventory( + 1, + (site(),), + FaultOnIter(), + ), + lambda adapter: adapter.ingest_inventory( + 1, + FaultOnNext(site()), + (), + ), + lambda adapter: adapter.ingest_inventory( + 1, + (site(),), + FaultOnNext(device()), + ), + ) + for action in cases: + with self.subTest(action=action): + adapter, state_store = publisher() + assert_value_free_validation_error( + self, + lambda: action(adapter), + raw_value, + ) + self.assertEqual(state_store.writes, 0) + + def test_liveness_and_feedback_invalidate_with_fresh_reads(self): + """Health advances while generated feedback persists nothing.""" + adapter, state_store = publisher() + adapter.ingest_inventory( + 3, + (site(),), + (device(),), + health=True, + ) + registry = FragmentAdapterRegistry(enabled=True) + registry.discover((adapter,)) + compiler = registry.create_compiler( + InMemoryCompiledLatticeStateStore(), + ) + first = compiler.drain() + self.assertEqual(first.status, CompileStatus.FRESH) + + self.assertTrue(adapter.set_liveness(None)) + degraded = compiler.drain() + self.assertEqual(degraded.status, CompileStatus.DEGRADED) + self.assertEqual(adapter.discovery_generation, 3) + self.assertEqual(adapter.generation, 2) + + before = adapter.read_state() + writes = state_store.writes + self.assertFalse( + adapter.ingest_inventory( + 4, + (site(),), + (device(model="Feedback"),), + feedback_token=degraded.publication.feedback_token, + ) + ) + self.assertEqual(adapter.read_state(), before) + self.assertEqual(state_store.writes, writes) + self.assertFalse(compiler.drain().pending) + + def test_removal_is_durable_invalidating_and_irreversible(self): + """Whole-provider removal survives restart and blocks resurrection.""" + adapter, state_store = publisher() + adapter.ingest_inventory( + 2, + (site(),), + (device(),), + health=True, + ) + invalidations = [] + adapter.subscribe_invalidation( + lambda provider_id, generation, reason, feedback_token: ( + invalidations.append( + (provider_id, generation, reason), + ) + ) + ) + + self.assertTrue(adapter.remove()) + self.assertFalse(adapter.remove()) + removed = adapter.read_state() + self.assertTrue(removed.removed) + self.assertEqual( + removed.snapshot.health, + ProviderHealth.OFFLINE, + ) + with self.assertRaises(FragmentAdapterRemoved): + adapter.read_snapshot() + with self.assertRaisesRegex( + FragmentAdapterRemoved, + "cannot re-enrol", + ): + adapter.ingest_inventory(99, (), ()) + self.assertEqual( + invalidations[-1], + ( + "inventory-provider", + 2, + "complete inventory integration removed", + ), + ) + + restarted = CompleteInventoryFragmentPublisher( + "inventory-provider", + "Inventory Provider", + state_store, + enabled=True, + ) + self.assertTrue(restarted.read_state().removed) + with self.assertRaises(FragmentAdapterRemoved): + restarted.read_snapshot() + with self.assertRaises(FragmentAdapterRemoved): + restarted.ingest_inventory(100, (), ()) + + +if __name__ == "__main__": + unittest.main() diff --git a/apps/predbat/tests/test_lattice_deye_inventory_fragment.py b/apps/predbat/tests/test_lattice_deye_inventory_fragment.py new file mode 100644 index 000000000..389ba5427 --- /dev/null +++ b/apps/predbat/tests/test_lattice_deye_inventory_fragment.py @@ -0,0 +1,766 @@ +"""Tests for the pure Deye complete-inventory normalizer.""" + +# cspell:ignore autoconfig + +import dataclasses +import os +import subprocess +import sys +import unittest + +sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..")) + +from lattice_autoconfig import ( # noqa: E402 + AliasRole, + ProviderHealth, + compile_auto_config, +) +from lattice_complete_inventory_fragment import ( # noqa: E402 + MAX_COMPLETE_INVENTORY_DEVICES, + MAX_COMPLETE_INVENTORY_SITES, +) +from lattice_deye_inventory_fragment import ( # noqa: E402 + DeyeInventoryFragmentPublisher, + DeyeInverterObservation, + DeyeStationObservation, +) +from lattice_fragment_adapters import ( # noqa: E402 + FragmentAdapterConflict, + FragmentAdapterReadError, + InMemoryFragmentAdapterStateStore, +) +from lattice_ge_cloud_fragment import ( # noqa: E402 + GECloudDeviceSnapshot, + GECloudFragmentPublisher, +) + + +def assert_value_free_validation_error(test_case, action, raw_value): + """Assert one public validation error retains no provider source text.""" + try: + action() + except ValueError as error: + test_case.assertIs(type(error), ValueError) + test_case.assertNotIn(raw_value, str(error)) + test_case.assertNotIn(raw_value, repr(error)) + test_case.assertNotIn(raw_value, repr(error.args)) + test_case.assertNotIn(raw_value, error.args) + test_case.assertFalse(hasattr(error, "object")) + test_case.assertIsNone(error.__cause__) + test_case.assertIsNone(error.__context__) + else: + test_case.fail("expected a value-free ValueError") + + +class HostileText(str): + """String subclass whose scalar hooks expose a fake provider secret.""" + + secret = "provider-secret-value" + + def _fail(self, *_args, **_kwargs): + """Fail if validation invokes any subclass-controlled hook.""" + raise RuntimeError(self.secret) + + strip = _fail + lower = _fail + upper = _fail + __format__ = _fail + __iter__ = _fail + __len__ = _fail + __hash__ = _fail + __eq__ = _fail + __lt__ = _fail + + +class HostileInt(int): + """Integer subclass whose scalar hooks expose a fake provider secret.""" + + secret = HostileText.secret + + def _fail(self, *_args, **_kwargs): + """Fail if validation invokes any subclass-controlled hook.""" + raise RuntimeError(self.secret) + + __str__ = _fail + __format__ = _fail + __hash__ = _fail + __eq__ = _fail + __lt__ = _fail + + +class DeyeStationObservationSubclass(DeyeStationObservation): + """Observation subclass that must not cross the public boundary.""" + + +class DeyeInverterObservationSubclass(DeyeInverterObservation): + """Observation subclass that must not cross the public boundary.""" + + +def station(station_id="STATION-A", online=True): + """Build one explicit Deye station observation.""" + return DeyeStationObservation( + station_id=station_id, + online=online, + ) + + +def inverter( + device_id="DEYE-DEVICE-A", + station_id="STATION-A", + api_verified_serial="DEYE-SERIAL-A", + model="SUN-6K", + online=True, +): + """Build one explicit Deye inverter observation.""" + return DeyeInverterObservation( + device_id=device_id, + station_id=station_id, + api_verified_serial=api_verified_serial, + model=model, + online=online, + ) + + +def publisher(enabled=True, state_store=None): + """Build one Deye inventory publisher and durable store.""" + state_store = state_store or InMemoryFragmentAdapterStateStore() + return ( + DeyeInventoryFragmentPublisher( + "deye-cloud", + state_store, + enabled=enabled, + ), + state_store, + ) + + +class TestDeyeInventoryFragmentPublisher(unittest.TestCase): + """Deye discovery remains pure, complete, and authority-free.""" + + def test_default_off_does_not_import_live_deye_or_write(self): + """Construction cannot touch live auth, discovery, or control code.""" + module_dir = os.path.abspath( + os.path.join(os.path.dirname(__file__), ".."), + ) + isolated = subprocess.run( + ( + sys.executable, + "-c", + "import sys; import lattice_deye_inventory_fragment; " "assert 'deye' not in sys.modules", + ), + cwd=module_dir, + check=False, + capture_output=True, + text=True, + ) + self.assertEqual(isolated.returncode, 0, isolated.stderr) + + adapter, state_store = publisher(enabled=False) + self.assertIsNone(adapter.lattice_fragment_adapter()) + self.assertFalse( + adapter.ingest_snapshot( + -1, + (object(),), + (object(),), + ) + ) + self.assertEqual(state_store.writes, 0) + with self.assertRaises(FragmentAdapterReadError): + adapter.read_state() + + def test_scalar_and_observation_subclasses_fail_value_free(self): + """Deye discovery cannot invoke provider-controlled scalar hooks.""" + constructor_cases = ( + lambda: station(HostileInt(1)), + lambda: inverter(device_id=HostileText("DEVICE")), + lambda: inverter(station_id=HostileText("STATION")), + lambda: inverter( + api_verified_serial=HostileText("SERIAL"), + ), + lambda: inverter(model=HostileText("Model")), + ) + for action in constructor_cases: + with self.subTest(action=action): + assert_value_free_validation_error( + self, + action, + HostileText.secret, + ) + + ingestion_cases = ( + lambda adapter: adapter.ingest_snapshot( + HostileInt(1), + (), + (), + ), + lambda adapter: adapter.ingest_snapshot( + 1, + ( + DeyeStationObservationSubclass( + "STATION-A", + ), + ), + (), + ), + lambda adapter: adapter.ingest_snapshot( + 1, + (station(),), + ( + DeyeInverterObservationSubclass( + "DEVICE-A", + "STATION-A", + ), + ), + ), + ) + for action in ingestion_cases: + with self.subTest(action=action): + adapter, state_store = publisher() + assert_value_free_validation_error( + self, + lambda: action(adapter), + HostileText.secret, + ) + self.assertEqual(state_store.writes, 0) + + def test_mutated_observations_are_revalidated_before_publication(self): + """Post-construction Deye mutation cannot execute source hooks.""" + mutated_station = station() + object.__setattr__( + mutated_station, + "station_id", + HostileText("STATION-A"), + ) + mutated_inverter = inverter() + object.__setattr__( + mutated_inverter, + "model", + HostileText("Model"), + ) + cases = ( + ( + (mutated_station,), + (), + ), + ( + (station(),), + (mutated_inverter,), + ), + ) + for stations, inverters in cases: + with self.subTest( + stations=stations, + inverters=inverters, + ): + adapter, state_store = publisher() + assert_value_free_validation_error( + self, + lambda: adapter.ingest_snapshot( + 1, + stations, + inverters, + ), + HostileText.secret, + ) + self.assertEqual(state_store.writes, 0) + + def test_post_init_normalization_cannot_heal_deye_observations(self): + """Whitespace and case mutations fail before durable publication.""" + + def station_id(): + candidate = station() + object.__setattr__( + candidate, + "station_id", + " STATION-A ", + ) + return (candidate,), (), " STATION-A " + + def inverter_device_id(): + candidate = inverter() + object.__setattr__( + candidate, + "device_id", + " DEYE-DEVICE-A ", + ) + return (station(),), (candidate,), " DEYE-DEVICE-A " + + def inverter_station_id(): + candidate = inverter() + object.__setattr__( + candidate, + "station_id", + " STATION-A ", + ) + return (station(),), (candidate,), " STATION-A " + + def inverter_serial(): + candidate = inverter() + object.__setattr__( + candidate, + "api_verified_serial", + "deye-serial-a", + ) + return (station(),), (candidate,), "deye-serial-a" + + def inverter_model(): + candidate = inverter() + object.__setattr__( + candidate, + "model", + " SUN-6K ", + ) + return (station(),), (candidate,), " SUN-6K " + + for build_case in ( + station_id, + inverter_device_id, + inverter_station_id, + inverter_serial, + inverter_model, + ): + with self.subTest(build_case=build_case): + stations, inverters, raw_value = build_case() + adapter, state_store = publisher() + assert_value_free_validation_error( + self, + lambda: adapter.ingest_snapshot( + 1, + stations, + inverters, + ), + raw_value, + ) + self.assertEqual(state_store.writes, 0) + + def test_multi_station_inventory_is_normalized_and_complete(self): + """Deye ordering is deterministic and a newer view removes absence.""" + adapter, state_store = publisher() + stations = ( + station("STATION-B", online=None), + station("STATION-A"), + ) + inverters = ( + inverter( + "DEVICE-B", + "STATION-B", + "SERIAL-B", + online=None, + ), + inverter( + "DEVICE-A", + "STATION-A", + "SERIAL-A", + ), + ) + + self.assertTrue( + adapter.ingest_snapshot( + 1, + reversed(stations), + reversed(inverters), + health=True, + ) + ) + first = adapter.read_snapshot() + self.assertEqual( + tuple(node["id"] for node in first.topology_fragment["nodes"]), + ( + "inventory:deye-cloud:site:STATION-A", + "inventory:deye-cloud:site:STATION-B", + "inventory:deye-cloud:device:DEVICE-A", + "inventory:deye-cloud:device:DEVICE-B", + ), + ) + + self.assertTrue( + adapter.ingest_snapshot( + 2, + (station("STATION-A"),), + ( + inverter( + "DEVICE-A", + "STATION-A", + "SERIAL-A", + ), + ), + ) + ) + second = adapter.read_snapshot() + serialized = repr(second) + self.assertNotIn("STATION-B", serialized) + self.assertNotIn("DEVICE-B", serialized) + self.assertNotIn("SERIAL-B", serialized) + self.assertEqual(adapter.generation, 2) + self.assertEqual(adapter.discovery_generation, 2) + self.assertEqual(state_store.writes, 2) + + def test_only_explicit_api_verified_serial_is_strong_identity(self): + """Provider device/station IDs and model never correlate globally.""" + adapter, _state_store = publisher() + adapter.ingest_snapshot( + 1, + (station(),), + ( + inverter( + device_id="LOOKS-LIKE-A-SERIAL", + api_verified_serial=None, + model="MODEL-IDENTIFIER", + ), + inverter( + device_id="DEVICE-VERIFIED", + api_verified_serial="verified-serial", + ), + ), + health=True, + ) + snapshot = adapter.read_snapshot() + + self.assertEqual( + tuple((alias.kind, alias.value, alias.node_id) for alias in snapshot.identity_aliases), + ( + ( + "serial", + "VERIFIED-SERIAL", + "inventory:deye-cloud:device:DEVICE-VERIFIED", + ), + ), + ) + self.assertTrue(all(alias.roles == frozenset((AliasRole.REFERENCE,)) for alias in snapshot.aliases)) + self.assertEqual(snapshot.role_assignments, ()) + self.assertEqual(snapshot.config_projections, ()) + self.assertTrue(all(node["capabilities"] == () for node in snapshot.topology_fragment["nodes"])) + self.assertEqual( + snapshot.topology_fragment["producer"]["authority"], + 0, + ) + + def test_verified_serial_composes_without_materialization(self): + """A verified serial may join another provider's hardware observation.""" + deye, _deye_store = publisher() + deye.ingest_snapshot( + 1, + (station(),), + ( + inverter( + api_verified_serial="shared-serial", + ), + ), + health=True, + ) + ge = GECloudFragmentPublisher( + "ge-cloud", + InMemoryFragmentAdapterStateStore(), + enabled=True, + ) + ge.ingest_discovery( + 1, + ( + GECloudDeviceSnapshot( + serial="SHARED-SERIAL", + kind="battery-inverter", + online=True, + ), + ), + health=True, + ) + + plan = compile_auto_config( + (deye.read_snapshot(), ge.read_snapshot()), + ) + inverter_nodes = tuple(node for node in plan.topology["nodes"] if node["kind"] == "inverter") + + self.assertEqual(len(inverter_nodes), 1) + self.assertEqual( + inverter_nodes[0]["id"], + "identity:serial:SHARED-SERIAL", + ) + self.assertEqual(plan.role_assignments, ()) + self.assertEqual(plan.config_arguments, ()) + self.assertFalse(plan.materialization_readiness.ready) + + def test_input_shape_has_no_secret_endpoint_feature_or_control_fields(self): + """The adapter cannot accept auth, endpoint, TOU, or feature claims.""" + station_fields = {field.name for field in dataclasses.fields(DeyeStationObservation)} + inverter_fields = {field.name for field in dataclasses.fields(DeyeInverterObservation)} + forbidden = { + "access_token", + "app_secret", + "auth", + "capabilities", + "endpoint", + "features", + "password", + "schedule", + "tou", + } + + self.assertTrue(station_fields.isdisjoint(forbidden)) + self.assertTrue(inverter_fields.isdisjoint(forbidden)) + self.assertEqual( + inverter_fields, + { + "api_verified_serial", + "device_id", + "model", + "online", + "station_id", + }, + ) + + def test_membership_conflict_fails_before_publication(self): + """An inverter cannot be attached to an undiscovered station.""" + adapter, state_store = publisher() + + with self.assertRaisesRegex( + ValueError, + "^complete inventory device references an unknown site_id$", + ): + adapter.ingest_snapshot( + 1, + (station(),), + (inverter(station_id="OTHER"),), + ) + with self.assertRaises(FragmentAdapterConflict): + adapter.ingest_snapshot( + 1, + (station(),), + ( + inverter( + "DEVICE-A", + api_verified_serial="SHARED", + ), + inverter( + "DEVICE-B", + api_verified_serial="shared", + ), + ), + ) + self.assertEqual(state_store.writes, 0) + + def test_maximum_station_and_inverter_cardinality_is_accepted(self): + """Deye permits the same bounded complete inventory limits.""" + station_adapter, station_store = publisher() + stations = (station("STATION-{}".format(index)) for index in range(MAX_COMPLETE_INVENTORY_SITES)) + self.assertTrue(station_adapter.ingest_snapshot(1, stations, ())) + self.assertEqual(station_store.writes, 1) + + inverter_adapter, inverter_store = publisher() + inverters = ( + inverter( + "DEVICE-{}".format(index), + api_verified_serial=None, + ) + for index in range(MAX_COMPLETE_INVENTORY_DEVICES) + ) + self.assertTrue( + inverter_adapter.ingest_snapshot( + 1, + (station(),), + inverters, + ) + ) + self.assertEqual(inverter_store.writes, 1) + + def test_maximum_plus_one_is_rejected_after_bounded_consumption(self): + """Finite oversized Deye inputs consume only max+1 and never write.""" + cases = ( + ( + "stations", + MAX_COMPLETE_INVENTORY_SITES, + lambda counter: ( + ( + counter.append(None) or station("STATION-{}".format(index)) + for index in range( + MAX_COMPLETE_INVENTORY_SITES + 5, + ) + ), + (), + ), + ), + ( + "inverters", + MAX_COMPLETE_INVENTORY_DEVICES, + lambda counter: ( + (station(),), + ( + counter.append(None) + or inverter( + "DEVICE-{}".format(index), + api_verified_serial=None, + ) + for index in range( + MAX_COMPLETE_INVENTORY_DEVICES + 5, + ) + ), + ), + ), + ) + for label, maximum, inventory in cases: + with self.subTest(label=label): + adapter, state_store = publisher() + counter = [] + stations, inverters = inventory(counter) + with self.assertRaisesRegex( + ValueError, + "^{} must contain at most {} observations$".format( + label, + maximum, + ), + ): + adapter.ingest_snapshot( + 1, + stations, + inverters, + ) + self.assertEqual(len(counter), maximum + 1) + self.assertEqual(state_store.writes, 0) + + def test_infinite_inputs_terminate_after_maximum_plus_one(self): + """Infinite Deye discovery streams are bounded before publication.""" + + def infinite_observations(observation, counter): + while True: + counter.append(None) + yield observation + + cases = ( + ( + MAX_COMPLETE_INVENTORY_SITES, + lambda counter: ( + infinite_observations(station(), counter), + (), + ), + ), + ( + MAX_COMPLETE_INVENTORY_DEVICES, + lambda counter: ( + (station(),), + infinite_observations( + inverter(api_verified_serial=None), + counter, + ), + ), + ), + ) + for maximum, inventory in cases: + with self.subTest(maximum=maximum): + adapter, state_store = publisher() + counter = [] + stations, inverters = inventory(counter) + with self.assertRaises(ValueError): + adapter.ingest_snapshot( + 1, + stations, + inverters, + ) + self.assertEqual(len(counter), maximum + 1) + self.assertEqual(state_store.writes, 0) + + def test_iterator_type_errors_do_not_retain_provider_values(self): + """Deye source faults cannot survive in public error objects.""" + raw_value = "provider-secret-value" + + class FaultingIterable: + def __iter__(self): + raise TypeError(raw_value) + + cases = ( + lambda adapter: adapter.ingest_snapshot( + 1, + FaultingIterable(), + (), + ), + lambda adapter: adapter.ingest_snapshot( + 1, + (station(),), + FaultingIterable(), + ), + ) + for action in cases: + with self.subTest(action=action): + adapter, state_store = publisher() + assert_value_free_validation_error( + self, + lambda: action(adapter), + raw_value, + ) + self.assertEqual(state_store.writes, 0) + + def test_non_type_iterator_errors_do_not_retain_provider_values(self): + """Deye replaces acquisition and mid-stream source faults.""" + raw_value = "provider-secret-value" + + class FaultOnIter: + def __iter__(self): + raise RuntimeError(raw_value) + + class FaultOnNext: + def __init__(self, first): + self._first = first + + def __iter__(self): + return self + + def __next__(self): + if self._first is not None: + first = self._first + self._first = None + return first + raise ValueError(raw_value) + + cases = ( + lambda adapter: adapter.ingest_snapshot( + 1, + FaultOnIter(), + (), + ), + lambda adapter: adapter.ingest_snapshot( + 1, + (station(),), + FaultOnIter(), + ), + lambda adapter: adapter.ingest_snapshot( + 1, + FaultOnNext(station()), + (), + ), + lambda adapter: adapter.ingest_snapshot( + 1, + (station(),), + FaultOnNext(inverter()), + ), + ) + for action in cases: + with self.subTest(action=action): + adapter, state_store = publisher() + assert_value_free_validation_error( + self, + lambda: action(adapter), + raw_value, + ) + self.assertEqual(state_store.writes, 0) + + def test_liveness_and_whole_provider_tombstone_delegate_safely(self): + """Deye wrapper preserves generic liveness/removal guarantees.""" + adapter, state_store = publisher() + adapter.ingest_snapshot( + 4, + (station(),), + (inverter(),), + health=True, + ) + + self.assertTrue(adapter.set_liveness(False)) + self.assertEqual( + adapter.read_snapshot().health, + ProviderHealth.OFFLINE, + ) + self.assertEqual(adapter.discovery_generation, 4) + self.assertTrue(adapter.remove()) + self.assertTrue(adapter.read_state().removed) + self.assertEqual(state_store.writes, 3) + + +if __name__ == "__main__": + unittest.main() diff --git a/apps/predbat/tests/test_lattice_fragment_adapters.py b/apps/predbat/tests/test_lattice_fragment_adapters.py index adf075213..f130f6f07 100644 --- a/apps/predbat/tests/test_lattice_fragment_adapters.py +++ b/apps/predbat/tests/test_lattice_fragment_adapters.py @@ -1,16 +1,28 @@ """Tests for generic durable Lattice fragment adapter discovery.""" -# cspell:ignore autoconfig +# cspell:ignore autoconfig noncanonical import os import sys import unittest +from itertools import repeat +from types import MappingProxyType sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..")) from lattice_autoconfig import ( # noqa: E402 + AliasRole, CompileStatus, + ProjectionCardinality, + ProjectionRouting, + ProjectionValueKind, + ProviderAlias, + ProviderConfigProjection, ProviderHealth, + ProviderIdentityAlias, + ProviderProjectionValue, + ProviderRoleAssignment, + ProviderSnapshot, ) from lattice_compiled_publication import ( # noqa: E402 InMemoryCompiledLatticeStateStore, @@ -23,10 +35,190 @@ FragmentAdapterRemoved, FragmentAdapterState, InMemoryFragmentAdapterStateStore, + _semantic_fingerprint, ) from tests.test_lattice_autoconfig import snapshot # noqa: E402 +def assert_value_free_error( + test_case, + expected_type, + action, + raw_value, +): + """Assert one boundary error retains no source-controlled value.""" + try: + action() + except expected_type as error: + test_case.assertIs(type(error), expected_type) + test_case.assertNotIn(raw_value, str(error)) + test_case.assertNotIn(raw_value, repr(error)) + test_case.assertNotIn(raw_value, repr(error.args)) + test_case.assertNotIn(raw_value, error.args) + test_case.assertIsNone(error.__cause__) + test_case.assertIsNone(error.__context__) + else: + test_case.fail( + "expected {}".format(expected_type.__name__), + ) + + +class HostileText(str): + """String subclass whose scalar hooks expose a fake provider secret.""" + + secret = "provider-secret-value" + + def _fail(self, *_args, **_kwargs): + """Fail if validation invokes any subclass-controlled hook.""" + raise RuntimeError(self.secret) + + strip = _fail + __format__ = _fail + __iter__ = _fail + __len__ = _fail + __hash__ = _fail + __eq__ = _fail + __lt__ = _fail + + +class HostileInt(int): + """Integer subclass whose scalar hooks expose a fake provider secret.""" + + secret = HostileText.secret + + def _fail(self, *_args, **_kwargs): + """Fail if validation invokes any subclass-controlled hook.""" + raise RuntimeError(self.secret) + + __str__ = _fail + __format__ = _fail + __hash__ = _fail + __eq__ = _fail + __lt__ = _fail + + +class HostileMapping(dict): + """Mapping whose traversal exposes a fake provider secret.""" + + def items(self): + """Fail if fingerprint validation traverses corrupt state.""" + raise RuntimeError(HostileText.secret) + + +class InfiniteItemsMapping(dict): + """Mapping whose item iterator never terminates.""" + + def items(self): + """Return an infinite exact key/value stream.""" + return repeat(("item", 1)) + + +class HostileProviderSnapshot(ProviderSnapshot): + """Snapshot subclass whose field access exposes a fake provider secret.""" + + def __getattribute__(self, _name): + """Fail if publication reads a subclass-controlled field.""" + raise RuntimeError(HostileText.secret) + + +class FragmentAdapterStateSubclass(FragmentAdapterState): + """Durable state subclass that must not dispatch validation.""" + + +class HostileFragmentAdapterState(FragmentAdapterState): + """State subclass whose field access exposes a fake provider secret.""" + + def __getattribute__(self, _name): + """Fail if registry validation reads a subclass-controlled field.""" + raise RuntimeError(HostileText.secret) + + +class HostileRegistryAdapter: + """Structural adapter whose attribute access exposes a fake secret.""" + + def __getattribute__(self, _name): + """Fail if registry validation leaks an adapter hook.""" + raise RuntimeError(HostileText.secret) + + +class FakeHealth: + """Enum-shaped offline health that must never become usable.""" + + value = "offline" + + +class FakeAlias: + """Alias-shaped value that is not an exact provider alias.""" + + name = "fake" + node_id = "GW-INV" + roles = frozenset((AliasRole.REFERENCE,)) + + +class FakeIdentityAlias: + """Identity-shaped value that is not an exact assertion.""" + + kind = "serial" + value = "SERIAL-A" + node_id = "GW-INV" + + +class FakeRoleAssignment: + """Role-shaped value that is not an exact assignment.""" + + role = AliasRole.PRIMARY + group = "battery" + index = 0 + node_id = "GW-INV" + + +class FakeProjectionValue: + """Projection-value-shaped object that is not an exact value.""" + + node_id = "GW-INV" + kind = ProjectionValueKind.NONE + value = None + capability = None + identity_kind = None + identity_value = None + access_path_id = None + + +class FakeConfigProjection: + """Projection-shaped object that is not an exact projection.""" + + argument = "fake_arg" + role = AliasRole.PRIMARY + group = "battery" + routing = ProjectionRouting.LEAF + cardinality = ProjectionCardinality.SCALAR + values = (FakeProjectionValue(),) + required = False + transforms = () + + +class MutableFragmentPublisher: + """Valid structural adapter whose state reader can fail after registration.""" + + def __init__(self, adapter): + """Delegate every surface while allowing a test reader replacement.""" + self.provider_id = adapter.provider_id + self._adapter = adapter + self.state_reader = adapter.read_state + + def read_state(self): + """Return the current test-selected state.""" + return self.state_reader() + + def read_snapshot(self): + """Delegate immutable snapshot reads.""" + return self._adapter.read_snapshot() + + def subscribe_invalidation(self, listener): + """Delegate compiler invalidation subscriptions.""" + return self._adapter.subscribe_invalidation(listener) + + class FragmentComponent: """Brand-neutral component exposing only the common discovery method.""" @@ -109,6 +301,36 @@ def advance( ) +def mutated_snapshot(field, value): + """Return one exact snapshot with a post-init field mutation.""" + candidate = snapshot( + "gateway", + generation=1, + node_id="GW-INV", + ) + object.__setattr__(candidate, field, value) + return candidate + + +def matched_corrupt_state(candidate, provider_id="gateway"): + """Bind a corrupt snapshot to its matching semantic fingerprint.""" + state = object.__new__(FragmentAdapterState) + object.__setattr__( + state, + "provider_id", + provider_id, + ) + object.__setattr__(state, "generation", 1) + object.__setattr__( + state, + "semantic_fingerprint", + _semantic_fingerprint(candidate), + ) + object.__setattr__(state, "snapshot", candidate) + object.__setattr__(state, "removed", False) + return state + + def compiled_registry(*adapters): """Discover generic components and create a durable compiler.""" registry = FragmentAdapterRegistry(enabled=True) @@ -133,13 +355,16 @@ def test_seed_and_fresh_reads_are_immutable_and_durable(self): self.assertIsInstance(state, FragmentAdapterState) self.assertEqual(state.generation, 1) self.assertEqual(len(state.semantic_fingerprint), 64) - self.assertIs(state.snapshot, initial) - self.assertIs(adapter.read_snapshot(), initial) + self.assertEqual(state.snapshot, initial) + self.assertIsNot(state.snapshot, initial) + self.assertEqual(adapter.read_snapshot(), initial) + self.assertIsNot(adapter.read_snapshot(), initial) self.assertEqual(state_store.writes, 1) restarted = DurableFragmentAdapter("gateway", state_store) self.assertEqual(restarted.read_state(), state) - self.assertIs(restarted.read_snapshot(), initial) + self.assertEqual(restarted.read_snapshot(), initial) + self.assertIsNot(restarted.read_snapshot(), initial) def test_restart_rejects_regression_and_generation_reuse(self): """Durable cursor restoration rejects regressions and mutations.""" @@ -165,11 +390,765 @@ def test_reader_store_failure_is_wrapped_and_fails_closed(self): """No synthetic or empty fragment is returned on durable read failure.""" store = RaisingLoadStore() - with self.assertRaisesRegex( + assert_value_free_error( + self, FragmentAdapterReadError, + lambda: DurableFragmentAdapter("gateway", store), "disk unavailable", - ): - DurableFragmentAdapter("gateway", store) + ) + + def test_scalar_subclasses_fail_before_write_or_invalidation(self): + """Adapter scalar validation invokes no subclass-controlled hook.""" + assert_value_free_error( + self, + ValueError, + lambda: DurableFragmentAdapter( + HostileText("gateway"), + InMemoryFragmentAdapterStateStore(), + ), + HostileText.secret, + ) + + adapter, state_store, initial = publisher("gateway") + invalidations = [] + adapter.subscribe_invalidation( + lambda *event: invalidations.append(event), + ) + writes = state_store.writes + cases = ( + lambda: adapter.publish( + initial, + HostileText("reason"), + ), + lambda: adapter.remove( + HostileInt(2), + "integration removed", + ), + ) + for action in cases: + with self.subTest(action=action): + assert_value_free_error( + self, + ValueError, + action, + HostileText.secret, + ) + self.assertEqual(state_store.writes, writes) + self.assertEqual(invalidations, []) + + def test_snapshot_subclasses_and_mutations_fail_before_side_effects(self): + """Publication validates exact snapshots without retaining source data.""" + + def mutated_snapshot(field, value): + candidate = snapshot( + "gateway", + generation=1, + node_id="GW-INV", + ) + object.__setattr__(candidate, field, value) + return candidate + + hostile_subclass = object.__new__( + HostileProviderSnapshot, + ) + cases = ( + ( + hostile_subclass, + False, + ), + ( + mutated_snapshot( + "provider_id", + HostileText("gateway"), + ), + False, + ), + ( + mutated_snapshot( + "generation", + HostileInt(1), + ), + False, + ), + ( + mutated_snapshot( + "topology_fragment", + HostileMapping(), + ), + False, + ), + ( + snapshot( + "gateway", + generation=1, + node_id="GW-INV", + ), + HostileText("false"), + ), + ) + for candidate, removed in cases: + with self.subTest( + candidate_type=type(candidate), + removed_type=type(removed), + ): + state_store = InMemoryFragmentAdapterStateStore() + adapter = DurableFragmentAdapter( + "gateway", + state_store, + ) + invalidations = [] + adapter.subscribe_invalidation( + lambda *event: invalidations.append(event), + ) + assert_value_free_error( + self, + ValueError, + lambda: adapter.publish( + candidate, + "hostile candidate", + removed=removed, + ), + HostileText.secret, + ) + self.assertEqual(state_store.writes, 0) + self.assertEqual(invalidations, []) + + def test_full_snapshot_reconstruction_rejects_invalid_types_and_bounds(self): + """Fingerprint-shaped mutations never become durable snapshots.""" + + def mutated_alias(): + alias = ProviderAlias( + "gateway", + "GW-INV", + ) + object.__setattr__( + alias, + "name", + HostileText("gateway"), + ) + return mutated_snapshot("aliases", (alias,)) + + def mutated_identity(): + alias = ProviderIdentityAlias( + "serial", + "SERIAL-A", + "GW-INV", + ) + object.__setattr__( + alias, + "value", + HostileText("SERIAL-A"), + ) + return mutated_snapshot( + "identity_aliases", + (alias,), + ) + + def mutated_role(): + assignment = ProviderRoleAssignment( + AliasRole.PRIMARY, + "battery", + 0, + "GW-INV", + ) + object.__setattr__( + assignment, + "index", + HostileInt(0), + ) + return mutated_snapshot( + "role_assignments", + (assignment,), + ) + + def projection(value=None): + if value is None: + value = ProviderProjectionValue( + node_id="GW-INV", + kind=ProjectionValueKind.NONE, + ) + return ProviderConfigProjection( + argument="fake_arg", + role=AliasRole.PRIMARY, + group="battery", + routing=ProjectionRouting.LEAF, + cardinality=ProjectionCardinality.SCALAR, + values=(value,), + required=False, + ) + + def mutated_projection(): + candidate = projection() + object.__setattr__( + candidate, + "argument", + HostileText("fake_arg"), + ) + return mutated_snapshot( + "config_projections", + (candidate,), + ) + + def fake_projection_value(): + candidate = projection() + object.__setattr__( + candidate, + "values", + (FakeProjectionValue(),), + ) + return mutated_snapshot( + "config_projections", + (candidate,), + ) + + def mutated_projection_value(): + value = ProviderProjectionValue( + node_id="GW-INV", + kind=ProjectionValueKind.NONE, + ) + object.__setattr__( + value, + "node_id", + HostileText("GW-INV"), + ) + return mutated_snapshot( + "config_projections", + (projection(value),), + ) + + def excessive_depth(): + value = None + for _depth in range(34): + value = (value,) + return mutated_snapshot( + "topology_fragment", + MappingProxyType( + { + "deep": value, + } + ), + ) + + def excessive_total_items(): + wide = (0,) * 65536 + return mutated_snapshot( + "topology_fragment", + MappingProxyType({"wide-{}".format(index): wide for index in range(16)}), + ) + + cases = ( + lambda: mutated_snapshot( + "health", + FakeHealth(), + ), + lambda: mutated_snapshot( + "topology_fragment", + [], + ), + lambda: mutated_snapshot( + "topology_fragment", + MappingProxyType( + { + "nested": MappingProxyType( + HostileMapping(), + ), + } + ), + ), + lambda: mutated_snapshot( + "topology_fragment", + MappingProxyType( + InfiniteItemsMapping(), + ), + ), + lambda: mutated_snapshot( + "topology_fragment", + MappingProxyType( + { + "huge": "x" * (16384 + 1), + } + ), + ), + lambda: mutated_snapshot( + "topology_fragment", + MappingProxyType( + { + "huge": 1 << 63, + } + ), + ), + lambda: mutated_snapshot( + "topology_fragment", + MappingProxyType( + { + "notFinite": float("inf"), + } + ), + ), + excessive_depth, + excessive_total_items, + lambda: mutated_snapshot( + "aliases", + (FakeAlias(),), + ), + mutated_alias, + lambda: mutated_snapshot( + "identity_aliases", + (FakeIdentityAlias(),), + ), + mutated_identity, + lambda: mutated_snapshot( + "role_assignments", + (FakeRoleAssignment(),), + ), + mutated_role, + lambda: mutated_snapshot( + "config_projections", + (FakeConfigProjection(),), + ), + mutated_projection, + fake_projection_value, + mutated_projection_value, + ) + for build_candidate in cases: + with self.subTest( + build_candidate=build_candidate, + ): + candidate = build_candidate() + state_store = InMemoryFragmentAdapterStateStore() + adapter = DurableFragmentAdapter( + "gateway", + state_store, + ) + invalidations = [] + adapter.subscribe_invalidation( + lambda *event: invalidations.append(event), + ) + assert_value_free_error( + self, + ValueError, + lambda: adapter.publish( + candidate, + "invalid snapshot", + ), + HostileText.secret, + ) + self.assertEqual(state_store.writes, 0) + self.assertEqual(invalidations, []) + + def test_matched_invalid_snapshot_fails_restart_registry_and_compiler(self): + """A matching fingerprint cannot legitimize an invalid snapshot.""" + corrupt_snapshot = mutated_snapshot( + "health", + FakeHealth(), + ) + corrupt_state = matched_corrupt_state( + corrupt_snapshot, + ) + + restart_store = InMemoryFragmentAdapterStateStore( + corrupt_state, + ) + assert_value_free_error( + self, + FragmentAdapterReadError, + lambda: DurableFragmentAdapter( + "gateway", + restart_store, + ), + HostileText.secret, + ) + self.assertEqual(restart_store.writes, 0) + + valid_adapter, valid_store, _initial = publisher( + "gateway", + ) + structural = MutableFragmentPublisher( + valid_adapter, + ) + structural.state_reader = lambda: corrupt_state + registry = FragmentAdapterRegistry(enabled=True) + assert_value_free_error( + self, + ValueError, + lambda: registry.register(structural), + HostileText.secret, + ) + self.assertEqual(registry.provider_ids, ()) + self.assertEqual(valid_store.writes, 1) + + live_registry = FragmentAdapterRegistry(enabled=True) + self.assertTrue( + live_registry.register(valid_adapter), + ) + compiled_store = InMemoryCompiledLatticeStateStore() + compiler = live_registry.create_compiler( + compiled_store, + ) + valid_store._state = corrupt_state + self.assertTrue( + compiler.invalidate( + "gateway", + 1, + "matched invalid state", + ) + ) + run = compiler.drain() + self.assertIsNot(run.status, CompileStatus.FRESH) + self.assertIsNone(run.plan) + self.assertTrue(any(issue.code == "provider_read_failed" for issue in run.issues)) + self.assertEqual(valid_store.writes, 1) + + def test_noncanonical_snapshot_fields_fail_every_boundary(self): + """Normalizing reconstruction cannot heal mutated durable input.""" + + def projection(value=None, transforms=()): + if value is None: + value = ProviderProjectionValue( + node_id="GW-INV", + kind=ProjectionValueKind.NONE, + ) + return ProviderConfigProjection( + argument="fake_arg", + role=AliasRole.PRIMARY, + group="battery", + routing=ProjectionRouting.LEAF, + cardinality=ProjectionCardinality.SCALAR, + values=(value,), + required=False, + transforms=transforms, + ) + + def snapshot_provider_id(): + return mutated_snapshot( + "provider_id", + " gateway ", + ) + + def alias_name(): + alias = ProviderAlias( + "gateway", + "GW-INV", + ) + object.__setattr__(alias, "name", " gateway ") + return mutated_snapshot("aliases", (alias,)) + + def identity_kind(): + alias = ProviderIdentityAlias( + "serial", + "SERIAL-A", + "GW-INV", + ) + object.__setattr__(alias, "kind", "SERIAL") + return mutated_snapshot( + "identity_aliases", + (alias,), + ) + + def role_group(): + assignment = ProviderRoleAssignment( + AliasRole.PRIMARY, + "battery", + 0, + "GW-INV", + ) + object.__setattr__( + assignment, + "group", + " battery ", + ) + return mutated_snapshot( + "role_assignments", + (assignment,), + ) + + def projection_argument(): + candidate = projection() + object.__setattr__( + candidate, + "argument", + " fake_arg ", + ) + return mutated_snapshot( + "config_projections", + (candidate,), + ) + + def projection_value_identity_kind(): + value = ProviderProjectionValue( + node_id="GW-INV", + kind=ProjectionValueKind.NONE, + identity_kind="serial", + identity_value="SERIAL-A", + ) + object.__setattr__( + value, + "identity_kind", + "SERIAL", + ) + return mutated_snapshot( + "config_projections", + (projection(value),), + ) + + def projection_transform(): + candidate = projection(transforms=("scale",)) + object.__setattr__( + candidate, + "transforms", + (" scale ",), + ) + return mutated_snapshot( + "config_projections", + (candidate,), + ) + + cases = ( + (snapshot_provider_id, " gateway "), + (alias_name, " gateway "), + (identity_kind, "SERIAL"), + (role_group, " battery "), + (projection_argument, " fake_arg "), + (projection_value_identity_kind, "SERIAL"), + (projection_transform, " scale "), + ) + for build_candidate, raw_value in cases: + with self.subTest( + build_candidate=build_candidate, + ): + candidate = build_candidate() + empty_store = InMemoryFragmentAdapterStateStore() + adapter = DurableFragmentAdapter( + "gateway", + empty_store, + ) + invalidations = [] + adapter.subscribe_invalidation( + lambda *event: invalidations.append(event), + ) + assert_value_free_error( + self, + ValueError, + lambda: adapter.publish( + candidate, + "noncanonical snapshot", + ), + raw_value, + ) + self.assertEqual(empty_store.writes, 0) + self.assertEqual(invalidations, []) + + corrupt_state = matched_corrupt_state(candidate) + restart_store = InMemoryFragmentAdapterStateStore( + corrupt_state, + ) + assert_value_free_error( + self, + FragmentAdapterReadError, + lambda: DurableFragmentAdapter( + "gateway", + restart_store, + ), + raw_value, + ) + self.assertEqual(restart_store.writes, 0) + + valid_adapter, valid_store, _initial = publisher( + "gateway", + ) + structural = MutableFragmentPublisher( + valid_adapter, + ) + structural.state_reader = lambda: corrupt_state + registry = FragmentAdapterRegistry(enabled=True) + assert_value_free_error( + self, + ValueError, + lambda: registry.register(structural), + raw_value, + ) + self.assertEqual(registry.provider_ids, ()) + self.assertEqual(valid_store.writes, 1) + + live_registry = FragmentAdapterRegistry( + enabled=True, + ) + self.assertTrue( + live_registry.register(valid_adapter), + ) + compiler = live_registry.create_compiler( + InMemoryCompiledLatticeStateStore(), + ) + valid_store._state = corrupt_state + self.assertTrue( + compiler.invalidate( + "gateway", + 1, + "noncanonical durable state", + ) + ) + run = compiler.drain() + self.assertIsNot( + run.status, + CompileStatus.FRESH, + ) + self.assertIsNone(run.plan) + self.assertTrue(any(issue.code == "provider_read_failed" for issue in run.issues)) + for issue in run.issues: + self.assertNotIn( + raw_value, + issue.detail, + ) + self.assertEqual(valid_store.writes, 1) + + def test_noncanonical_durable_provider_id_fails_every_read_boundary(self): + """The durable cursor provider ID must already be canonical.""" + candidate = mutated_snapshot( + "provider_id", + "gateway", + ) + raw_value = " gateway " + corrupt_state = matched_corrupt_state( + candidate, + provider_id=raw_value, + ) + + restart_store = InMemoryFragmentAdapterStateStore( + corrupt_state, + ) + assert_value_free_error( + self, + FragmentAdapterReadError, + lambda: DurableFragmentAdapter( + "gateway", + restart_store, + ), + raw_value, + ) + self.assertEqual(restart_store.writes, 0) + + valid_adapter, valid_store, _initial = publisher( + "gateway", + ) + structural = MutableFragmentPublisher(valid_adapter) + structural.state_reader = lambda: corrupt_state + registry = FragmentAdapterRegistry(enabled=True) + assert_value_free_error( + self, + ValueError, + lambda: registry.register(structural), + raw_value, + ) + self.assertEqual(registry.provider_ids, ()) + self.assertEqual(valid_store.writes, 1) + + live_registry = FragmentAdapterRegistry(enabled=True) + self.assertTrue(live_registry.register(valid_adapter)) + compiler = live_registry.create_compiler( + InMemoryCompiledLatticeStateStore(), + ) + valid_store._state = corrupt_state + self.assertTrue( + compiler.invalidate( + "gateway", + 1, + "noncanonical durable provider ID", + ) + ) + run = compiler.drain() + self.assertIsNot(run.status, CompileStatus.FRESH) + self.assertIsNone(run.plan) + self.assertTrue(any(issue.code == "provider_read_failed" for issue in run.issues)) + for issue in run.issues: + self.assertNotIn(raw_value, issue.detail) + self.assertEqual(valid_store.writes, 1) + + def test_hostile_durable_state_fails_value_free_after_restart(self): + """Every corrupt scalar and snapshot hook is sanitized on read.""" + + def mutate_state(field, value): + return lambda state: object.__setattr__( + state, + field, + value, + ) + + def mutate_snapshot(field, value): + return lambda state: object.__setattr__( + state.snapshot, + field, + value, + ) + + def replace_with_subclass(state): + replacement = object.__new__( + FragmentAdapterStateSubclass, + ) + for field in ( + "provider_id", + "generation", + "semantic_fingerprint", + "snapshot", + "removed", + ): + object.__setattr__( + replacement, + field, + getattr(state, field), + ) + return replacement + + cases = ( + mutate_state( + "provider_id", + HostileText("gateway"), + ), + mutate_state( + "generation", + HostileInt(1), + ), + mutate_state( + "semantic_fingerprint", + HostileText("fingerprint"), + ), + mutate_state( + "removed", + HostileText("false"), + ), + mutate_snapshot( + "provider_id", + HostileText("gateway"), + ), + mutate_snapshot( + "generation", + HostileInt(1), + ), + mutate_snapshot( + "topology_fragment", + HostileMapping(), + ), + replace_with_subclass, + ) + for corrupt in cases: + with self.subTest(corrupt=corrupt): + adapter, state_store, _initial = publisher( + "gateway", + ) + state = state_store.load() + replacement = corrupt(state) + if replacement is not None: + state_store._state = replacement + writes = state_store.writes + assert_value_free_error( + self, + FragmentAdapterReadError, + lambda: DurableFragmentAdapter( + "gateway", + state_store, + ), + HostileText.secret, + ) + self.assertEqual(state_store.writes, writes) def test_conflicting_atomic_write_leaves_requested_generation_pending(self): """A rejected CAS never presents an uncommitted fragment as current.""" @@ -275,6 +1254,72 @@ def test_invalid_discovery_batch_is_transactional(self): self.assertEqual(registry.provider_ids, ()) + def test_hostile_adapter_and_state_hooks_fail_value_free(self): + """Every registry entry path sanitizes adapter and state faults.""" + adapter, state_store, _initial = publisher("gateway") + state = adapter.read_state() + hostile_state = object.__new__( + HostileFragmentAdapterState, + ) + for field in ( + "provider_id", + "generation", + "semantic_fingerprint", + "snapshot", + "removed", + ): + object.__setattr__( + hostile_state, + field, + object.__getattribute__(state, field), + ) + + register_registry = FragmentAdapterRegistry(enabled=True) + assert_value_free_error( + self, + ValueError, + lambda: register_registry.register( + HostileRegistryAdapter(), + ), + HostileText.secret, + ) + self.assertEqual(register_registry.provider_ids, ()) + + discovered = MutableFragmentPublisher(adapter) + discovered.state_reader = lambda: hostile_state + discover_registry = FragmentAdapterRegistry(enabled=True) + assert_value_free_error( + self, + ValueError, + lambda: discover_registry.discover( + (FragmentComponent(discovered),), + ), + HostileText.secret, + ) + self.assertEqual(discover_registry.provider_ids, ()) + + compiling = MutableFragmentPublisher(adapter) + compiler_registry = FragmentAdapterRegistry(enabled=True) + self.assertTrue( + compiler_registry.register(compiling), + ) + + def fail_state_read(): + raise RuntimeError(HostileText.secret) + + compiling.state_reader = fail_state_read + compiled_store = InMemoryCompiledLatticeStateStore() + assert_value_free_error( + self, + ValueError, + lambda: compiler_registry.create_compiler( + compiled_store, + ), + HostileText.secret, + ) + self.assertEqual(compiled_store.writes, 0) + self.assertEqual(state_store.writes, 1) + def test_register_unregister_only_before_compiler_is_sealed(self): """Runtime membership changes cannot silently alter compiler inputs.""" adapter, _store, _initial = publisher("gateway") diff --git a/apps/predbat/tests/test_lattice_fragment_tombstone_compilation.py b/apps/predbat/tests/test_lattice_fragment_tombstone_compilation.py index cb251b76c..ad6fac191 100644 --- a/apps/predbat/tests/test_lattice_fragment_tombstone_compilation.py +++ b/apps/predbat/tests/test_lattice_fragment_tombstone_compilation.py @@ -14,6 +14,7 @@ ) from lattice_fragment_adapters import ( # noqa: E402 DurableFragmentAdapter, + FragmentAdapterReadError, FragmentAdapterState, FragmentAdapterRegistry, FragmentAdapterRemoved, @@ -23,6 +24,69 @@ from tests.test_lattice_autoconfig import snapshot # noqa: E402 +def assert_value_free_read_error(test_case, action, raw_value): + """Assert compiler structural reads retain no source-controlled value.""" + try: + action() + except FragmentAdapterReadError as error: + test_case.assertIs(type(error), FragmentAdapterReadError) + test_case.assertNotIn(raw_value, str(error)) + test_case.assertNotIn(raw_value, repr(error)) + test_case.assertNotIn(raw_value, repr(error.args)) + test_case.assertNotIn(raw_value, error.args) + test_case.assertIsNone(error.__cause__) + test_case.assertIsNone(error.__context__) + else: + test_case.fail("expected a value-free fragment read error") + + +class HostileText(str): + """String subclass whose scalar hooks expose a fake provider secret.""" + + secret = "provider-secret-value" + + def _fail(self, *_args, **_kwargs): + """Fail if validation invokes any subclass-controlled hook.""" + raise RuntimeError(self.secret) + + strip = _fail + __format__ = _fail + __iter__ = _fail + __len__ = _fail + __hash__ = _fail + __eq__ = _fail + __lt__ = _fail + + +class HostileMapping(dict): + """Mapping whose traversal exposes a fake provider secret.""" + + def items(self): + """Fail if fingerprint validation traverses corrupt state.""" + raise RuntimeError(HostileText.secret) + + +class HostileFragmentAdapterState(FragmentAdapterState): + """State subclass whose validator must never be dispatched.""" + + def __post_init__(self): + """Expose a fake secret if subclass dispatch occurs.""" + raise RuntimeError(HostileText.secret) + + +class StructuralAdapter: + """Minimal compiler-facing adapter for hostile-read tests.""" + + def __init__(self, provider_id, reader): + """Store one provider scalar and structural reader.""" + self.provider_id = provider_id + self._reader = reader + + def read_state(self): + """Delegate to the hostile structural reader.""" + return self._reader() + + def publisher(provider_id, node_id): """Create one seeded integration-owned fragment publisher.""" store = InMemoryFragmentAdapterStateStore() @@ -72,7 +136,9 @@ def test_live_snapshot_is_exact_and_removed_snapshot_is_empty(self): """Translation changes only a durable removal state.""" adapter, _store, initial = publisher("gateway", "GW-INV") - self.assertIs(_compiler_fragment_snapshot(adapter), initial) + live = _compiler_fragment_snapshot(adapter) + self.assertEqual(live, initial) + self.assertIsNot(live, initial) self.assertTrue(adapter.remove(2, "integration removed")) with self.assertRaises(FragmentAdapterRemoved): @@ -89,6 +155,70 @@ def test_live_snapshot_is_exact_and_removed_snapshot_is_empty(self): self.assertEqual(tombstone.role_assignments, ()) self.assertEqual(tombstone.config_projections, ()) + def test_hostile_structural_reads_fail_value_free_without_dispatch(self): + """Compiler reads sanitize faults, subclasses, and corrupt snapshots.""" + adapter, state_store, _initial = publisher( + "gateway", + "GW-INV", + ) + state = state_store.load() + hostile_state = object.__new__( + HostileFragmentAdapterState, + ) + for field in ( + "provider_id", + "generation", + "semantic_fingerprint", + "snapshot", + "removed", + ): + object.__setattr__( + hostile_state, + field, + getattr(state, field), + ) + + cases = ( + StructuralAdapter( + "gateway", + lambda: (_ for _ in ()).throw( + RuntimeError(HostileText.secret), + ), + ), + StructuralAdapter( + HostileText("gateway"), + lambda: state, + ), + StructuralAdapter( + "gateway", + lambda: hostile_state, + ), + ) + for structural_adapter in cases: + with self.subTest(adapter=structural_adapter): + writes = state_store.writes + assert_value_free_read_error( + self, + lambda: _compiler_fragment_snapshot( + structural_adapter, + ), + HostileText.secret, + ) + self.assertEqual(state_store.writes, writes) + + object.__setattr__( + state.snapshot, + "topology_fragment", + HostileMapping(), + ) + writes = state_store.writes + assert_value_free_read_error( + self, + lambda: _compiler_fragment_snapshot(adapter), + HostileText.secret, + ) + self.assertEqual(state_store.writes, writes) + def test_all_removed_publishes_deterministic_empty_plan_and_restarts(self): """All tombstones settle, persist, and restore as one empty plan.""" alpha, alpha_store, _alpha = publisher("alpha", "ALPHA-INV") diff --git a/apps/predbat/tests/test_lattice_gateway_fragment.py b/apps/predbat/tests/test_lattice_gateway_fragment.py index 553ef3d41..12adaec93 100644 --- a/apps/predbat/tests/test_lattice_gateway_fragment.py +++ b/apps/predbat/tests/test_lattice_gateway_fragment.py @@ -124,16 +124,31 @@ def load(self): adapter.ingest_retained_topology(topology(), online=True) state_store.fail_reads = True - with self.assertRaisesRegex( - FragmentAdapterReadError, - "durable Gateway fragment unavailable", - ): - adapter.lattice_fragment_adapter() - with self.assertRaisesRegex( - FragmentAdapterReadError, - "durable Gateway fragment unavailable", + raw_value = "durable Gateway fragment unavailable" + for action in ( + adapter.lattice_fragment_adapter, + lambda: adapter.set_liveness(False), ): - adapter.set_liveness(False) + with self.subTest(action=action): + try: + action() + except FragmentAdapterReadError as error: + self.assertIs( + type(error), + FragmentAdapterReadError, + ) + self.assertNotIn(raw_value, str(error)) + self.assertNotIn(raw_value, repr(error)) + self.assertNotIn( + raw_value, + repr(error.args), + ) + self.assertIsNone(error.__cause__) + self.assertIsNone(error.__context__) + else: + self.fail( + "expected a value-free fragment read error", + ) def test_retained_topology_publishes_immutable_reference_snapshot(self): """A retained fragment becomes detached provider-local metadata."""