diff --git a/apps/predbat/lattice_gateway_fragment.py b/apps/predbat/lattice_gateway_fragment.py new file mode 100644 index 000000000..21d4b0712 --- /dev/null +++ b/apps/predbat/lattice_gateway_fragment.py @@ -0,0 +1,401 @@ +# ----------------------------------------------------------------------------- +# Predbat Home Battery System - Gateway retained-topology Lattice adapter +# Copyright Trefor Southwell 2026 - All Rights Reserved +# This application maybe used for personal use only and not for commercial use +# ----------------------------------------------------------------------------- +"""Pure, default-off Gateway retained-topology fragment publisher. + +The live Gateway component already receives a provider-owned retained topology +document and a separate liveness signal. This module adapts those two inputs +to the common ``lattice_fragment_adapter()`` surface without registering +itself, mutating PredBat configuration, or publishing control. + +Gateway retained topology remains on the existing v0.2/v0.3 document surface +in this tranche. The v0.4 schedule transport is independent; retained v0.4 +topology is rejected until the shared topology compiler is promoted explicitly. + +Gateway aliases are deliberately REFERENCE-only. The adapter exposes +provider-local capability projection metadata for inspection and future +composition, but publishes no PRIMARY/CONTROL role assignment and no +configuration projection ready for materialization. +""" + +# cspell:ignore autoconfig + +import threading +from dataclasses import dataclass +from typing import Optional + +from lattice_autoconfig import ( + AliasRole, + ProviderAlias, + ProviderHealth, + ProviderIdentityAlias, + ProviderSnapshot, + _plain, +) +from lattice_fragment_adapters import ( + DurableFragmentAdapter, + FragmentAdapterReadError, + FragmentAdapterRemoved, +) +from lattice_topology import TopologyValidationError, decode_topology + + +_LIVENESS_UNCHANGED = object() + + +@dataclass(frozen=True) +class GatewayCapabilityProjection: + """One provider-local capability coordinate from the retained document.""" + + node_id: str + capability: str + access_path_id: str + cap_ref: Optional[int] + readable: bool + controllable: bool + + +@dataclass(frozen=True) +class GatewayTopologyProjection: + """Immutable inspection metadata for one durable Gateway fragment.""" + + provider_id: str + generation: int + semantic_fingerprint: str + health: ProviderHealth + topology_version: str + document_version: int + node_ids: tuple + capabilities: tuple + removed: bool + + +def _document_version(document): + """Return an explicit version or the legacy v0.2 retained-doc default.""" + value = document.get("docVersion") + if value is None: + return 0 + return value + + +def _health_from_liveness(online): + """Map the independent Gateway liveness signal to provider health.""" + if online is True: + return ProviderHealth.HEALTHY + if online is False: + return ProviderHealth.OFFLINE + if online is None: + return ProviderHealth.DEGRADED + raise ValueError("online must be True, False, or None") + + +def _live_nodes(document): + """Return provider nodes that are not topology tombstones.""" + return tuple(node for node in document.get("nodes", ()) if node.get("removed") is not True) + + +def _reference_aliases(document): + """Build deterministic REFERENCE-only aliases for every provider node.""" + return tuple( + ProviderAlias( + name="node:{}".format(node_id), + node_id=node_id, + roles=frozenset((AliasRole.REFERENCE,)), + ) + for node_id in sorted(str(node["id"]) for node in _live_nodes(document)) + ) + + +def _identity_aliases(document): + """Publish strong identities asserted by the provider-owned fragment.""" + aliases = [] + for node in _live_nodes(document): + node_id = str(node["id"]) + aliases.append( + ProviderIdentityAlias( + kind="lattice-node-id", + value=node_id, + node_id=node_id, + ) + ) + attributes = node.get("attributes") + serial = attributes.get("serial") if isinstance(attributes, dict) else None + if isinstance(serial, str) and serial.strip(): + aliases.append( + ProviderIdentityAlias( + kind="serial", + value=serial.strip(), + node_id=node_id, + ) + ) + return tuple( + sorted( + aliases, + key=lambda item: (item.kind, item.value, item.node_id), + ) + ) + + +def _valid_binding(binding): + """Return whether a read/control binding has usable protocol coordinates.""" + if not isinstance(binding, dict): + return False + protocol = binding.get("protocol") + if not isinstance(protocol, str) or not protocol.strip(): + return False + if "address" not in binding: + return False + address = binding["address"] + if isinstance(address, bool): + return False + if isinstance(address, str): + return bool(address.strip()) + return isinstance(address, (int, dict)) + + +def _capability_projections(document): + """Extract immutable provider-local refs without inferring brand behavior.""" + projections = [] + for node in _live_nodes(document): + node_id = str(node["id"]) + for offer in node.get("capabilities", ()): + if not isinstance(offer, dict) or offer.get("removed") is True: + continue + capability = offer.get("capability") + if capability is None: + continue + cap_ref = offer.get("ref") + if not isinstance(cap_ref, int) or isinstance(cap_ref, bool) or cap_ref <= 0: + cap_ref = None + projections.append( + GatewayCapabilityProjection( + node_id=node_id, + capability=str(capability), + access_path_id=str(offer.get("accessPath", "")), + cap_ref=cap_ref, + readable=_valid_binding(offer.get("read")) or isinstance(offer.get("derived"), dict), + controllable=_valid_binding(offer.get("control")), + ) + ) + return tuple( + sorted( + projections, + key=lambda item: ( + item.node_id, + item.capability, + item.access_path_id, + item.cap_ref or 0, + ), + ) + ) + + +def _validate_fragment(document, provider_id): + """Require one provider-local fragment owned by this exact adapter.""" + if document.get("scope") != "fragment": + raise TopologyValidationError("Gateway retained topology scope must be fragment") + document_provider = document["producer"]["provider"] + if document_provider != provider_id: + raise TopologyValidationError( + "Gateway adapter {} cannot publish fragment owned by {}".format( + provider_id, + document_provider, + ) + ) + + +class GatewayRetainedTopologyFragmentPublisher: + """Adapt retained topology and liveness into a durable fragment publisher.""" + + def __init__(self, provider_id, 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 + 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 reference publisher runs.""" + 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 projection_metadata(self): + """Return immutable provider-local projection metadata.""" + state = self.read_state() + document = _plain(state.snapshot.topology_fragment) + return GatewayTopologyProjection( + provider_id=state.provider_id, + generation=state.generation, + semantic_fingerprint=state.semantic_fingerprint, + health=state.snapshot.health, + topology_version=document["topologyVersion"], + document_version=_document_version(document), + node_ids=tuple(sorted(str(node["id"]) for node in _live_nodes(document))), + capabilities=_capability_projections(document), + removed=state.removed, + ) + + def lattice_fragment_adapter(self): + """Expose the structural registry surface 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 topology, liveness, and removal invalidations.""" + return self._adapter.subscribe_invalidation(listener) + + def _current_state(self): + """Return the current state, treating an unseeded store as empty.""" + if not self._seeded: + return None + return self._adapter.read_state() + + def ingest_retained_topology(self, payload, online=_LIVENESS_UNCHANGED): + """Publish one accepted retained fragment and invalidate subscribers.""" + if not self._enabled: + return False + document = decode_topology(payload) + _validate_fragment(document, self.provider_id) + + with self._lock: + current = self._current_state() + if online is _LIVENESS_UNCHANGED and current is not None: + health = current.snapshot.health + elif online is _LIVENESS_UNCHANGED: + health = ProviderHealth.DEGRADED + else: + health = _health_from_liveness(online) + if current is not None and current.removed: + raise FragmentAdapterRemoved( + "provider {} was removed at generation {}; retained topology " + "cannot re-enrol it".format( + self.provider_id, + current.generation, + ) + ) + if current is not None: + previous = _plain(current.snapshot.topology_fragment) + previous_explicit = "docVersion" in previous + incoming_explicit = "docVersion" in document + previous_version = _document_version(previous) + incoming_version = _document_version(document) + if previous_explicit and not incoming_explicit: + return False + if previous_explicit and incoming_explicit: + if incoming_version < previous_version: + return False + if incoming_version == previous_version: + if document != previous: + raise TopologyValidationError( + "provider {} reused docVersion {} for different " + "content".format( + self.provider_id, + incoming_version, + ) + ) + if health is current.snapshot.health: + return False + elif document == previous and 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=health, + topology_fragment=document, + aliases=_reference_aliases(document), + identity_aliases=_identity_aliases(document), + role_assignments=(), + config_projections=(), + ) + published = self._adapter.publish( + snapshot, + "gateway retained topology changed", + ) + if published: + self._seeded = True + return published + + def set_liveness(self, online): + """Publish a liveness-only generation when provider health changes.""" + if not self._enabled: + return False + health = _health_from_liveness(online) + with self._lock: + current = self._current_state() + if current is None: + raise FragmentAdapterReadError("Gateway topology 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, + "gateway liveness changed to {}".format(health.value), + ) + + def remove(self): + """Publish a durable provider-removal tombstone and invalidate.""" + if not self._enabled: + return False + with self._lock: + current = self._current_state() + if current is None: + raise FragmentAdapterReadError("Gateway topology must be seeded before removal") + if current.removed: + return False + return self._adapter.remove( + current.generation + 1, + "gateway integration removed", + ) diff --git a/apps/predbat/tests/test_lattice_gateway_fragment.py b/apps/predbat/tests/test_lattice_gateway_fragment.py new file mode 100644 index 000000000..553ef3d41 --- /dev/null +++ b/apps/predbat/tests/test_lattice_gateway_fragment.py @@ -0,0 +1,450 @@ +"""Tests for the pure Gateway retained-topology Lattice fragment 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 + FragmentAdapterReadError, + FragmentAdapterRegistry, + FragmentAdapterRemoved, + InMemoryFragmentAdapterStateStore, +) +from lattice_gateway_fragment import ( # noqa: E402 + GatewayCapabilityProjection, + GatewayRetainedTopologyFragmentPublisher, + GatewayTopologyProjection, +) +from lattice_topology import TopologyValidationError # noqa: E402 + + +def topology( + doc_version=1, + provider="predbat-gateway", + node_id="GW-INV-1", + capability="battery.target_soc", + online_binding=True, +): + """Build one compact provider-owned retained Gateway fragment.""" + offer = { + "capability": capability, + "accessPath": "gateway-mqtt", + "ref": 41, + "shape": "setpoint", + "read": {"protocol": "mqtt", "address": "state"}, + } + if online_binding: + offer["control"] = {"protocol": "mqtt", "address": "setpoint"} + return { + "topologyVersion": "0.3.0", + "scope": "fragment", + "docVersion": doc_version, + "producer": { + "name": "PredBat Gateway", + "provider": provider, + "authority": 10, + }, + "nodes": [ + { + "id": node_id, + "kind": "inverter", + "attributes": {"serial": node_id}, + "accessPaths": [ + { + "id": "gateway-mqtt", + "provider": provider, + "preference": 10, + } + ], + "capabilities": [offer], + } + ], + } + + +def publisher(enabled=True, state_store=None): + """Build one reference publisher and its in-memory durable store.""" + if state_store is None: + state_store = InMemoryFragmentAdapterStateStore() + return ( + GatewayRetainedTopologyFragmentPublisher( + "predbat-gateway", + state_store, + enabled=enabled, + ), + state_store, + ) + + +class TestGatewayRetainedTopologyFragmentPublisher(unittest.TestCase): + """Gateway inputs publish immutable, REFERENCE-only provider snapshots.""" + + def test_default_off_has_no_discovery_or_durable_side_effect(self): + """Disabled construction never parses, publishes, or registers.""" + adapter, state_store = publisher(enabled=False) + + self.assertIsNone(adapter.lattice_fragment_adapter()) + self.assertFalse(adapter.ingest_retained_topology(b"{not json")) + 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_seeded_store_failure_is_not_hidden_as_absent_discovery(self): + """A registered provider read fault fails closed instead of disappearing.""" + + class FailingAfterSeedStore(InMemoryFragmentAdapterStateStore): + """Make durable reads fail only after a valid seed.""" + + def __init__(self): + super().__init__() + self.fail_reads = False + + def load(self): + if self.fail_reads: + raise OSError("durable Gateway fragment unavailable") + return super().load() + + state_store = FailingAfterSeedStore() + adapter, _state_store = publisher(state_store=state_store) + 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", + ): + adapter.set_liveness(False) + + def test_retained_topology_publishes_immutable_reference_snapshot(self): + """A retained fragment becomes detached provider-local metadata.""" + adapter, state_store = publisher() + document = topology() + + self.assertTrue(adapter.ingest_retained_topology(document, online=True)) + document["nodes"][0]["id"] = "MUTATED" + snapshot = adapter.read_snapshot() + + self.assertIs(adapter.lattice_fragment_adapter(), adapter) + self.assertEqual(state_store.writes, 1) + self.assertEqual(adapter.generation, 1) + self.assertEqual(len(adapter.semantic_fingerprint), 64) + self.assertEqual(snapshot.health, ProviderHealth.HEALTHY) + self.assertEqual(snapshot.topology_fragment["nodes"][0]["id"], "GW-INV-1") + self.assertEqual(snapshot.role_assignments, ()) + self.assertEqual(snapshot.config_projections, ()) + self.assertEqual(len(snapshot.aliases), 1) + self.assertEqual(snapshot.aliases[0].roles, frozenset((AliasRole.REFERENCE,))) + self.assertEqual( + tuple((alias.kind, alias.value) for alias in snapshot.identity_aliases), + ( + ("lattice-node-id", "GW-INV-1"), + ("serial", "GW-INV-1"), + ), + ) + + def test_projection_metadata_retains_provider_local_capability_coordinates(self): + """Projection inspection exposes refs but no materialization authority.""" + adapter, _state_store = publisher() + adapter.ingest_retained_topology(topology(), online=None) + + projection = adapter.projection_metadata + + self.assertIsInstance(projection, GatewayTopologyProjection) + self.assertEqual(projection.provider_id, "predbat-gateway") + self.assertEqual(projection.health, ProviderHealth.DEGRADED) + self.assertEqual(projection.document_version, 1) + self.assertEqual(projection.node_ids, ("GW-INV-1",)) + self.assertEqual( + projection.capabilities, + ( + GatewayCapabilityProjection( + node_id="GW-INV-1", + capability="battery.target_soc", + access_path_id="gateway-mqtt", + cap_ref=41, + readable=True, + controllable=True, + ), + ), + ) + + def test_projection_metadata_requires_valid_bindings(self): + """Malformed binding keys do not claim readable or controllable paths.""" + adapter, _state_store = publisher() + document = topology() + offer = document["nodes"][0]["capabilities"][0] + offer["read"] = None + offer["control"] = {"protocol": "mqtt"} + + adapter.ingest_retained_topology(document, online=True) + + self.assertEqual( + adapter.projection_metadata.capabilities, + ( + GatewayCapabilityProjection( + node_id="GW-INV-1", + capability="battery.target_soc", + access_path_id="gateway-mqtt", + cap_ref=41, + readable=False, + controllable=False, + ), + ), + ) + + def test_removed_nodes_publish_no_reference_or_projection_metadata(self): + """A node tombstone cannot survive as an alias, identity, or capability.""" + adapter, _state_store = publisher() + document = topology() + document["nodes"][0]["removed"] = True + + adapter.ingest_retained_topology(document, online=True) + snapshot = adapter.read_snapshot() + projection = adapter.projection_metadata + + self.assertEqual(snapshot.aliases, ()) + self.assertEqual(snapshot.identity_aliases, ()) + self.assertEqual(projection.node_ids, ()) + self.assertEqual(projection.capabilities, ()) + + def test_serial_and_stable_node_id_correlate_mixed_provider_views(self): + """A provider with only the stable node id still joins a serial view.""" + gateway, _gateway_store = publisher() + gateway.ingest_retained_topology(topology(), online=True) + cloud_document = topology(provider="cloud") + del cloud_document["nodes"][0]["attributes"]["serial"] + cloud = GatewayRetainedTopologyFragmentPublisher( + "cloud", + InMemoryFragmentAdapterStateStore(), + enabled=True, + ) + cloud.ingest_retained_topology(cloud_document, online=True) + + plan = compile_auto_config( + ( + gateway.read_snapshot(), + cloud.read_snapshot(), + ) + ) + + self.assertEqual(len(plan.topology["nodes"]), 1) + self.assertEqual(plan.topology["nodes"][0]["id"], "identity:serial:GW-INV-1") + + def test_retained_topology_version_matrix_is_explicit(self): + """Retained topology stays on v0.2/v0.3 while v0.4 remains separate.""" + for version in ("0.2.0", "0.3.0"): + with self.subTest(version=version): + adapter, _state_store = publisher() + document = topology() + document["topologyVersion"] = version + if version == "0.2.0": + del document["docVersion"] + self.assertTrue(adapter.ingest_retained_topology(document)) + + adapter, _state_store = publisher() + document = topology() + document["topologyVersion"] = "0.4.0" + with self.assertRaisesRegex( + TopologyValidationError, + "unsupported topologyVersion '0.4.0'", + ): + adapter.ingest_retained_topology(document) + + def test_topology_and_liveness_changes_invalidate_for_fresh_recompile(self): + """Every accepted input change notifies the common registry compiler.""" + adapter, _state_store = publisher() + adapter.ingest_retained_topology(topology(), online=True) + registry = FragmentAdapterRegistry(enabled=True) + self.assertEqual(registry.discover((adapter,)), ("predbat-gateway",)) + compiler = registry.create_compiler( + InMemoryCompiledLatticeStateStore(), + ) + first = compiler.drain() + self.assertEqual(first.status, CompileStatus.FRESH) + plan = first.publication.plan + self.assertFalse(plan.materialization_readiness.ready) + self.assertEqual(plan.role_assignments, ()) + self.assertEqual(plan.primary_targets, ()) + self.assertEqual(plan.control_targets, ()) + self.assertEqual(plan.config_arguments, ()) + self.assertEqual(dict(plan.projected_config), {}) + + self.assertTrue(adapter.set_liveness(False)) + offline = compiler.drain() + self.assertEqual(offline.status, CompileStatus.STALE) + self.assertTrue(offline.pending) + self.assertEqual(offline.publication, first.publication) + self.assertEqual(adapter.generation, 2) + + self.assertTrue( + adapter.ingest_retained_topology( + topology(doc_version=2, capability="battery.charge_power_limit"), + online=True, + ) + ) + changed = compiler.drain() + self.assertEqual(changed.status, CompileStatus.FRESH) + self.assertEqual( + changed.publication.provider_generations, + (("predbat-gateway", 3),), + ) + + def test_duplicate_stale_and_reused_versions_do_not_recompile(self): + """Retained replay is idempotent and explicit generations fail closed.""" + adapter, state_store = publisher() + reasons = [] + adapter.subscribe_invalidation(lambda provider_id, generation, reason, feedback_token: reasons.append((provider_id, generation, reason, feedback_token))) + first = topology(doc_version=5) + adapter.ingest_retained_topology(first, online=True) + + self.assertFalse(adapter.ingest_retained_topology(first, online=True)) + self.assertFalse( + adapter.ingest_retained_topology( + topology(doc_version=4), + online=True, + ) + ) + with self.assertRaisesRegex(TopologyValidationError, "reused docVersion"): + adapter.ingest_retained_topology( + topology(doc_version=5, capability="battery.charge_power_limit"), + online=True, + ) + self.assertEqual(adapter.generation, 1) + self.assertEqual(state_store.writes, 1) + self.assertEqual(len(reasons), 1) + + def test_legacy_retained_changes_use_adapter_owned_monotonic_generation(self): + """Unversioned v0.2 arrival changes remain monotonic and replay-safe.""" + adapter, _state_store = publisher() + first = topology(doc_version=1) + second = topology(doc_version=1, capability="battery.charge_power_limit") + for document in (first, second): + document["topologyVersion"] = "0.2.0" + del document["docVersion"] + + self.assertTrue(adapter.ingest_retained_topology(first, online=True)) + self.assertFalse(adapter.ingest_retained_topology(first, online=True)) + self.assertTrue(adapter.ingest_retained_topology(second, online=True)) + self.assertEqual(adapter.generation, 2) + self.assertEqual(adapter.projection_metadata.document_version, 0) + + def test_invalid_or_foreign_fragment_preserves_durable_state(self): + """Malformed ownership and site documents cannot replace good state.""" + adapter, state_store = publisher() + adapter.ingest_retained_topology(topology(), online=True) + before = adapter.read_state() + + with self.assertRaisesRegex(TopologyValidationError, "owned by cloud"): + adapter.ingest_retained_topology( + topology(doc_version=2, provider="cloud"), + online=True, + ) + site = topology(doc_version=2) + site["scope"] = "site" + with self.assertRaisesRegex(TopologyValidationError, "scope"): + adapter.ingest_retained_topology(site, online=True) + + self.assertEqual(adapter.read_state(), before) + self.assertEqual(state_store.writes, 1) + + def test_liveness_is_independent_idempotent_and_requires_topology(self): + """LWT health changes preserve exact topology and skip duplicates.""" + unseeded, _store = publisher() + with self.assertRaisesRegex(FragmentAdapterReadError, "seeded"): + unseeded.set_liveness(True) + + adapter, state_store = publisher() + adapter.ingest_retained_topology(topology(), online=None) + before_fragment = adapter.read_snapshot().topology_fragment + + self.assertTrue(adapter.set_liveness(True)) + self.assertFalse(adapter.set_liveness(True)) + self.assertTrue(adapter.set_liveness(False)) + self.assertEqual(adapter.read_snapshot().health, ProviderHealth.OFFLINE) + self.assertEqual(adapter.read_snapshot().topology_fragment, before_fragment) + self.assertEqual(adapter.generation, 3) + self.assertEqual(state_store.writes, 3) + + def test_topology_change_without_liveness_preserves_current_health(self): + """An independent retained update does not invent a liveness change.""" + adapter, _state_store = publisher() + adapter.ingest_retained_topology(topology(), online=True) + + adapter.ingest_retained_topology( + topology( + doc_version=2, + capability="battery.charge_power_limit", + ) + ) + + self.assertEqual(adapter.read_snapshot().health, ProviderHealth.HEALTHY) + self.assertEqual(adapter.generation, 2) + + def test_removal_is_durable_invalidating_and_restart_visible(self): + """Integration removal publishes a tombstone rather than disappearing.""" + adapter, state_store = publisher() + adapter.ingest_retained_topology(topology(), online=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()) + self.assertTrue(adapter.read_state().removed) + self.assertTrue(adapter.projection_metadata.removed) + with self.assertRaises(FragmentAdapterRemoved): + adapter.read_snapshot() + self.assertEqual( + invalidations[-1], + ("predbat-gateway", 2, "gateway integration removed"), + ) + + restarted = GatewayRetainedTopologyFragmentPublisher( + "predbat-gateway", + state_store, + enabled=True, + ) + self.assertTrue(restarted.read_state().removed) + self.assertTrue(restarted.projection_metadata.removed) + with self.assertRaises(FragmentAdapterRemoved): + restarted.read_snapshot() + + def test_removal_cannot_be_reversed_by_retained_replay(self): + """Ambient retained topology cannot re-enrol a removed integration.""" + adapter, state_store = publisher() + adapter.ingest_retained_topology(topology(doc_version=5), online=True) + adapter.remove() + before = adapter.read_state() + + with self.assertRaisesRegex( + FragmentAdapterRemoved, + "retained topology cannot re-enrol", + ): + adapter.ingest_retained_topology(topology(doc_version=1), online=True) + + self.assertEqual(adapter.read_state(), before) + self.assertTrue(adapter.read_state().removed) + self.assertEqual(state_store.writes, 2) + + +if __name__ == "__main__": + unittest.main()