diff --git a/apps/predbat/lattice_fox_fragment.py b/apps/predbat/lattice_fox_fragment.py new file mode 100644 index 000000000..b10e8847a --- /dev/null +++ b/apps/predbat/lattice_fox_fragment.py @@ -0,0 +1,564 @@ +# ----------------------------------------------------------------------------- +# Predbat Home Battery System - Fox Cloud 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 Fox Cloud fragment publisher. + +The live ``FoxAPI`` component owns OAuth/API-key authentication, discovery, +polling, scheduler state, automatic configuration, and control writes. This +module deliberately does not import, register, or mutate that high-impact +component. A later gated seam may pass explicit immutable discovery snapshots +here after a successful cloud read. + +Accepted discovery, liveness, and removal changes advance the common durable +fragment generation and invalidate the fresh-read compiler. This tranche is +REFERENCE-only: it publishes topology and identity, but no roles, +configuration projections, capabilities, or write authority. +""" + +# 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, + FragmentAdapterConflict, + FragmentAdapterReadError, + FragmentAdapterRemoved, +) + + +_HEALTH_UNCHANGED = object() + + +def _required_text(value, name): + """Return one normalized non-empty text field.""" + if not isinstance(value, str) or not value.strip(): + raise ValueError("{} must be a non-empty string".format(name)) + return value.strip() + + +def _optional_text(value, name): + """Return one normalized optional text field.""" + if value is None: + return None + return _required_text(value, name) + + +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_version(value): + """Validate one provider-owned monotonic discovery version.""" + if not isinstance(value, int) or isinstance(value, bool) or value < 0: + raise ValueError("discovery_version 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 FoxStationSnapshot: + """One explicit provider-local Fox station.""" + + station_id: str + name: Optional[str] = None + + def __post_init__(self): + """Normalize the provider-owned station identity.""" + object.__setattr__( + self, + "station_id", + _required_text(self.station_id, "station_id"), + ) + object.__setattr__( + self, + "name", + _optional_text(self.name, "name"), + ) + + @property + def node_id(self): + """Return the deterministic provider-local station node.""" + return "fox:station:{}".format(self.station_id) + + +@dataclass(frozen=True) +class FoxDeviceSnapshot: + """One explicit provider-local Fox inverter/device.""" + + serial: str + station_id: str + device_type: Optional[str] = None + online: Optional[bool] = None + has_battery: Optional[bool] = None + has_pv: Optional[bool] = None + third_party_generation: Optional[bool] = None + scheduler_supported: Optional[bool] = None + + def __post_init__(self): + """Normalize strong hardware identity and explicit feature evidence.""" + object.__setattr__( + self, + "serial", + _required_text(self.serial, "serial").upper(), + ) + object.__setattr__( + self, + "station_id", + _required_text(self.station_id, "station_id"), + ) + object.__setattr__( + self, + "device_type", + _optional_text(self.device_type, "device_type"), + ) + object.__setattr__( + self, + "online", + _optional_bool(self.online, "online"), + ) + object.__setattr__( + self, + "has_battery", + _optional_bool(self.has_battery, "has_battery"), + ) + object.__setattr__( + self, + "has_pv", + _optional_bool(self.has_pv, "has_pv"), + ) + object.__setattr__( + self, + "third_party_generation", + _optional_bool( + self.third_party_generation, + "third_party_generation", + ), + ) + object.__setattr__( + self, + "scheduler_supported", + _optional_bool( + self.scheduler_supported, + "scheduler_supported", + ), + ) + + @property + def node_id(self): + """Return the deterministic provider-local hardware node.""" + return "fox:device:{}".format(self.serial) + + +def _normalize_stations(stations): + """Freeze, sort, and collision-check station snapshots.""" + try: + stations = tuple(stations) + except TypeError as exc: + raise ValueError( + "stations must be an iterable of FoxStationSnapshot", + ) from exc + if any(not isinstance(item, FoxStationSnapshot) for item in stations): + raise ValueError("stations must contain only FoxStationSnapshot values") + if not stations: + raise ValueError("stations must contain at least one FoxStationSnapshot") + + seen = set() + for station in stations: + if station.station_id in seen: + raise FragmentAdapterConflict( + "Fox discovery contains duplicate station {}".format( + station.station_id, + ) + ) + seen.add(station.station_id) + return tuple(sorted(stations, key=lambda item: item.station_id)) + + +def _normalize_devices(devices, station_ids): + """Freeze, sort, and reject ambiguous hardware ownership.""" + try: + devices = tuple(devices) + except TypeError as exc: + raise ValueError( + "devices must be an iterable of FoxDeviceSnapshot", + ) from exc + if any(not isinstance(item, FoxDeviceSnapshot) for item in devices): + raise ValueError("devices must contain only FoxDeviceSnapshot values") + if not devices: + raise ValueError("devices must contain at least one FoxDeviceSnapshot") + + serial_owners = {} + for device in devices: + if device.station_id not in station_ids: + raise ValueError( + "device {} references unknown station {}".format( + device.serial, + device.station_id, + ) + ) + owner = serial_owners.get(device.serial) + if owner is not None: + raise FragmentAdapterConflict( + "Fox serial {} belongs to both stations {} and {}".format( + device.serial, + owner, + device.station_id, + ) + ) + serial_owners[device.serial] = device.station_id + return tuple( + sorted( + devices, + key=lambda item: (item.station_id, item.serial), + ) + ) + + +def _station_node(station, provider_id): + """Project one provider-local site without claiming authority.""" + attributes = {"stationId": station.station_id} + if station.name is not None: + attributes["stationName"] = station.name + return { + "id": station.node_id, + "kind": "site", + "attributes": attributes, + "accessPaths": [ + { + "id": "fox-cloud-api", + "provider": provider_id, + "preference": 0, + } + ], + "capabilities": [], + } + + +def _device_node(device, provider_id): + """Project explicit Fox observations without deriving capabilities.""" + attributes = { + "deviceSn": device.serial, + "stationId": device.station_id, + } + optional = ( + ("deviceType", device.device_type), + ("online", device.online), + ("hasBattery", device.has_battery), + ("hasPv", device.has_pv), + ("thirdPartyGeneration", device.third_party_generation), + ("schedulerSupported", device.scheduler_supported), + ) + for key, value in optional: + if value is not None: + attributes[key] = value + return { + "id": device.node_id, + "kind": "inverter", + "attributes": attributes, + "accessPaths": [ + { + "id": "fox-cloud-api", + "provider": provider_id, + "preference": 0, + } + ], + "capabilities": [], + } + + +def _topology_document(provider_id, discovery_version, stations, devices): + """Build one deterministic provider-owned v0.3 fragment.""" + nodes = [_station_node(station, provider_id) for station in stations] + nodes.extend(_device_node(device, provider_id) for device in devices) + relationships = [ + { + "from": "fox:station:{}".format(device.station_id), + "to": device.node_id, + "type": "contains", + } + for device in devices + ] + return { + "topologyVersion": "0.3.0", + "scope": "fragment", + "docVersion": discovery_version, + "producer": { + "name": "Fox Cloud", + "provider": provider_id, + "authority": 0, + }, + "nodes": nodes, + "relationships": relationships, + } + + +def _reference_aliases(stations, devices): + """Publish provider-local names with REFERENCE role only.""" + aliases = [ + ProviderAlias( + name="station:{}".format(station.station_id), + node_id=station.node_id, + roles=frozenset((AliasRole.REFERENCE,)), + ) + for station in stations + ] + aliases.extend( + ProviderAlias( + name="serial:{}".format(device.serial), + node_id=device.node_id, + roles=frozenset((AliasRole.REFERENCE,)), + ) + for device in devices + ) + return tuple(sorted(aliases, key=lambda item: (item.name, item.node_id))) + + +def _identity_aliases(stations, devices): + """Publish provider station IDs and globally useful hardware serials.""" + aliases = [ + ProviderIdentityAlias( + kind="fox-station-id", + value=station.station_id, + node_id=station.node_id, + ) + for station in stations + ] + aliases.extend( + ProviderIdentityAlias( + kind="serial", + value=device.serial, + node_id=device.node_id, + ) + for device in devices + ) + return tuple( + sorted( + aliases, + key=lambda item: (item.kind, item.value, item.node_id), + ) + ) + + +class FoxCloudFragmentPublisher: + """Adapt explicit Fox snapshots into one durable compiler fragment.""" + + 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 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_version(self): + """Fresh-read the provider-owned discovery version.""" + 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 discovery, 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_discovery( + self, + discovery_version, + stations, + devices, + health=_HEALTH_UNCHANGED, + ): + """Publish one accepted complete station/device snapshot.""" + if not self._enabled: + return False + discovery_version = _discovery_version(discovery_version) + stations = _normalize_stations(stations) + devices = _normalize_devices( + devices, + frozenset(item.station_id for item in stations), + ) + document = _topology_document( + self.provider_id, + discovery_version, + stations, + devices, + ) + + with self._lock: + current = self._current_state() + if current is not None and current.removed: + raise FragmentAdapterRemoved( + "provider {} was removed at generation {}; Fox " + "discovery 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_version = previous["docVersion"] + if discovery_version < previous_version: + return False + if discovery_version == previous_version and document != previous: + raise FragmentAdapterConflict( + "provider {} reused Fox discovery version {} " + "for different content".format( + self.provider_id, + discovery_version, + ) + ) + next_health = current.snapshot.health if health is _HEALTH_UNCHANGED else _provider_health(health) + if discovery_version == previous_version 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(stations, devices), + identity_aliases=_identity_aliases(stations, devices), + role_assignments=(), + config_projections=(), + ) + published = self._adapter.publish( + snapshot, + "Fox discovery changed to version {}".format( + discovery_version, + ), + ) + if published: + self._seeded = True + return published + + def set_liveness(self, health): + """Publish liveness without changing discovery contents.""" + if not self._enabled: + return False + health = _provider_health(health) + with self._lock: + current = self._current_state() + if current is None: + raise FragmentAdapterReadError( + "Fox discovery 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, + "Fox liveness changed to {}".format(health.value), + ) + + def remove(self): + """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( + "Fox discovery must be seeded before removal", + ) + if current.removed: + return False + return self._adapter.remove( + current.generation + 1, + "Fox integration removed", + ) diff --git a/apps/predbat/tests/test_lattice_fox_fragment.py b/apps/predbat/tests/test_lattice_fox_fragment.py new file mode 100644 index 000000000..12aa76ce4 --- /dev/null +++ b/apps/predbat/tests/test_lattice_fox_fragment.py @@ -0,0 +1,433 @@ +"""Tests for the pure Fox Cloud Lattice fragment publisher.""" + +# cspell:ignore autoconfig invalidatable + +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, + CompileStatus, + ProviderHealth, + compile_auto_config, +) +from lattice_compiled_publication import ( # noqa: E402 + InMemoryCompiledLatticeStateStore, +) +from lattice_fox_fragment import ( # noqa: E402 + FoxCloudFragmentPublisher, + FoxDeviceSnapshot, + FoxStationSnapshot, +) +from lattice_fragment_adapters import ( # noqa: E402 + FragmentAdapterConflict, + FragmentAdapterReadError, + FragmentAdapterRegistry, + FragmentAdapterRemoved, + InMemoryFragmentAdapterStateStore, +) +from lattice_ge_cloud_fragment import ( # noqa: E402 + GECloudDeviceSnapshot, + GECloudFragmentPublisher, +) + + +def station(station_id="SITE-1", name="Home"): + """Build one explicit Fox station snapshot.""" + return FoxStationSnapshot(station_id=station_id, name=name) + + +def device( + serial="60KE8020479C034", + station_id="SITE-1", + device_type="H3", + online=True, + has_battery=True, + has_pv=True, + third_party_generation=False, + scheduler_supported=True, +): + """Build one explicit Fox hardware snapshot.""" + return FoxDeviceSnapshot( + serial=serial, + station_id=station_id, + device_type=device_type, + online=online, + has_battery=has_battery, + has_pv=has_pv, + third_party_generation=third_party_generation, + scheduler_supported=scheduler_supported, + ) + + +def publisher(enabled=True, state_store=None, provider_id="fox-cloud"): + """Build one publisher and its in-memory durable store.""" + if state_store is None: + state_store = InMemoryFragmentAdapterStateStore() + return ( + FoxCloudFragmentPublisher( + provider_id, + state_store, + enabled=enabled, + ), + state_store, + ) + + +class TestFoxCloudFragmentPublisher(unittest.TestCase): + """Fox discovery remains pure, reference-only, and invalidatable.""" + + def test_default_off_has_no_validation_or_durable_side_effect(self): + """Disabled construction never imports live Fox or writes state.""" + adapter, state_store = publisher(enabled=False) + + module_dir = os.path.abspath( + os.path.join(os.path.dirname(__file__), ".."), + ) + isolated = subprocess.run( + ( + sys.executable, + "-c", + "import sys; import lattice_fox_fragment; " "assert 'fox' not in sys.modules", + ), + cwd=module_dir, + check=False, + capture_output=True, + text=True, + ) + self.assertEqual( + isolated.returncode, + 0, + isolated.stderr, + ) + self.assertIsNone(adapter.lattice_fragment_adapter()) + self.assertFalse(adapter.ingest_discovery(-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_discovery_is_immutable_reference_only_and_station_grouped(self): + """Snapshot topology carries observations but no authority.""" + adapter, state_store = publisher() + stations = [station()] + devices = [ + device(serial="60ke8020479c034"), + device( + serial="PV-ONLY", + has_battery=False, + scheduler_supported=False, + ), + ] + + self.assertTrue( + adapter.ingest_discovery( + 7, + stations, + devices, + health=ProviderHealth.HEALTHY, + ) + ) + stations.append(station("MUTATED")) + devices.append(device(serial="MUTATED")) + snapshot = adapter.read_snapshot() + topology = snapshot.topology_fragment + + self.assertIs(adapter.lattice_fragment_adapter(), adapter) + self.assertEqual(state_store.writes, 1) + self.assertEqual(adapter.generation, 1) + self.assertEqual(adapter.discovery_version, 7) + self.assertEqual(len(adapter.semantic_fingerprint), 64) + self.assertEqual(snapshot.health, ProviderHealth.HEALTHY) + self.assertEqual( + tuple(node["id"] for node in topology["nodes"]), + ( + "fox:station:SITE-1", + "fox:device:60KE8020479C034", + "fox:device:PV-ONLY", + ), + ) + self.assertEqual( + tuple((edge["from"], edge["to"], edge["type"]) for edge in topology["relationships"]), + ( + ( + "fox:station:SITE-1", + "fox:device:60KE8020479C034", + "contains", + ), + ( + "fox:station:SITE-1", + "fox:device:PV-ONLY", + "contains", + ), + ), + ) + self.assertEqual(snapshot.role_assignments, ()) + self.assertEqual(snapshot.config_projections, ()) + self.assertTrue(all(alias.roles == frozenset((AliasRole.REFERENCE,)) for alias in snapshot.aliases)) + self.assertTrue(all(node["capabilities"] == () for node in topology["nodes"])) + self.assertEqual( + tuple((alias.kind, alias.value) for alias in snapshot.identity_aliases), + ( + ("fox-station-id", "SITE-1"), + ("serial", "60KE8020479C034"), + ("serial", "PV-ONLY"), + ), + ) + + def test_order_case_and_explicit_features_are_replay_stable(self): + """Equivalent provider observations do not invent generations.""" + adapter, state_store = publisher() + first = ( + device(serial="INV-2"), + device(serial="inv-1"), + ) + replay = ( + device(serial="INV-1"), + device(serial="INV-2"), + ) + + self.assertTrue(adapter.ingest_discovery(3, (station(),), first)) + self.assertFalse(adapter.ingest_discovery(3, (station(),), replay)) + self.assertEqual(adapter.generation, 1) + self.assertEqual(state_store.writes, 1) + with self.assertRaisesRegex(ValueError, "has_pv"): + device(has_pv=1) + + def test_versions_are_monotonic_and_reuse_collision_fails_closed(self): + """Stale replay is ignored and same-version mutation is rejected.""" + adapter, state_store = publisher() + adapter.ingest_discovery(5, (station(),), (device(),), health=True) + before = adapter.read_state() + + self.assertFalse( + adapter.ingest_discovery( + 4, + (station(name="Stale"),), + (device(online=False),), + health=False, + ) + ) + with self.assertRaisesRegex( + FragmentAdapterConflict, + "reused Fox discovery version 5", + ): + adapter.ingest_discovery( + 5, + (station(),), + (device(device_type="Different"),), + ) + + self.assertEqual(adapter.read_state(), before) + self.assertEqual(state_store.writes, 1) + + def test_station_serial_and_membership_collisions_fail_before_write(self): + """Ambiguous provider identity never enters durable state.""" + adapter, state_store = publisher() + + with self.assertRaisesRegex(ValueError, "at least one"): + adapter.ingest_discovery(1, (), (device(),)) + with self.assertRaisesRegex( + FragmentAdapterConflict, + "duplicate station SITE-1", + ): + adapter.ingest_discovery( + 1, + (station(), station()), + (device(),), + ) + with self.assertRaisesRegex(ValueError, "unknown station OTHER"): + adapter.ingest_discovery( + 1, + (station(),), + (device(station_id="OTHER"),), + ) + with self.assertRaisesRegex( + FragmentAdapterConflict, + "belongs to both stations SITE-1 and SITE-2", + ): + adapter.ingest_discovery( + 1, + (station(), station("SITE-2")), + ( + device(), + device( + serial="60ke8020479c034", + station_id="SITE-2", + ), + ), + ) + + self.assertEqual(state_store.writes, 0) + with self.assertRaises(FragmentAdapterReadError): + adapter.read_state() + + def test_accepted_changes_trigger_common_fresh_read_compilation(self): + """Discovery and liveness both invalidate the shared compiler.""" + adapter, _state_store = publisher() + adapter.ingest_discovery(1, (station(),), (device(),), health=True) + reasons = [] + adapter.subscribe_invalidation( + lambda provider_id, generation, reason, feedback_token: reasons.append( + (provider_id, generation, reason, feedback_token), + ) + ) + registry = FragmentAdapterRegistry(enabled=True) + self.assertEqual(registry.discover((adapter,)), ("fox-cloud",)) + compiler = registry.create_compiler( + InMemoryCompiledLatticeStateStore(), + ) + + first = compiler.drain() + self.assertEqual(first.status, CompileStatus.FRESH) + self.assertEqual( + first.publication.provider_generations, + (("fox-cloud", 1),), + ) + + self.assertTrue( + adapter.ingest_discovery( + 2, + (station(),), + (device(), device(serial="INV-2")), + ) + ) + changed = compiler.drain() + self.assertEqual(changed.status, CompileStatus.FRESH) + self.assertEqual( + changed.publication.provider_generations, + (("fox-cloud", 2),), + ) + self.assertTrue(adapter.set_liveness(False)) + offline = compiler.drain() + self.assertEqual(offline.status, CompileStatus.STALE) + self.assertTrue(offline.pending) + self.assertEqual( + tuple(reasons), + ( + ( + "fox-cloud", + 2, + "Fox discovery changed to version 2", + None, + ), + ( + "fox-cloud", + 3, + "Fox liveness changed to offline", + None, + ), + ), + ) + + def test_liveness_is_monotonic_and_preserves_exact_topology(self): + """Health changes advance generation without rediscovery.""" + adapter, state_store = publisher() + with self.assertRaisesRegex(FragmentAdapterReadError, "seeded"): + adapter.set_liveness(True) + adapter.ingest_discovery(1, (station(),), (device(),)) + topology = adapter.read_snapshot().topology_fragment + + self.assertTrue(adapter.set_liveness(True)) + self.assertFalse(adapter.set_liveness(True)) + self.assertTrue(adapter.set_liveness(None)) + self.assertEqual( + adapter.read_snapshot().health, + ProviderHealth.DEGRADED, + ) + self.assertEqual(adapter.read_snapshot().topology_fragment, topology) + self.assertEqual(adapter.generation, 3) + self.assertEqual(state_store.writes, 3) + + def test_serial_composes_with_ge_without_materialization(self): + """Explicit strong serial joins cloud views but grants no role.""" + fox, _fox_store = publisher() + fox.ingest_discovery(1, (station(),), (device(),), health=True) + ge = GECloudFragmentPublisher( + "ge-cloud", + InMemoryFragmentAdapterStateStore(), + enabled=True, + ) + ge.ingest_discovery( + 1, + ( + GECloudDeviceSnapshot( + serial="60KE8020479C034", + kind="battery-inverter", + online=True, + ), + ), + health=True, + ) + + plan = compile_auto_config((fox.read_snapshot(), ge.read_snapshot())) + + device_nodes = tuple(node for node in plan.topology["nodes"] if node["kind"] == "inverter") + self.assertEqual(len(device_nodes), 1) + self.assertEqual( + device_nodes[0]["id"], + "identity:serial:60KE8020479C034", + ) + self.assertEqual(plan.role_assignments, ()) + self.assertEqual(plan.config_arguments, ()) + self.assertFalse(plan.materialization_readiness.ready) + + def test_removal_is_durable_invalidating_and_irreversible(self): + """Ambient discovery replay cannot resurrect a removed provider.""" + adapter, state_store = publisher() + adapter.ingest_discovery(5, (station(),), (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()) + before = adapter.read_state() + self.assertTrue(before.removed) + with self.assertRaises(FragmentAdapterRemoved): + adapter.read_snapshot() + with self.assertRaisesRegex( + FragmentAdapterRemoved, + "discovery cannot re-enrol", + ): + adapter.ingest_discovery( + 99, + (station(),), + (device(),), + health=True, + ) + self.assertEqual(adapter.read_state(), before) + self.assertEqual(state_store.writes, 2) + self.assertEqual( + invalidations[-1], + ("fox-cloud", 2, "Fox integration removed"), + ) + + restarted = FoxCloudFragmentPublisher( + "fox-cloud", + state_store, + enabled=True, + ) + self.assertTrue(restarted.read_state().removed) + with self.assertRaises(FragmentAdapterRemoved): + restarted.read_snapshot() + with self.assertRaises(FragmentAdapterRemoved): + restarted.ingest_discovery( + 100, + (station(),), + (device(),), + ) + + +if __name__ == "__main__": + unittest.main()