From 613929f29133fa0d8a403cfbbcdc8ecc65d7afea Mon Sep 17 00:00:00 2001 From: Mark Gascoyne Date: Tue, 28 Jul 2026 11:01:15 +0100 Subject: [PATCH] feat: add SolaX cloud lattice fragment publisher --- apps/predbat/lattice_solax_fragment.py | 605 ++++++++++++++++++ .../tests/test_lattice_solax_fragment.py | 493 ++++++++++++++ 2 files changed, 1098 insertions(+) create mode 100644 apps/predbat/lattice_solax_fragment.py create mode 100644 apps/predbat/tests/test_lattice_solax_fragment.py diff --git a/apps/predbat/lattice_solax_fragment.py b/apps/predbat/lattice_solax_fragment.py new file mode 100644 index 000000000..c3647ef49 --- /dev/null +++ b/apps/predbat/lattice_solax_fragment.py @@ -0,0 +1,605 @@ +# ----------------------------------------------------------------------------- +# Predbat Home Battery System - SolaX 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 SolaX Cloud fragment publisher. + +The live ``SolaxAPI`` component owns authentication, plant/device discovery, +telemetry polling, automatic configuration, and controls. This module does +not import, register, or mutate that component. A future gated seam may pass +explicit immutable discovery snapshots here after successful cloud reads. + +Every accepted discovery version, plant/device change, liveness change, or +removal publishes a new durable generation through the common fragment +adapter. Published aliases remain REFERENCE-only: there are no materialization +roles, configuration projections, capabilities, or control writes. +""" + +# 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() +_DEVICE_KINDS = frozenset( + ( + "battery", + "ev-charger", + "inverter", + "meter", + ) +) +_TOPOLOGY_KINDS = { + "battery": "battery", + "ev-charger": "ev-charger", + "inverter": "inverter", + "meter": "meter", +} + + +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_model(value): + """Normalize the cloud model code/name without accepting booleans.""" + if value is None: + return None + if isinstance(value, int) and not isinstance(value, bool): + return str(value) + return _required_text(value, "model") + + +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", + ) + + +def _online_status(value): + """Normalize SolaX boolean/zero/one online status without guessing others.""" + if value is None: + return None + if isinstance(value, bool): + return value + if isinstance(value, int) and value in (0, 1): + return value == 1 + raise ValueError("online must be True, False, 1, 0, or None") + + +@dataclass(frozen=True) +class SolaxPlantSnapshot: + """One explicit provider-local SolaX plant.""" + + plant_id: str + name: Optional[str] = None + + def __post_init__(self): + """Normalize the cloud-owned stable plant identity.""" + object.__setattr__( + self, + "plant_id", + _required_text(self.plant_id, "plant_id"), + ) + object.__setattr__(self, "name", _optional_text(self.name, "name")) + + @property + def node_id(self): + """Return the deterministic provider-local plant node identity.""" + return "solax:plant:{}".format(self.plant_id) + + +@dataclass(frozen=True) +class SolaxDeviceSnapshot: + """One explicit provider-local SolaX device.""" + + serial: str + plant_id: str + kind: str + online: Optional[object] = None + model: Optional[object] = None + synthetic: bool = False + source_serial: Optional[str] = None + + def __post_init__(self): + """Normalize identities and quarantine synthetic battery placeholders.""" + serial = _required_text(self.serial, "serial").upper() + plant_id = _required_text(self.plant_id, "plant_id") + kind = _required_text(self.kind, "kind").lower().replace("_", "-") + if kind not in _DEVICE_KINDS: + raise ValueError( + "kind must be one of {}".format( + ", ".join(sorted(_DEVICE_KINDS)), + ) + ) + if not isinstance(self.synthetic, bool): + raise ValueError("synthetic must be a boolean") + source_serial = _optional_text(self.source_serial, "source_serial") + if source_serial is not None: + source_serial = source_serial.upper() + if self.synthetic: + if kind != "battery": + raise ValueError("only a battery snapshot may be synthetic") + if source_serial is None: + raise ValueError( + "a synthetic battery requires source_serial", + ) + expected = "{}_BATTERY".format(source_serial) + if serial != expected: + raise ValueError( + "synthetic battery serial must be {}".format(expected), + ) + elif source_serial is not None: + raise ValueError( + "source_serial is only valid for a synthetic battery", + ) + + object.__setattr__(self, "serial", serial) + object.__setattr__(self, "plant_id", plant_id) + object.__setattr__(self, "kind", kind) + object.__setattr__(self, "online", _online_status(self.online)) + object.__setattr__(self, "model", _optional_model(self.model)) + object.__setattr__(self, "source_serial", source_serial) + + @property + def node_id(self): + """Return the deterministic provider-local device node identity.""" + if self.synthetic: + return "solax:synthetic:{}:battery:{}".format( + self.plant_id, + self.source_serial, + ) + return "solax:device:{}".format(self.serial) + + +def _normalize_plants(plants): + """Freeze, sort, and collision-check plant snapshots.""" + try: + plants = tuple(plants) + except TypeError as exc: + raise ValueError( + "plants must be an iterable of SolaxPlantSnapshot", + ) from exc + if any(not isinstance(plant, SolaxPlantSnapshot) for plant in plants): + raise ValueError("plants must contain only SolaxPlantSnapshot values") + if not plants: + raise ValueError("plants must contain at least one SolaxPlantSnapshot") + + seen = set() + for plant in plants: + if plant.plant_id in seen: + raise FragmentAdapterConflict( + "SolaX discovery contains duplicate plant {}".format( + plant.plant_id, + ) + ) + seen.add(plant.plant_id) + return tuple(sorted(plants, key=lambda item: item.plant_id)) + + +def _normalize_devices(devices, plant_ids): + """Freeze, sort, and collision-check device snapshots.""" + try: + devices = tuple(devices) + except TypeError as exc: + raise ValueError( + "devices must be an iterable of SolaxDeviceSnapshot", + ) from exc + if any(not isinstance(device, SolaxDeviceSnapshot) for device in devices): + raise ValueError("devices must contain only SolaxDeviceSnapshot values") + if not devices: + raise ValueError("devices must contain at least one SolaxDeviceSnapshot") + + serial_owners = {} + node_owners = {} + for device in devices: + if device.plant_id not in plant_ids: + raise ValueError( + "device {} references unknown plant {}".format( + device.serial, + device.plant_id, + ) + ) + owner = serial_owners.get(device.serial) + if owner is not None: + raise FragmentAdapterConflict( + "SolaX serial {} belongs to both plants {} and {}".format( + device.serial, + owner, + device.plant_id, + ) + ) + serial_owners[device.serial] = device.plant_id + node_owner = node_owners.get(device.node_id) + if node_owner is not None: + raise FragmentAdapterConflict( + "SolaX node {} is claimed by both {} and {}".format( + device.node_id, + node_owner, + device.serial, + ) + ) + node_owners[device.node_id] = device.serial + return tuple( + sorted( + devices, + key=lambda item: ( + item.plant_id, + item.kind, + item.serial, + ), + ) + ) + + +def _plant_node(plant, provider_id): + """Project one plant without claiming a control capability.""" + attributes = {"plantId": plant.plant_id} + if plant.name is not None: + attributes["name"] = plant.name + return { + "id": plant.node_id, + "kind": "site", + "attributes": attributes, + "accessPaths": [ + { + "id": "solax-cloud-api", + "provider": provider_id, + "preference": 0, + } + ], + "capabilities": [], + } + + +def _device_node(device, provider_id): + """Project one device without claiming a read or control capability.""" + attributes = { + "deviceSn": device.serial, + "plantId": device.plant_id, + "solaxKind": device.kind, + "synthetic": device.synthetic, + } + if device.online is not None: + attributes["online"] = device.online + if device.model is not None: + attributes["model"] = device.model + if device.source_serial is not None: + attributes["sourceSerial"] = device.source_serial + return { + "id": device.node_id, + "kind": _TOPOLOGY_KINDS[device.kind], + "attributes": attributes, + "accessPaths": [ + { + "id": "solax-cloud-api", + "provider": provider_id, + "preference": 0, + } + ], + "capabilities": [], + } + + +def _topology_document( + provider_id, + discovery_version, + plants, + devices, +): + """Build one deterministic provider-owned v0.3 topology fragment.""" + nodes = [_plant_node(plant, provider_id) for plant in plants] + nodes.extend(_device_node(device, provider_id) for device in devices) + relationships = [ + { + "from": "solax:plant:{}".format(device.plant_id), + "to": device.node_id, + "type": "contains", + } + for device in devices + ] + return { + "topologyVersion": "0.3.0", + "scope": "fragment", + "docVersion": discovery_version, + "producer": { + "name": "SolaX Cloud", + "provider": provider_id, + "authority": 0, + }, + "nodes": nodes, + "relationships": relationships, + } + + +def _reference_aliases(plants, devices): + """Publish provider-local names that never grant materialization roles.""" + aliases = [ + ProviderAlias( + name="plant:{}".format(plant.plant_id), + node_id=plant.node_id, + roles=frozenset((AliasRole.REFERENCE,)), + ) + for plant in plants + ] + aliases.extend( + ProviderAlias( + name=("synthetic:{}".format(device.serial) if device.synthetic else "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(plants, devices): + """Publish only globally unambiguous plant and hardware identities.""" + aliases = [ + ProviderIdentityAlias( + kind="solax-plant-id", + value=plant.plant_id, + node_id=plant.node_id, + ) + for plant in plants + ] + aliases.extend( + ProviderIdentityAlias( + kind="serial", + value=device.serial, + node_id=device.node_id, + ) + for device in devices + if not device.synthetic + ) + return tuple( + sorted( + aliases, + key=lambda item: (item.kind, item.value, item.node_id), + ) + ) + + +class SolaxCloudFragmentPublisher: + """Adapt explicit SolaX 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 current provider-owned discovery version.""" + state = self.read_state() + return _plain(state.snapshot.topology_fragment)["docVersion"] + + def lattice_fragment_adapter(self): + """Expose structural discovery only when enabled and durably 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, device, liveness, and removal changes.""" + 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_discovery( + self, + discovery_version, + plants, + devices, + health=_HEALTH_UNCHANGED, + ): + """Publish one accepted complete plant/device discovery snapshot.""" + if not self._enabled: + return False + discovery_version = _discovery_version(discovery_version) + plants = _normalize_plants(plants) + devices = _normalize_devices( + devices, + frozenset(plant.plant_id for plant in plants), + ) + document = _topology_document( + self.provider_id, + discovery_version, + plants, + devices, + ) + + with self._lock: + current = self._current_state() + if current is not None and current.removed: + raise FragmentAdapterRemoved( + "provider {} was removed at generation {}; SolaX " + "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 SolaX 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(plants, devices), + identity_aliases=_identity_aliases(plants, devices), + role_assignments=(), + config_projections=(), + ) + published = self._adapter.publish( + snapshot, + "SolaX discovery changed to version {}".format( + discovery_version, + ), + ) + if published: + self._seeded = True + return published + + def set_liveness(self, health): + """Publish provider 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( + "SolaX 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, + "SolaX 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( + "SolaX discovery must be seeded before removal", + ) + if current.removed: + return False + return self._adapter.remove( + current.generation + 1, + "SolaX integration removed", + ) diff --git a/apps/predbat/tests/test_lattice_solax_fragment.py b/apps/predbat/tests/test_lattice_solax_fragment.py new file mode 100644 index 000000000..f0fc54cc7 --- /dev/null +++ b/apps/predbat/tests/test_lattice_solax_fragment.py @@ -0,0 +1,493 @@ +"""Tests for the pure SolaX Cloud 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 + FragmentAdapterConflict, + FragmentAdapterReadError, + FragmentAdapterRegistry, + FragmentAdapterRemoved, + InMemoryFragmentAdapterStateStore, +) +from lattice_ge_cloud_fragment import ( # noqa: E402 + GECloudDeviceSnapshot, + GECloudFragmentPublisher, +) +from lattice_solax_fragment import ( # noqa: E402 + SolaxCloudFragmentPublisher, + SolaxDeviceSnapshot, + SolaxPlantSnapshot, +) + + +def plant(plant_id="1618699116555534337", name="Home"): + """Build one explicit provider-local SolaX plant snapshot.""" + return SolaxPlantSnapshot( + plant_id=plant_id, + name=name, + ) + + +def device( + serial="H1231231932123", + plant_id="1618699116555534337", + kind="inverter", + online=1, + model=14, + synthetic=False, + source_serial=None, +): + """Build one explicit provider-local SolaX device snapshot.""" + return SolaxDeviceSnapshot( + serial=serial, + plant_id=plant_id, + kind=kind, + online=online, + model=model, + synthetic=synthetic, + source_serial=source_serial, + ) + + +def publisher(enabled=True, state_store=None, provider_id="solax-cloud"): + """Build one reference publisher and its in-memory durable store.""" + if state_store is None: + state_store = InMemoryFragmentAdapterStateStore() + return ( + SolaxCloudFragmentPublisher( + provider_id, + state_store, + enabled=enabled, + ), + state_store, + ) + + +class TestSolaxCloudFragmentPublisher(unittest.TestCase): + """SolaX discovery publishes immutable, REFERENCE-only snapshots.""" + + def test_default_off_has_no_discovery_or_durable_side_effect(self): + """Disabled construction never validates, publishes, or registers.""" + adapter, state_store = publisher(enabled=False) + + 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_publishes_immutable_reference_only_snapshot(self): + """Plant/device state is detached and carries no write authority.""" + adapter, state_store = publisher() + plants = [plant()] + devices = [ + device(), + device( + serial="TP123456123123", + kind="battery", + model=1, + ), + ] + + self.assertTrue( + adapter.ingest_discovery( + 7, + plants, + devices, + health=ProviderHealth.HEALTHY, + ) + ) + plants.append(plant("MUTATED")) + devices.append(device(serial="MUTATED")) + snapshot = adapter.read_snapshot() + nodes = snapshot.topology_fragment["nodes"] + + 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 nodes), + ( + "solax:plant:1618699116555534337", + "solax:device:TP123456123123", + "solax:device:H1231231932123", + ), + ) + self.assertEqual( + nodes[1]["attributes"]["model"], + "1", + ) + self.assertEqual(snapshot.role_assignments, ()) + self.assertEqual(snapshot.config_projections, ()) + self.assertTrue(all(node["capabilities"] == () for node in nodes)) + self.assertTrue(all(alias.roles == frozenset((AliasRole.REFERENCE,)) for alias in snapshot.aliases)) + self.assertEqual( + tuple((alias.kind, alias.value, alias.node_id) for alias in snapshot.identity_aliases), + ( + ( + "serial", + "H1231231932123", + "solax:device:H1231231932123", + ), + ( + "serial", + "TP123456123123", + "solax:device:TP123456123123", + ), + ( + "solax-plant-id", + "1618699116555534337", + "solax:plant:1618699116555534337", + ), + ), + ) + self.assertEqual( + tuple( + ( + relationship["from"], + relationship["to"], + relationship["type"], + ) + for relationship in snapshot.topology_fragment["relationships"] + ), + ( + ( + "solax:plant:1618699116555534337", + "solax:device:TP123456123123", + "contains", + ), + ( + "solax:plant:1618699116555534337", + "solax:device:H1231231932123", + "contains", + ), + ), + ) + + def test_order_case_and_online_encoding_are_replay_stable(self): + """Equivalent cloud snapshots do not invent a new generation.""" + adapter, state_store = publisher() + first = ( + device(serial="inv-1", online=1), + device(serial="bat-1", kind="battery", online=0), + ) + replay = ( + device(serial="BAT-1", kind="battery", online=False), + device(serial="INV-1", online=True), + ) + + self.assertTrue( + adapter.ingest_discovery( + 3, + (plant(),), + first, + health=True, + ) + ) + self.assertFalse( + adapter.ingest_discovery( + 3, + (plant(),), + replay, + health=True, + ) + ) + self.assertEqual(adapter.generation, 1) + self.assertEqual(state_store.writes, 1) + + def test_versions_are_monotonic_and_reuse_collision_fails_closed(self): + """Stale replay is ignored and same-version content reuse is rejected.""" + adapter, state_store = publisher() + adapter.ingest_discovery(5, (plant(),), (device(),), health=True) + before = adapter.read_state() + + self.assertFalse( + adapter.ingest_discovery( + 4, + (plant(name="Stale"),), + (device(model=99),), + health=False, + ) + ) + with self.assertRaisesRegex( + FragmentAdapterConflict, + "reused SolaX discovery version 5", + ): + adapter.ingest_discovery( + 5, + (plant(name="Different"),), + (device(),), + ) + + self.assertEqual(adapter.read_state(), before) + self.assertEqual(state_store.writes, 1) + + def test_collisions_and_unknown_plants_fail_before_publication(self): + """Ambiguous provider identities cannot enter durable state.""" + adapter, state_store = publisher() + + with self.assertRaisesRegex(ValueError, "at least one"): + adapter.ingest_discovery(1, (), (device(),)) + with self.assertRaisesRegex(ValueError, "name"): + plant(name=1) + with self.assertRaisesRegex(ValueError, "unknown plant"): + adapter.ingest_discovery( + 1, + (plant(),), + (device(plant_id="other"),), + ) + with self.assertRaisesRegex( + FragmentAdapterConflict, + "belongs to both plants", + ): + adapter.ingest_discovery( + 1, + (plant(), plant("other")), + ( + device(), + device(plant_id="other"), + ), + ) + + self.assertEqual(state_store.writes, 0) + with self.assertRaises(FragmentAdapterReadError): + adapter.read_state() + + def test_synthetic_battery_has_no_false_hardware_identity(self): + """SolaX placeholder batteries remain provider-local references.""" + adapter, _state_store = publisher() + synthetic = device( + serial="H1231231932123_battery", + kind="battery", + synthetic=True, + source_serial="H1231231932123", + ) + + adapter.ingest_discovery( + 1, + (plant(),), + (device(), synthetic), + health=True, + ) + snapshot = adapter.read_snapshot() + identity_values = tuple((alias.kind, alias.value) for alias in snapshot.identity_aliases) + + self.assertIn(("serial", "H1231231932123"), identity_values) + self.assertNotIn(("serial", "H1231231932123_BATTERY"), identity_values) + self.assertEqual( + snapshot.topology_fragment["nodes"][1]["id"], + ("solax:synthetic:1618699116555534337:" "battery:H1231231932123"), + ) + self.assertEqual( + snapshot.topology_fragment["nodes"][1]["attributes"]["sourceSerial"], + "H1231231932123", + ) + + def test_accepted_changes_trigger_common_fresh_read_composition(self): + """Discovery/device/liveness changes invalidate the common compiler.""" + adapter, _state_store = publisher() + adapter.ingest_discovery( + 1, + (plant(),), + (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,)), ("solax-cloud",)) + compiler = registry.create_compiler( + InMemoryCompiledLatticeStateStore(), + ) + + first = compiler.drain() + self.assertEqual(first.status, CompileStatus.FRESH) + self.assertEqual( + first.publication.provider_generations, + (("solax-cloud", 1),), + ) + + self.assertTrue( + adapter.ingest_discovery( + 2, + (plant(),), + ( + device(), + device(serial="BAT-1", kind="battery"), + ), + ) + ) + changed = compiler.drain() + self.assertEqual(changed.status, CompileStatus.FRESH) + self.assertEqual( + changed.publication.provider_generations, + (("solax-cloud", 2),), + ) + + self.assertTrue(adapter.set_liveness(False)) + offline = compiler.drain() + self.assertEqual(offline.status, CompileStatus.STALE) + self.assertTrue(offline.pending) + self.assertEqual(adapter.generation, 3) + self.assertEqual( + tuple(reasons), + ( + ( + "solax-cloud", + 2, + "SolaX discovery changed to version 2", + None, + ), + ( + "solax-cloud", + 3, + "SolaX liveness changed to offline", + None, + ), + ), + ) + + def test_liveness_is_independent_monotonic_and_preserves_topology(self): + """Provider health changes retain exact discovery contents.""" + adapter, state_store = publisher() + with self.assertRaisesRegex(FragmentAdapterReadError, "seeded"): + adapter.set_liveness(True) + adapter.ingest_discovery(1, (plant(),), (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_identity_composes_with_ge_cloud_without_materialization(self): + """Two cloud views join without creating config or control authority.""" + solax, _solax_store = publisher() + solax.ingest_discovery( + 1, + (plant(),), + (device(),), + health=True, + ) + ge_cloud = GECloudFragmentPublisher( + "ge-cloud", + InMemoryFragmentAdapterStateStore(), + enabled=True, + ) + ge_cloud.ingest_discovery( + 1, + ( + GECloudDeviceSnapshot( + serial="H1231231932123", + kind="battery-inverter", + online=True, + ), + ), + health=True, + ) + + plan = compile_auto_config( + ( + ge_cloud.read_snapshot(), + solax.read_snapshot(), + ) + ) + + self.assertEqual( + tuple(node["id"] for node in plan.topology["nodes"]), + ( + "identity:serial:H1231231932123", + "identity:solax-plant-id:1618699116555534337", + ), + ) + self.assertEqual( + plan.provider_generations, + (("ge-cloud", 1), ("solax-cloud", 1)), + ) + 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.assertFalse(plan.materialization_readiness.ready) + + def test_removal_is_durable_invalidating_and_irreversible_by_replay(self): + """Ordinary discovery replay cannot re-enrol a removed integration.""" + adapter, state_store = publisher() + adapter.ingest_discovery( + 5, + (plant(),), + (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( + 1, + (plant(),), + (device(),), + health=True, + ) + self.assertEqual(adapter.read_state(), before) + self.assertEqual(state_store.writes, 2) + self.assertEqual( + invalidations[-1], + ("solax-cloud", 2, "SolaX integration removed"), + ) + + restarted = SolaxCloudFragmentPublisher( + "solax-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( + 99, + (plant(),), + (device(),), + ) + + +if __name__ == "__main__": + unittest.main()