From b930e1aac0a3d32bc7cc13cfd2e415b06fdcea9f Mon Sep 17 00:00:00 2001 From: Mark Gascoyne Date: Tue, 28 Jul 2026 18:33:22 +0100 Subject: [PATCH] feat(lattice): add Carbon and Axle reference fragments --- apps/predbat/lattice_axle_fragment.py | 167 ++++++ apps/predbat/lattice_carbon_fragment.py | 120 +++++ .../lattice_reference_feed_fragment.py | 480 ++++++++++++++++++ .../tests/test_lattice_axle_fragment.py | 272 ++++++++++ .../tests/test_lattice_carbon_fragment.py | 246 +++++++++ .../test_lattice_reference_feed_fragment.py | 439 ++++++++++++++++ 6 files changed, 1724 insertions(+) create mode 100644 apps/predbat/lattice_axle_fragment.py create mode 100644 apps/predbat/lattice_carbon_fragment.py create mode 100644 apps/predbat/lattice_reference_feed_fragment.py create mode 100644 apps/predbat/tests/test_lattice_axle_fragment.py create mode 100644 apps/predbat/tests/test_lattice_carbon_fragment.py create mode 100644 apps/predbat/tests/test_lattice_reference_feed_fragment.py diff --git a/apps/predbat/lattice_axle_fragment.py b/apps/predbat/lattice_axle_fragment.py new file mode 100644 index 000000000..472fb8b04 --- /dev/null +++ b/apps/predbat/lattice_axle_fragment.py @@ -0,0 +1,167 @@ +# ----------------------------------------------------------------------------- +# Predbat Home Battery System - Axle reference-feed Lattice adapter +# Copyright Trefor Southwell 2026 - All Rights Reserved +# This application maybe used for personal use only and not for commercial use +# ----------------------------------------------------------------------------- +"""Pure Axle flexibility-session normalization for Lattice fragments. + +Managed site IDs and BYOK installation IDs are reduced to provider-local +digests. Live VPP/session state is deliberately absent: observing a session +can affect planning, but it does not grant battery or account control +authority. The live Axle component is not imported or registered. +""" + +from dataclasses import dataclass + +from lattice_reference_feed_fragment import ( + HEALTH_UNCHANGED, + ReferenceFeedFragmentPublisher, + ReferenceFeedSource, + _bounded_text, + canonical_provider_id, + provider_local_digest, +) + + +_CONNECTION_KINDS = frozenset(("byok-local", "managed-site")) + + +@dataclass(frozen=True) +class AxleReferenceFeedSnapshot: + """One provider-local flexibility-session observation source.""" + + provider_id: str + connection_kind: str + source_digest: str + entity_id: str + + def __post_init__(self): + """Validate bounded non-correlating source metadata.""" + provider_id = canonical_provider_id(self.provider_id) + connection_kind = _bounded_text( + self.connection_kind, + "connection_kind", + 32, + ).lower() + if connection_kind not in _CONNECTION_KINDS: + raise ValueError( + "connection_kind must be byok-local or managed-site", + ) + digest = _bounded_text( + self.source_digest, + "source_digest", + 32, + ).lower() + if len(digest) != 32 or any(character not in "0123456789abcdef" for character in digest): + raise ValueError( + "source_digest must be a 32-character hexadecimal digest", + ) + source = ReferenceFeedSource( + source_id="{}-{}".format(connection_kind, digest), + feed_type="flex-session", + entity_id=self.entity_id, + ) + object.__setattr__(self, "provider_id", provider_id) + object.__setattr__(self, "connection_kind", connection_kind) + object.__setattr__(self, "source_digest", digest) + object.__setattr__(self, "entity_id", source.entity_id) + + @classmethod + def managed( + cls, + provider_id, + site_id, + entity_id, + installation_privacy_key, + ): + """Normalize one provider-managed site without a global assertion.""" + provider_id = canonical_provider_id(provider_id) + site_id = _bounded_text(site_id, "site_id", 256) + return cls( + provider_id=provider_id, + connection_kind="managed-site", + source_digest=provider_local_digest( + provider_id, + "axle-managed-site", + site_id, + installation_privacy_key, + ), + entity_id=entity_id, + ) + + @classmethod + def byok( + cls, + provider_id, + local_instance_id, + entity_id, + installation_privacy_key, + ): + """Normalize one installation-local BYOK identity. + + ``local_instance_id`` must be a stable installation-local opaque value, + not an account token or credential. Only its provider-scoped digest is + retained. + """ + provider_id = canonical_provider_id(provider_id) + local_instance_id = _bounded_text( + local_instance_id, + "local_instance_id", + 256, + ) + return cls( + provider_id=provider_id, + connection_kind="byok-local", + source_digest=provider_local_digest( + provider_id, + "axle-byok-instance", + local_instance_id, + installation_privacy_key, + ), + entity_id=entity_id, + ) + + def reference_source(self): + """Return the generic immutable source projection.""" + return ReferenceFeedSource( + source_id="{}-{}".format( + self.connection_kind, + self.source_digest, + ), + feed_type="flex-session", + entity_id=self.entity_id, + ) + + +class AxleReferenceFeedPublisher(ReferenceFeedFragmentPublisher): + """Default-off publisher for explicit Axle feed snapshots.""" + + def __init__(self, provider_id, state_store, enabled=False): + """Create an unwired Axle reference publisher.""" + super().__init__( + provider_id, + "Axle Flex", + state_store, + enabled=enabled, + ) + + def ingest_snapshot( + self, + source_generation, + snapshot, + health=HEALTH_UNCHANGED, + feedback_token=None, + ): + """Publish one explicit Axle source generation.""" + if not isinstance(snapshot, AxleReferenceFeedSnapshot): + raise ValueError( + "snapshot must be AxleReferenceFeedSnapshot", + ) + if snapshot.provider_id != self.provider_id: + raise ValueError("snapshot provider_id does not match publisher") + return self.ingest_sources( + source_generation, + (snapshot.reference_source(),), + health=health, + feedback_token=feedback_token, + ) diff --git a/apps/predbat/lattice_carbon_fragment.py b/apps/predbat/lattice_carbon_fragment.py new file mode 100644 index 000000000..647d09c60 --- /dev/null +++ b/apps/predbat/lattice_carbon_fragment.py @@ -0,0 +1,120 @@ +# ----------------------------------------------------------------------------- +# Predbat Home Battery System - Carbon reference-feed Lattice adapter +# Copyright Trefor Southwell 2026 - All Rights Reserved +# This application maybe used for personal use only and not for commercial use +# ----------------------------------------------------------------------------- +"""Pure Carbon reference-feed normalization over the generic fragment surface. + +The live Carbon API component is intentionally not imported. Raw postcode +data is accepted only by the normalizer and replaced immediately with a +provider-local opaque digest. The resulting fragment is REFERENCE-only and +cannot grant configuration or control authority. +""" + +from dataclasses import dataclass + +from lattice_reference_feed_fragment import ( + HEALTH_UNCHANGED, + ReferenceFeedFragmentPublisher, + ReferenceFeedSource, + _bounded_text, + canonical_provider_id, + provider_local_digest, +) + + +@dataclass(frozen=True) +class CarbonReferenceFeedSnapshot: + """One carbon-intensity source with no retained raw location.""" + + provider_id: str + location_digest: str + entity_id: str + + def __post_init__(self): + """Validate only the opaque digest and local read reference.""" + provider_id = canonical_provider_id(self.provider_id) + digest = _bounded_text( + self.location_digest, + "location_digest", + 32, + ).lower() + if len(digest) != 32 or any(character not in "0123456789abcdef" for character in digest): + raise ValueError( + "location_digest must be a 32-character hexadecimal digest", + ) + source = ReferenceFeedSource( + source_id="location-{}".format(digest), + feed_type="carbon-intensity", + entity_id=self.entity_id, + ) + object.__setattr__(self, "provider_id", provider_id) + object.__setattr__(self, "location_digest", digest) + object.__setattr__(self, "entity_id", source.entity_id) + + @classmethod + def from_postcode( + cls, + provider_id, + postcode, + entity_id, + installation_privacy_key, + ): + """Replace one raw postcode with a provider-local opaque digest.""" + provider_id = canonical_provider_id(provider_id) + postcode = _bounded_text(postcode, "postcode", 32) + normalized = "".join(postcode.upper().split()) + if not normalized: + raise ValueError("postcode must contain non-whitespace characters") + return cls( + provider_id=provider_id, + location_digest=provider_local_digest( + provider_id, + "carbon-location", + normalized, + installation_privacy_key, + ), + entity_id=entity_id, + ) + + def reference_source(self): + """Return the generic immutable source projection.""" + return ReferenceFeedSource( + source_id="location-{}".format(self.location_digest), + feed_type="carbon-intensity", + entity_id=self.entity_id, + ) + + +class CarbonReferenceFeedPublisher(ReferenceFeedFragmentPublisher): + """Default-off publisher for explicit Carbon feed snapshots.""" + + def __init__(self, provider_id, state_store, enabled=False): + """Create an unwired Carbon reference publisher.""" + super().__init__( + provider_id, + "Carbon Intensity", + state_store, + enabled=enabled, + ) + + def ingest_snapshot( + self, + source_generation, + snapshot, + health=HEALTH_UNCHANGED, + feedback_token=None, + ): + """Publish one explicit Carbon source generation.""" + if not isinstance(snapshot, CarbonReferenceFeedSnapshot): + raise ValueError( + "snapshot must be CarbonReferenceFeedSnapshot", + ) + if snapshot.provider_id != self.provider_id: + raise ValueError("snapshot provider_id does not match publisher") + return self.ingest_sources( + source_generation, + (snapshot.reference_source(),), + health=health, + feedback_token=feedback_token, + ) diff --git a/apps/predbat/lattice_reference_feed_fragment.py b/apps/predbat/lattice_reference_feed_fragment.py new file mode 100644 index 000000000..b556cc82d --- /dev/null +++ b/apps/predbat/lattice_reference_feed_fragment.py @@ -0,0 +1,480 @@ +# ----------------------------------------------------------------------------- +# Predbat Home Battery System - Lattice reference-feed 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 provider-local reference data feeds. + +Reference feeds are observations such as carbon intensity or a flexibility +session. Their presence may influence planning, but it does not prove control +authority over a battery, inverter, charger, or provider account. This module +therefore publishes only provider-qualified REFERENCE aliases. It never emits +strong cross-provider identity assertions, role assignments, configuration +projections, or topology capabilities. + +The caller owns discovery and supplies a monotonically increasing source +generation. Accepted source, liveness, and removal changes are published +through the existing durable fragment contract and invalidate its compiler +subscribers. Nothing in this module imports or registers a live integration. +""" + +# cspell:ignore autoconfig + +import hashlib +import hmac +import re +import threading +from dataclasses import dataclass +from typing import Optional + +from lattice_autoconfig import ( + AliasRole, + ProviderAlias, + ProviderHealth, + ProviderSnapshot, + _plain, +) +from lattice_fragment_adapters import ( + DurableFragmentAdapter, + FragmentAdapterConflict, + FragmentAdapterReadError, + FragmentAdapterRemoved, +) + + +HEALTH_UNCHANGED = object() +_OPAQUE_ID_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]*$") +_PROVIDER_ID_RE = re.compile(r"^[a-z0-9][a-z0-9._-]*$") +_FEED_TYPE_RE = re.compile(r"^[a-z0-9][a-z0-9._-]*$") +_ENTITY_ID_RE = re.compile(r"^[a-z_]+\.[a-z0-9_]+$") +_PRIVACY_KEY_MINIMUM = 16 +_PRIVACY_KEY_MAXIMUM = 128 +_DIGEST_PROTOCOL = b"predbat-reference-feed-v1" + + +def _bounded_text(value, name, maximum): + """Return bounded text containing only UTF-8-encodable Unicode scalars.""" + if not isinstance(value, str): + raise ValueError("{} must be a non-empty string".format(name)) + if len(value) > maximum: + raise ValueError("{} must be at most {} characters".format(name, maximum)) + if any(ord(character) < 32 or ord(character) == 127 for character in value): + raise ValueError("{} must not contain control characters".format(name)) + if any(0xD800 <= ord(character) <= 0xDFFF for character in value): + raise ValueError("{} must contain only UTF-8 encodable text".format(name)) + normalized = value.strip() + if not normalized: + raise ValueError("{} must be a non-empty string".format(name)) + return normalized + + +def canonical_provider_id(value): + """Validate one bounded canonical provider identifier.""" + normalized = _bounded_text(value, "provider_id", 128) + if normalized != value or _PROVIDER_ID_RE.fullmatch(normalized) is None: + raise ValueError( + "provider_id must use canonical lowercase identifier characters", + ) + return normalized + + +def _installation_privacy_key(value): + """Validate but never retain or render one caller-owned privacy key.""" + if not isinstance(value, bytes): + raise ValueError("installation_privacy_key must be bytes") + if not _PRIVACY_KEY_MINIMUM <= len(value) <= _PRIVACY_KEY_MAXIMUM: + raise ValueError( + "installation_privacy_key must be between 16 and 128 bytes", + ) + return value + + +def _length_prefixed_payload(values): + """Frame digest inputs unambiguously before keyed hashing.""" + payload = bytearray(_DIGEST_PROTOCOL) + for value in values: + encoded = value.encode("utf-8") + payload.extend(len(encoded).to_bytes(4, byteorder="big")) + payload.extend(encoded) + return bytes(payload) + + +def provider_local_digest( + provider_id, + namespace, + value, + installation_privacy_key, +): + """Return a keyed opaque digest scoped to one provider installation. + + The raw value and caller-owned key are used only for this HMAC operation + and are never retained. Length framing prevents tuple ambiguity. This is + a provider-local reference key, not a globally correlatable identity alias. + """ + provider_id = canonical_provider_id(provider_id) + namespace = _bounded_text(namespace, "namespace", 64) + if _PROVIDER_ID_RE.fullmatch(namespace) is None: + raise ValueError( + "namespace must use canonical lowercase identifier characters", + ) + value = _bounded_text(value, "identity value", 256) + privacy_key = _installation_privacy_key(installation_privacy_key) + payload = _length_prefixed_payload((provider_id, namespace, value)) + return hmac.new(privacy_key, payload, hashlib.sha256).hexdigest()[:32] + + +def _source_generation(value): + """Validate one caller-owned monotonic source generation.""" + if not isinstance(value, int) or isinstance(value, bool) or value < 0: + raise ValueError("source_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") + + +@dataclass(frozen=True) +class ReferenceFeedSource: + """One bounded provider-local observation source. + + ``source_id`` must already be a non-secret opaque identifier. It is never + promoted to ``ProviderIdentityAlias`` and therefore cannot correlate nodes + across providers. ``entity_id`` is an optional local read reference only; + publishing it does not create a PredBat configuration projection. + """ + + source_id: str + feed_type: str + entity_id: Optional[str] = None + + def __post_init__(self): + """Normalize the bounded deterministic source metadata.""" + source_id = _bounded_text(self.source_id, "source_id", 128) + if _OPAQUE_ID_RE.fullmatch(source_id) is None: + raise ValueError("source_id must be an opaque identifier") + feed_type = _bounded_text(self.feed_type, "feed_type", 64).lower() + if _FEED_TYPE_RE.fullmatch(feed_type) is None: + raise ValueError("feed_type must use lowercase identifier characters") + entity_id = self.entity_id + if entity_id is not None: + entity_id = _bounded_text(entity_id, "entity_id", 255).lower() + if _ENTITY_ID_RE.fullmatch(entity_id) is None: + raise ValueError("entity_id must be a Home Assistant entity identifier") + object.__setattr__(self, "source_id", source_id) + object.__setattr__(self, "feed_type", feed_type) + object.__setattr__(self, "entity_id", entity_id) + + def node_id(self, provider_id): + """Return one stable provider-qualified topology node identity.""" + return "reference-feed:{}:{}".format(provider_id, self.source_id) + + +def _normalize_sources(sources): + """Freeze, sort, and collision-check a complete source snapshot.""" + try: + sources = tuple(sources) + except TypeError as exc: + raise ValueError( + "sources must be an iterable of ReferenceFeedSource", + ) from exc + if any(not isinstance(source, ReferenceFeedSource) for source in sources): + raise ValueError("sources must contain only ReferenceFeedSource values") + if not sources: + raise ValueError("sources must contain at least one ReferenceFeedSource") + + seen = {} + for source in sources: + previous = seen.get(source.source_id) + if previous is not None: + raise FragmentAdapterConflict( + "reference feed contains duplicate source {}".format( + source.source_id, + ) + ) + seen[source.source_id] = source + return tuple( + sorted( + sources, + key=lambda item: ( + item.source_id, + item.feed_type, + item.entity_id or "", + ), + ) + ) + + +def _source_node(source, provider_id): + """Project one read reference without deriving any capability.""" + attributes = { + "feedType": source.feed_type, + "sourceScope": "provider-local", + } + if source.entity_id is not None: + attributes["entityId"] = source.entity_id + return { + "id": source.node_id(provider_id), + "kind": "data-source", + "attributes": attributes, + "accessPaths": [ + { + "id": "reference-feed-read", + "provider": provider_id, + "preference": 0, + } + ], + "capabilities": [], + } + + +def _topology_document( + provider_id, + producer_name, + source_generation, + sources, +): + """Build one deterministic provider-owned topology fragment.""" + return { + "topologyVersion": "0.3.0", + "scope": "fragment", + "docVersion": source_generation, + "producer": { + "name": producer_name, + "provider": provider_id, + "authority": 0, + }, + "nodes": [_source_node(source, provider_id) for source in sources], + } + + +def _reference_aliases(provider_id, sources): + """Publish only provider-qualified REFERENCE aliases.""" + return tuple( + ProviderAlias( + name="feed:{}:{}".format( + source.feed_type, + source.source_id, + ), + node_id=source.node_id(provider_id), + roles=frozenset((AliasRole.REFERENCE,)), + ) + for source in sources + ) + + +class ReferenceFeedFragmentPublisher: + """Adapt explicit reference sources into one durable compiler fragment.""" + + 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") + self._enabled = enabled + provider_id = canonical_provider_id(provider_id) + self._producer_name = _bounded_text( + producer_name, + "producer_name", + 128, + ) + 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 source_generation(self): + """Fresh-read the current caller-owned source 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 source, 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_sources( + self, + source_generation, + sources, + health=HEALTH_UNCHANGED, + feedback_token=None, + ): + """Publish one accepted complete reference-source snapshot.""" + if not self._enabled: + return False + source_generation = _source_generation(source_generation) + sources = _normalize_sources(sources) + document = _topology_document( + self.provider_id, + self._producer_name, + source_generation, + sources, + ) + + with self._lock: + current = self._current_state() + if current is not None and current.removed: + raise FragmentAdapterRemoved( + "provider {} was removed at generation {}; reference " + "feed cannot re-enrol it".format( + self.provider_id, + current.generation, + ) + ) + if current is None: + next_health = ProviderHealth.DEGRADED if health is HEALTH_UNCHANGED else _provider_health(health) + else: + previous = _plain(current.snapshot.topology_fragment) + previous_generation = previous["docVersion"] + if source_generation < previous_generation: + return False + if source_generation == previous_generation and document != previous: + raise FragmentAdapterConflict( + "provider {} reused reference-feed source generation " + "{} for different content".format( + self.provider_id, + source_generation, + ) + ) + next_health = current.snapshot.health if health is HEALTH_UNCHANGED else _provider_health(health) + if source_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, sources), + identity_aliases=(), + role_assignments=(), + config_projections=(), + ) + published = self._adapter.publish( + snapshot, + "reference feed changed to source generation {}".format( + source_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 source contents.""" + if not self._enabled: + return False + health = _provider_health(health) + with self._lock: + current = self._current_state() + if current is None: + raise FragmentAdapterReadError( + "reference feed 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=(), + role_assignments=(), + config_projections=(), + ) + return self._adapter.publish( + snapshot, + "reference feed liveness changed to {}".format(health.value), + feedback_token=feedback_token, + ) + + def remove(self, feedback_token=None): + """Publish an irreversible provider-removal tombstone.""" + if not self._enabled: + return False + with self._lock: + current = self._current_state() + if current is None: + raise FragmentAdapterReadError( + "reference feed must be seeded before removal", + ) + if current.removed: + return False + return self._adapter.remove( + current.generation + 1, + "reference feed integration removed", + feedback_token=feedback_token, + ) diff --git a/apps/predbat/tests/test_lattice_axle_fragment.py b/apps/predbat/tests/test_lattice_axle_fragment.py new file mode 100644 index 000000000..05fa02628 --- /dev/null +++ b/apps/predbat/tests/test_lattice_axle_fragment.py @@ -0,0 +1,272 @@ +"""Tests for the pure Axle reference-feed normalizer and 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 AliasRole, ProviderHealth # noqa: E402 +from lattice_axle_fragment import ( # noqa: E402 + AxleReferenceFeedPublisher, + AxleReferenceFeedSnapshot, +) +from lattice_fragment_adapters import ( # noqa: E402 + FragmentAdapterConflict, + InMemoryFragmentAdapterStateStore, +) + + +PRIVACY_KEY_A = b"axle-installation-privacy-key-a" +PRIVACY_KEY_B = b"axle-installation-privacy-key-b" + + +def assert_value_free_validation_error(test_case, action, raw_value): + """Assert validation errors retain no secret-bearing source object.""" + secret_marker = raw_value[:-1] + try: + action() + except ValueError as error: + test_case.assertIs(type(error), ValueError) + test_case.assertNotIn(secret_marker, str(error)) + test_case.assertNotIn(secret_marker, repr(error)) + test_case.assertNotIn(secret_marker, 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 TestAxleReferenceFeedSnapshot(unittest.TestCase): + """Managed and BYOK sources remain provider-local references.""" + + def test_managed_site_is_opaque_and_provider_scoped(self): + """A managed provider site never becomes a global identity alias.""" + first = AxleReferenceFeedSnapshot.managed( + "axle-home", + "site-live-123", + "binary_sensor.predbat_axle_event", + PRIVACY_KEY_A, + ) + equivalent = AxleReferenceFeedSnapshot.managed( + "axle-home", + "site-live-123", + "binary_sensor.predbat_axle_event", + PRIVACY_KEY_A, + ) + other_provider = AxleReferenceFeedSnapshot.managed( + "axle-other", + "site-live-123", + "binary_sensor.predbat_axle_event", + PRIVACY_KEY_A, + ) + other_key = AxleReferenceFeedSnapshot.managed( + "axle-home", + "site-live-123", + "binary_sensor.predbat_axle_event", + PRIVACY_KEY_B, + ) + + self.assertEqual(first, equivalent) + self.assertNotEqual(first.source_digest, other_provider.source_digest) + self.assertNotEqual(first.source_digest, other_key.source_digest) + self.assertEqual(first.provider_id, "axle-home") + self.assertNotIn("site-live-123", repr(first)) + self.assertNotIn(PRIVACY_KEY_A.decode("ascii"), repr(first)) + self.assertEqual(first.connection_kind, "managed-site") + + def test_byok_uses_only_installation_local_opaque_input(self): + """BYOK source material is hashed and cannot correlate globally.""" + first = AxleReferenceFeedSnapshot.byok( + "axle-home", + "installation-random-id", + "binary_sensor.predbat_axle_event", + PRIVACY_KEY_A, + ) + other_provider = AxleReferenceFeedSnapshot.byok( + "axle-other", + "installation-random-id", + "binary_sensor.predbat_axle_event", + PRIVACY_KEY_A, + ) + + self.assertEqual(first.connection_kind, "byok-local") + self.assertNotEqual(first.source_digest, other_provider.source_digest) + self.assertNotIn("installation-random-id", repr(first)) + + def test_connection_kind_and_digest_are_closed_fields(self): + """Arbitrary account modes and raw tokens cannot enter metadata.""" + with self.assertRaisesRegex(ValueError, "connection_kind"): + AxleReferenceFeedSnapshot( + provider_id="axle-home", + connection_kind="oauth-account", + source_digest="0" * 32, + entity_id="binary_sensor.predbat_axle_event", + ) + with self.assertRaisesRegex(ValueError, "hexadecimal digest"): + AxleReferenceFeedSnapshot( + provider_id="axle-home", + connection_kind="byok-local", + source_digest="raw-account-token", + entity_id="binary_sensor.predbat_axle_event", + ) + + def test_hash_inputs_are_control_free_bounded_and_private(self): + """Site and BYOK inputs fail closed without retaining secrets.""" + for raw_value in ("site\x00secret", "site\nsecret", "x" * 257): + with self.subTest(raw_value=raw_value): + with self.assertRaises(ValueError) as raised: + AxleReferenceFeedSnapshot.managed( + "axle-home", + raw_value, + "binary_sensor.predbat_axle_event", + PRIVACY_KEY_A, + ) + error = str(raised.exception) + self.assertNotIn(raw_value, error) + self.assertNotIn(PRIVACY_KEY_A.decode("ascii"), error) + + with self.assertRaisesRegex(ValueError, "control characters"): + AxleReferenceFeedSnapshot.byok( + "axle-home", + "local\x00secret", + "binary_sensor.predbat_axle_event", + PRIVACY_KEY_A, + ) + + def test_managed_site_surrogate_error_does_not_retain_source(self): + """Non-encodable managed identifiers fail before UTF-8 encoding.""" + raw_value = "private-managed-site\ud800" + assert_value_free_validation_error( + self, + lambda: AxleReferenceFeedSnapshot.managed( + "axle-home", + raw_value, + "binary_sensor.predbat_axle_event", + PRIVACY_KEY_A, + ), + raw_value, + ) + + def test_byok_surrogate_error_does_not_retain_source(self): + """Non-encodable BYOK identifiers fail before UTF-8 encoding.""" + raw_value = "private-byok-identity\ud800" + assert_value_free_validation_error( + self, + lambda: AxleReferenceFeedSnapshot.byok( + "axle-home", + raw_value, + "binary_sensor.predbat_axle_event", + PRIVACY_KEY_A, + ), + raw_value, + ) + + +class TestAxleReferenceFeedPublisher(unittest.TestCase): + """Axle session observation never grants VPP or battery authority.""" + + def test_managed_and_byok_are_reference_only(self): + """No connection mode creates roles, projections, or capabilities.""" + for index, snapshot in enumerate( + ( + AxleReferenceFeedSnapshot.managed( + "axle-{}".format(index), + "site-{}".format(index), + "binary_sensor.predbat_axle_event", + PRIVACY_KEY_A, + ) + for index in range(2) + ) + ): + adapter = AxleReferenceFeedPublisher( + "axle-{}".format(index), + InMemoryFragmentAdapterStateStore(), + enabled=True, + ) + self.assertTrue(adapter.ingest_snapshot(1, snapshot, health=True)) + published = adapter.read_snapshot() + + self.assertEqual(published.health, ProviderHealth.HEALTHY) + self.assertEqual( + published.aliases[0].roles, + frozenset((AliasRole.REFERENCE,)), + ) + self.assertEqual(published.identity_aliases, ()) + self.assertEqual(published.role_assignments, ()) + self.assertEqual(published.config_projections, ()) + self.assertEqual( + published.topology_fragment["nodes"][0]["capabilities"], + (), + ) + self.assertEqual( + published.topology_fragment["producer"]["authority"], + 0, + ) + + def test_session_entity_change_requires_new_source_generation(self): + """A local observation seam cannot mutate one generation in place.""" + adapter = AxleReferenceFeedPublisher( + "axle-home", + InMemoryFragmentAdapterStateStore(), + enabled=True, + ) + first = AxleReferenceFeedSnapshot.byok( + "axle-home", + "installation-random-id", + "binary_sensor.predbat_axle_event", + PRIVACY_KEY_A, + ) + changed = AxleReferenceFeedSnapshot.byok( + "axle-home", + "installation-random-id", + "binary_sensor.predbat_axle_event_v2", + PRIVACY_KEY_A, + ) + adapter.ingest_snapshot(8, first, health=True) + + with self.assertRaises(FragmentAdapterConflict): + adapter.ingest_snapshot(8, changed) + self.assertTrue(adapter.ingest_snapshot(9, changed)) + + def test_provider_mismatch_is_rejected_after_restart(self): + """A directly constructed valid digest remains provider-bound.""" + state_store = InMemoryFragmentAdapterStateStore() + adapter = AxleReferenceFeedPublisher( + "axle-home", + state_store, + enabled=True, + ) + matching = AxleReferenceFeedSnapshot.byok( + "axle-home", + "installation-random-id", + "binary_sensor.predbat_axle_event", + PRIVACY_KEY_A, + ) + self.assertTrue(adapter.ingest_snapshot(1, matching, health=True)) + + restarted = AxleReferenceFeedPublisher( + "axle-home", + state_store, + enabled=True, + ) + mismatched = AxleReferenceFeedSnapshot( + provider_id="axle-other", + connection_kind="byok-local", + source_digest="f" * 32, + entity_id="binary_sensor.predbat_axle_event", + ) + before = restarted.read_state() + with self.assertRaisesRegex(ValueError, "does not match") as raised: + restarted.ingest_snapshot(2, mismatched) + self.assertEqual(restarted.read_state(), before) + self.assertNotIn("f" * 32, str(raised.exception)) + self.assertNotIn(PRIVACY_KEY_A.decode("ascii"), str(raised.exception)) + + +if __name__ == "__main__": + unittest.main() diff --git a/apps/predbat/tests/test_lattice_carbon_fragment.py b/apps/predbat/tests/test_lattice_carbon_fragment.py new file mode 100644 index 000000000..c82a00901 --- /dev/null +++ b/apps/predbat/tests/test_lattice_carbon_fragment.py @@ -0,0 +1,246 @@ +"""Tests for the pure Carbon reference-feed normalizer and 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 AliasRole, ProviderHealth # noqa: E402 +from lattice_carbon_fragment import ( # noqa: E402 + CarbonReferenceFeedPublisher, + CarbonReferenceFeedSnapshot, +) +from lattice_fragment_adapters import ( # noqa: E402 + FragmentAdapterConflict, + FragmentAdapterReadError, + InMemoryFragmentAdapterStateStore, +) + + +PRIVACY_KEY_A = b"carbon-installation-privacy-key-a" +PRIVACY_KEY_B = b"carbon-installation-privacy-key-b" + + +def assert_value_free_validation_error(test_case, action, raw_value): + """Assert validation errors retain no secret-bearing source object.""" + secret_marker = raw_value[:-1] + try: + action() + except ValueError as error: + test_case.assertIs(type(error), ValueError) + test_case.assertNotIn(secret_marker, str(error)) + test_case.assertNotIn(secret_marker, repr(error)) + test_case.assertNotIn(secret_marker, 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 TestCarbonReferenceFeedSnapshot(unittest.TestCase): + """Raw location input never enters durable fragment metadata.""" + + def test_postcode_is_replaced_by_provider_local_opaque_digest(self): + """Equivalent postcodes normalize while raw location is absent.""" + first = CarbonReferenceFeedSnapshot.from_postcode( + "carbon-home", + "SW1A 1AA", + "sensor.predbat_carbon_intensity", + PRIVACY_KEY_A, + ) + equivalent = CarbonReferenceFeedSnapshot.from_postcode( + "carbon-home", + "sw1a1aa", + "sensor.predbat_carbon_intensity", + PRIVACY_KEY_A, + ) + other_provider = CarbonReferenceFeedSnapshot.from_postcode( + "carbon-other", + "SW1A 1AA", + "sensor.predbat_carbon_intensity", + PRIVACY_KEY_A, + ) + other_key = CarbonReferenceFeedSnapshot.from_postcode( + "carbon-home", + "SW1A 1AA", + "sensor.predbat_carbon_intensity", + PRIVACY_KEY_B, + ) + + self.assertEqual(first, equivalent) + self.assertNotEqual( + first.location_digest, + other_provider.location_digest, + ) + self.assertNotEqual(first.location_digest, other_key.location_digest) + self.assertEqual(first.provider_id, "carbon-home") + self.assertEqual(len(first.location_digest), 32) + self.assertNotIn("SW1A", repr(first)) + self.assertNotIn("sw1a", repr(first)) + self.assertNotIn(PRIVACY_KEY_A.decode("ascii"), repr(first)) + + def test_snapshot_rejects_non_digest_identity(self): + """A caller cannot persist a raw postcode in the digest field.""" + with self.assertRaisesRegex(ValueError, "hexadecimal digest"): + CarbonReferenceFeedSnapshot( + provider_id="carbon-home", + location_digest="SW1A 1AA", + entity_id="sensor.predbat_carbon_intensity", + ) + + def test_postcode_hash_inputs_are_control_free_and_bounded(self): + """Raw locations and provider IDs fail closed without disclosure.""" + for postcode in ("SW1A\x001AA", "SW1A\n1AA", "x" * 33): + with self.subTest(postcode=postcode): + with self.assertRaises(ValueError) as raised: + CarbonReferenceFeedSnapshot.from_postcode( + "carbon-home", + postcode, + "sensor.predbat_carbon_intensity", + PRIVACY_KEY_A, + ) + error = str(raised.exception) + self.assertNotIn(postcode, error) + self.assertNotIn(PRIVACY_KEY_A.decode("ascii"), error) + + for provider_id in ("Carbon-home", "carbon-home\x00other", "x" * 129): + with self.subTest(provider_id=provider_id): + with self.assertRaisesRegex(ValueError, "provider_id"): + CarbonReferenceFeedSnapshot.from_postcode( + provider_id, + "SW1A 1AA", + "sensor.predbat_carbon_intensity", + PRIVACY_KEY_A, + ) + + def test_postcode_rejects_unpaired_surrogate_without_retaining_source(self): + """Non-encodable postcode input fails with a value-free error.""" + raw_value = "private-postcode\ud800" + assert_value_free_validation_error( + self, + lambda: CarbonReferenceFeedSnapshot.from_postcode( + "carbon-home", + raw_value, + "sensor.predbat_carbon_intensity", + PRIVACY_KEY_A, + ), + raw_value, + ) + + +class TestCarbonReferenceFeedPublisher(unittest.TestCase): + """Carbon adapter remains pure, default-off, and REFERENCE-only.""" + + def test_default_off_and_explicit_ingestion(self): + """Nothing is written until an enabled caller supplies a snapshot.""" + state_store = InMemoryFragmentAdapterStateStore() + adapter = CarbonReferenceFeedPublisher( + "carbon-home", + state_store, + ) + snapshot = CarbonReferenceFeedSnapshot.from_postcode( + "carbon-home", + "SW1A 1AA", + "sensor.predbat_carbon_intensity", + PRIVACY_KEY_A, + ) + + self.assertFalse(adapter.ingest_snapshot(1, snapshot, health=True)) + self.assertEqual(state_store.writes, 0) + with self.assertRaises(FragmentAdapterReadError): + adapter.read_state() + + adapter = CarbonReferenceFeedPublisher( + "carbon-home", + state_store, + enabled=True, + ) + self.assertTrue(adapter.ingest_snapshot(1, snapshot, health=True)) + published = adapter.read_snapshot() + serialized = repr(published) + + self.assertEqual(published.health, ProviderHealth.HEALTHY) + self.assertNotIn("SW1A", serialized) + self.assertNotIn(PRIVACY_KEY_A.decode("ascii"), serialized) + self.assertEqual( + published.aliases[0].roles, + frozenset((AliasRole.REFERENCE,)), + ) + self.assertEqual(published.identity_aliases, ()) + self.assertEqual(published.role_assignments, ()) + self.assertEqual(published.config_projections, ()) + self.assertEqual( + published.topology_fragment["nodes"][0]["capabilities"], + (), + ) + self.assertEqual( + published.topology_fragment["producer"]["authority"], + 0, + ) + + def test_location_change_requires_a_new_source_generation(self): + """A changed opaque location cannot reuse a source generation.""" + adapter = CarbonReferenceFeedPublisher( + "carbon-home", + InMemoryFragmentAdapterStateStore(), + enabled=True, + ) + first = CarbonReferenceFeedSnapshot.from_postcode( + "carbon-home", + "SW1A 1AA", + "sensor.predbat_carbon_intensity", + PRIVACY_KEY_A, + ) + changed = CarbonReferenceFeedSnapshot.from_postcode( + "carbon-home", + "EH1 1YZ", + "sensor.predbat_carbon_intensity", + PRIVACY_KEY_A, + ) + adapter.ingest_snapshot(4, first, health=True) + + with self.assertRaises(FragmentAdapterConflict): + adapter.ingest_snapshot(4, changed) + self.assertTrue(adapter.ingest_snapshot(5, changed)) + + def test_provider_mismatch_is_rejected_after_restart(self): + """A valid opaque digest cannot cross a provider boundary.""" + state_store = InMemoryFragmentAdapterStateStore() + adapter = CarbonReferenceFeedPublisher( + "carbon-home", + state_store, + enabled=True, + ) + matching = CarbonReferenceFeedSnapshot.from_postcode( + "carbon-home", + "SW1A 1AA", + "sensor.predbat_carbon_intensity", + PRIVACY_KEY_A, + ) + self.assertTrue(adapter.ingest_snapshot(1, matching, health=True)) + + restarted = CarbonReferenceFeedPublisher( + "carbon-home", + state_store, + enabled=True, + ) + mismatched = CarbonReferenceFeedSnapshot( + provider_id="carbon-other", + location_digest="0" * 32, + entity_id="sensor.predbat_carbon_intensity", + ) + before = restarted.read_state() + with self.assertRaisesRegex(ValueError, "does not match") as raised: + restarted.ingest_snapshot(2, mismatched) + self.assertEqual(restarted.read_state(), before) + self.assertNotIn("0" * 32, str(raised.exception)) + self.assertNotIn(PRIVACY_KEY_A.decode("ascii"), str(raised.exception)) + + +if __name__ == "__main__": + unittest.main() diff --git a/apps/predbat/tests/test_lattice_reference_feed_fragment.py b/apps/predbat/tests/test_lattice_reference_feed_fragment.py new file mode 100644 index 000000000..d4c4f85f1 --- /dev/null +++ b/apps/predbat/tests/test_lattice_reference_feed_fragment.py @@ -0,0 +1,439 @@ +"""Tests for the generic pure reference-feed 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_fragment_adapters import ( # noqa: E402 + FragmentAdapterConflict, + FragmentAdapterReadError, + FragmentAdapterRegistry, + FragmentAdapterRemoved, + InMemoryFragmentAdapterStateStore, +) +from lattice_reference_feed_fragment import ( # noqa: E402 + ReferenceFeedFragmentPublisher, + ReferenceFeedSource, + _length_prefixed_payload, + provider_local_digest, +) + + +PRIVACY_KEY_A = b"reference-feed-privacy-key-a" +PRIVACY_KEY_B = b"reference-feed-privacy-key-b" + + +def assert_value_free_validation_error(test_case, action, raw_value): + """Assert validation errors retain no secret-bearing source object.""" + secret_marker = raw_value[:-1] + try: + action() + except ValueError as error: + test_case.assertIs(type(error), ValueError) + test_case.assertNotIn(secret_marker, str(error)) + test_case.assertNotIn(secret_marker, repr(error)) + test_case.assertNotIn(secret_marker, 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") + + +def source( + source_id="source-a", + feed_type="reference-data", + entity_id="sensor.predbat_reference_data", +): + """Build one bounded provider-local source.""" + return ReferenceFeedSource( + source_id=source_id, + feed_type=feed_type, + entity_id=entity_id, + ) + + +def publisher(enabled=True, state_store=None): + """Build one generic publisher with its durable store.""" + state_store = state_store or InMemoryFragmentAdapterStateStore() + return ( + ReferenceFeedFragmentPublisher( + "reference-provider", + "Reference Provider", + state_store, + enabled=enabled, + ), + state_store, + ) + + +class TestReferenceFeedSource(unittest.TestCase): + """Reference source metadata is bounded and deterministic.""" + + def test_source_normalizes_without_accepting_arbitrary_metadata(self): + """Only opaque identity, feed kind, and local entity are retained.""" + item = ReferenceFeedSource( + "Source-A", + "CARBON-INTENSITY", + "SENSOR.PREDBAT_CARBON", + ) + + self.assertEqual(item.source_id, "Source-A") + self.assertEqual(item.feed_type, "carbon-intensity") + self.assertEqual(item.entity_id, "sensor.predbat_carbon") + self.assertEqual( + item.node_id("carbon-local"), + "reference-feed:carbon-local:Source-A", + ) + self.assertFalse(hasattr(item, "metadata")) + + def test_source_rejects_unbounded_or_non_entity_values(self): + """Free-form metadata, paths, and oversized IDs cannot enter a fragment.""" + with self.assertRaisesRegex(ValueError, "opaque identifier"): + ReferenceFeedSource("contains a secret", "feed") + with self.assertRaisesRegex(ValueError, "at most 128"): + ReferenceFeedSource("x" * 129, "feed") + with self.assertRaisesRegex(ValueError, "lowercase"): + ReferenceFeedSource("source", "not a feed") + with self.assertRaisesRegex(ValueError, "Home Assistant"): + ReferenceFeedSource("source", "feed", "not-an-entity") + with self.assertRaisesRegex(ValueError, "control characters"): + ReferenceFeedSource("source\x00secret", "feed") + with self.assertRaisesRegex(ValueError, "control characters"): + ReferenceFeedSource("source", "feed", "sensor.feed\nsecret") + + def test_digest_is_keyed_length_framed_and_strictly_bounded(self): + """Digest fields cannot exploit old delimiter ambiguity or leak inputs.""" + old_left = "\x00".join(("a", "b\x00c", "d")).encode("utf-8") + old_right = "\x00".join(("a\x00b", "c", "d")).encode("utf-8") + self.assertEqual(old_left, old_right) + self.assertNotEqual( + _length_prefixed_payload(("a", "b\x00c", "d")), + _length_prefixed_payload(("a\x00b", "c", "d")), + ) + + digest_a = provider_local_digest( + "reference-provider", + "reference-source", + "private-source-value", + PRIVACY_KEY_A, + ) + digest_b = provider_local_digest( + "reference-provider", + "reference-source", + "private-source-value", + PRIVACY_KEY_B, + ) + self.assertNotEqual(digest_a, digest_b) + self.assertNotIn("private-source-value", digest_a) + self.assertNotIn(PRIVACY_KEY_A.decode("ascii"), digest_a) + + invalid_values = ( + ("Reference-provider", "reference-source", "value", PRIVACY_KEY_A), + ("reference-provider\x00x", "reference-source", "value", PRIVACY_KEY_A), + ("x" * 129, "reference-source", "value", PRIVACY_KEY_A), + ("reference-provider", "reference\x00source", "value", PRIVACY_KEY_A), + ("reference-provider", "x" * 65, "value", PRIVACY_KEY_A), + ("reference-provider", "reference-source", "value\x00secret", PRIVACY_KEY_A), + ("reference-provider", "reference-source", "x" * 257, PRIVACY_KEY_A), + ("reference-provider", "reference-source", "value", b"short"), + ("reference-provider", "reference-source", "value", b"x" * 129), + ( + "reference-provider", + "reference-source", + "value", + "secret-key-as-text", + ), + ) + for values in invalid_values: + with self.subTest(values=values[:3]): + with self.assertRaises(ValueError) as raised: + provider_local_digest(*values) + error = str(raised.exception) + self.assertNotIn("value\x00secret", error) + self.assertNotIn(PRIVACY_KEY_A.decode("ascii"), error) + self.assertNotIn("secret-key-as-text", error) + + def test_digest_rejects_unpaired_surrogate_without_retaining_source(self): + """Non-encodable identity text fails before UTF-8 encoding.""" + raw_value = "private-digest-source\ud800" + assert_value_free_validation_error( + self, + lambda: provider_local_digest( + "reference-provider", + "reference-source", + raw_value, + PRIVACY_KEY_A, + ), + raw_value, + ) + + def test_digest_accepts_valid_non_ascii_printable_identity(self): + """Printable Unicode scalar input is valid deterministic source data.""" + raw_value = "café-家庭" + first = provider_local_digest( + "reference-provider", + "reference-source", + raw_value, + PRIVACY_KEY_A, + ) + second = provider_local_digest( + "reference-provider", + "reference-source", + raw_value, + PRIVACY_KEY_A, + ) + + self.assertEqual(first, second) + self.assertEqual(len(first), 32) + self.assertNotIn(raw_value, first) + + def test_publisher_requires_a_canonical_provider_id(self): + """Provider namespaces cannot be normalized from ambiguous input.""" + for provider_id in ( + "Reference-provider", + " reference-provider", + "reference/provider", + "reference-provider\x00other", + "x" * 129, + ): + with self.subTest(provider_id=provider_id): + with self.assertRaisesRegex(ValueError, "provider_id"): + ReferenceFeedFragmentPublisher( + provider_id, + "Reference Provider", + InMemoryFragmentAdapterStateStore(), + ) + + +class TestReferenceFeedFragmentPublisher(unittest.TestCase): + """A reference feed owns a monotonic durable immutable fragment.""" + + def test_default_off_is_unseeded_unregistered_and_write_free(self): + """Construction and disabled ingestion cannot discover or persist.""" + adapter, state_store = publisher(enabled=False) + + self.assertIsNone(adapter.lattice_fragment_adapter()) + self.assertFalse(adapter.ingest_sources(1, (source(),), health=True)) + self.assertEqual(state_store.writes, 0) + with self.assertRaises(FragmentAdapterReadError): + adapter.read_state() + + def test_snapshot_is_deterministic_reference_only_and_authority_free(self): + """Source order cannot affect a fragment or manufacture authority.""" + adapter, state_store = publisher() + sources = ( + source( + "source-b", + "flex-session", + "binary_sensor.predbat_session", + ), + source(), + ) + + self.assertTrue( + adapter.ingest_sources( + 7, + reversed(sources), + health=True, + ) + ) + snapshot = adapter.read_snapshot() + document = snapshot.topology_fragment + + self.assertEqual(adapter.source_generation, 7) + self.assertEqual(adapter.generation, 1) + self.assertEqual(state_store.writes, 1) + self.assertEqual(snapshot.health, ProviderHealth.HEALTHY) + self.assertEqual( + tuple(node["id"] for node in document["nodes"]), + ( + "reference-feed:reference-provider:source-a", + "reference-feed:reference-provider:source-b", + ), + ) + self.assertEqual(document["producer"]["authority"], 0) + self.assertTrue(all(node["kind"] == "data-source" for node in document["nodes"])) + self.assertTrue(all(node["capabilities"] == () for node in document["nodes"])) + self.assertTrue(all(alias.roles == frozenset((AliasRole.REFERENCE,)) for alias in snapshot.aliases)) + self.assertEqual(snapshot.identity_aliases, ()) + self.assertEqual(snapshot.role_assignments, ()) + self.assertEqual(snapshot.config_projections, ()) + + plan = compile_auto_config((snapshot,)) + self.assertEqual(plan.role_assignments, ()) + self.assertEqual(plan.config_arguments, ()) + self.assertFalse(plan.materialization_readiness.ready) + + def test_source_generation_replay_regression_and_conflict(self): + """Exact replay is inert and same-generation mutation fails closed.""" + adapter, state_store = publisher() + initial = source() + adapter.ingest_sources(4, (initial,), health=True) + + self.assertFalse(adapter.ingest_sources(4, (initial,))) + self.assertFalse(adapter.ingest_sources(3, (initial,))) + with self.assertRaisesRegex( + FragmentAdapterConflict, + "reused reference-feed source generation 4", + ): + adapter.ingest_sources( + 4, + ( + source( + entity_id="sensor.predbat_changed_reference", + ), + ), + ) + self.assertEqual(adapter.generation, 1) + self.assertEqual(state_store.writes, 1) + + self.assertTrue(adapter.ingest_sources(5, (initial,))) + self.assertEqual(adapter.source_generation, 5) + self.assertEqual(adapter.generation, 2) + + def test_liveness_advances_without_changing_source_generation(self): + """Health changes independently and exact replay is inert.""" + adapter, state_store = publisher() + adapter.ingest_sources(9, (source(),), health=True) + invalidations = [] + adapter.subscribe_invalidation( + lambda provider_id, generation, reason, feedback_token: ( + invalidations.append( + ( + provider_id, + generation, + reason, + feedback_token, + ) + ) + ) + ) + + self.assertTrue(adapter.set_liveness(False)) + self.assertFalse(adapter.set_liveness(False)) + self.assertEqual(adapter.generation, 2) + self.assertEqual(adapter.source_generation, 9) + self.assertEqual( + adapter.read_snapshot().health, + ProviderHealth.OFFLINE, + ) + self.assertEqual(state_store.writes, 2) + self.assertEqual(invalidations[-1][0:2], ("reference-provider", 2)) + + def test_compiler_invalidation_and_feedback_suppression(self): + """Accepted changes recompile; publication feedback persists nothing.""" + adapter, state_store = publisher() + adapter.ingest_sources(1, (source(),), health=True) + registry = FragmentAdapterRegistry(enabled=True) + self.assertEqual( + registry.discover((adapter,)), + ("reference-provider",), + ) + compiler = registry.create_compiler( + InMemoryCompiledLatticeStateStore(), + ) + first = compiler.drain() + self.assertEqual(first.status, CompileStatus.FRESH) + + self.assertTrue( + adapter.ingest_sources( + 2, + ( + source( + entity_id="sensor.predbat_reference_data_v2", + ), + ), + ) + ) + second = compiler.drain() + self.assertEqual(second.status, CompileStatus.FRESH) + self.assertEqual( + dict(second.publication.provider_generations), + {"reference-provider": 2}, + ) + + before = adapter.read_state() + writes = state_store.writes + self.assertFalse( + adapter.ingest_sources( + 3, + ( + source( + entity_id="sensor.predbat_reference_data_v3", + ), + ), + feedback_token=second.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): + """Ambient replay cannot resurrect a removed reference source.""" + adapter, state_store = publisher() + adapter.ingest_sources(5, (source(),), 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.generation, 2) + self.assertEqual( + removed.snapshot.health, + ProviderHealth.OFFLINE, + ) + with self.assertRaises(FragmentAdapterRemoved): + adapter.read_snapshot() + with self.assertRaisesRegex( + FragmentAdapterRemoved, + "cannot re-enrol", + ): + adapter.ingest_sources(99, (source(),), health=True) + self.assertEqual( + invalidations[-1], + ( + "reference-provider", + 2, + "reference feed integration removed", + ), + ) + + restarted = ReferenceFeedFragmentPublisher( + "reference-provider", + "Reference Provider", + state_store, + enabled=True, + ) + self.assertTrue(restarted.read_state().removed) + with self.assertRaises(FragmentAdapterRemoved): + restarted.read_snapshot() + with self.assertRaises(FragmentAdapterRemoved): + restarted.ingest_sources(100, (source(),)) + + +if __name__ == "__main__": + unittest.main()