From 44027ca0d9e8fdd0e687a4c19a9801d3dae6b71e Mon Sep 17 00:00:00 2001 From: Mark Gascoyne Date: Tue, 28 Jul 2026 15:29:30 +0100 Subject: [PATCH] feat: add Solis lattice fragment publisher --- apps/predbat/lattice_solis_fragment.py | 624 ++++++++++++++++++ .../tests/test_lattice_solis_fragment.py | 571 ++++++++++++++++ 2 files changed, 1195 insertions(+) create mode 100644 apps/predbat/lattice_solis_fragment.py create mode 100644 apps/predbat/tests/test_lattice_solis_fragment.py diff --git a/apps/predbat/lattice_solis_fragment.py b/apps/predbat/lattice_solis_fragment.py new file mode 100644 index 000000000..e31b11c8c --- /dev/null +++ b/apps/predbat/lattice_solis_fragment.py @@ -0,0 +1,624 @@ +# ----------------------------------------------------------------------------- +# Predbat Home Battery System - Solis 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 Solis fragment publisher. + +The live ``SolisAPI`` component owns cloud authentication, discovery, CID +reads/writes, automatic configuration, and controls. Local Solis inverter +profiles are a separate configuration path. This module imports, registers, +and mutates neither path. A future gated seam may pass explicit immutable +snapshots here after it has proved which local/cloud paths address one device. + +Only a provider-owned device id and an explicitly verified hardware serial are +identity. Station ids, configured serials, endpoints, authentication methods, +protocol versions, and CID/TOU profiles remain provider metadata. Published +aliases are REFERENCE-only, with no 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() +_CHANNELS = frozenset(("cloud", "local")) +_CLOUD_AUTH_METHODS = frozenset(("hmac", "oauth")) + + +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 one provider 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 _input_version(value, name): + """Validate one provider-owned monotonic input version.""" + if not isinstance(value, int) or isinstance(value, bool) or value < 0: + raise ValueError("{} must be a non-negative integer".format(name)) + 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 Solis 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 SolisAccessPathSnapshot: + """One explicit access path; its coordinates never establish identity.""" + + path_id: str + channel: str + protocol_version: str + cid_profile: str + auth_method: Optional[str] = None + endpoint_id: Optional[str] = None + + def __post_init__(self): + """Normalize access metadata and reject local/cloud ambiguity.""" + path_id = _required_text(self.path_id, "path_id") + channel = _required_text(self.channel, "channel").lower() + if channel not in _CHANNELS: + raise ValueError("channel must be cloud or local") + protocol_version = _required_text( + self.protocol_version, + "protocol_version", + ) + cid_profile = _required_text(self.cid_profile, "cid_profile") + auth_method = _optional_text(self.auth_method, "auth_method") + if auth_method is not None: + auth_method = auth_method.lower() + endpoint_id = _optional_text(self.endpoint_id, "endpoint_id") + + if channel == "cloud": + if auth_method not in _CLOUD_AUTH_METHODS: + raise ValueError( + "a cloud access path requires hmac or oauth auth_method", + ) + elif auth_method is not None: + raise ValueError( + "a local access path must not carry a cloud auth_method", + ) + + object.__setattr__(self, "path_id", path_id) + object.__setattr__(self, "channel", channel) + object.__setattr__(self, "protocol_version", protocol_version) + object.__setattr__(self, "cid_profile", cid_profile) + object.__setattr__(self, "auth_method", auth_method) + object.__setattr__(self, "endpoint_id", endpoint_id) + + +@dataclass(frozen=True) +class SolisDeviceSnapshot: + """One explicitly grouped provider-local Solis inverter.""" + + device_id: str + access_paths: tuple + serial: Optional[str] = None + serial_verified: bool = False + online: Optional[object] = None + model: Optional[object] = None + has_battery: Optional[bool] = None + station_id: Optional[str] = None + + def __post_init__(self): + """Normalize state while keeping unproved serials provider-local.""" + device_id = _required_text(self.device_id, "device_id") + try: + access_paths = tuple(self.access_paths) + except TypeError as exc: + raise ValueError( + "access_paths must be an iterable of " "SolisAccessPathSnapshot", + ) from exc + if not access_paths: + raise ValueError( + "access_paths must contain at least one " "SolisAccessPathSnapshot", + ) + if any(not isinstance(path, SolisAccessPathSnapshot) for path in access_paths): + raise ValueError( + "access_paths must contain only SolisAccessPathSnapshot " "values", + ) + path_ids = set() + for path in access_paths: + if path.path_id in path_ids: + raise FragmentAdapterConflict( + "Solis device {} contains duplicate access path {}".format( + device_id, + path.path_id, + ) + ) + path_ids.add(path.path_id) + + serial = _optional_text(self.serial, "serial") + if serial is not None: + serial = serial.upper() + if not isinstance(self.serial_verified, bool): + raise ValueError("serial_verified must be a boolean") + if self.serial_verified and serial is None: + raise ValueError("serial_verified requires serial") + if self.has_battery is not None and not isinstance( + self.has_battery, + bool, + ): + raise ValueError("has_battery must be True, False, or None") + + object.__setattr__(self, "device_id", device_id) + object.__setattr__( + self, + "access_paths", + tuple(sorted(access_paths, key=lambda item: item.path_id)), + ) + object.__setattr__(self, "serial", serial) + object.__setattr__(self, "online", _online_status(self.online)) + object.__setattr__(self, "model", _optional_model(self.model)) + object.__setattr__( + self, + "station_id", + _optional_text(self.station_id, "station_id"), + ) + + @property + def node_id(self): + """Return the deterministic provider-local device node identity.""" + return "solis:device:{}".format(self.device_id) + + +def _normalize_devices(devices): + """Freeze, sort, and collision-check explicit device snapshots.""" + try: + devices = tuple(devices) + except TypeError as exc: + raise ValueError( + "devices must be an iterable of SolisDeviceSnapshot", + ) from exc + if any(not isinstance(device, SolisDeviceSnapshot) for device in devices): + raise ValueError( + "devices must contain only SolisDeviceSnapshot values", + ) + if not devices: + raise ValueError( + "devices must contain at least one SolisDeviceSnapshot", + ) + + device_ids = set() + verified_serials = {} + for device in devices: + if device.device_id in device_ids: + raise FragmentAdapterConflict( + "Solis discovery contains duplicate device {}".format( + device.device_id, + ) + ) + device_ids.add(device.device_id) + if device.serial_verified: + owner = verified_serials.get(device.serial) + if owner is not None: + raise FragmentAdapterConflict( + "verified Solis serial {} belongs to both {} and {}".format( + device.serial, + owner, + device.device_id, + ) + ) + verified_serials[device.serial] = device.device_id + return tuple(sorted(devices, key=lambda item: item.device_id)) + + +def _access_path(path, provider_id): + """Project provider coordinates without asserting a capability.""" + projected = { + "id": path.path_id, + "provider": provider_id, + "preference": 0, + "channel": path.channel, + "protocolVersion": path.protocol_version, + "cidProfile": path.cid_profile, + } + if path.auth_method is not None: + projected["authMethod"] = path.auth_method + if path.endpoint_id is not None: + projected["endpointId"] = path.endpoint_id + return projected + + +def _device_node(device, provider_id): + """Project one inverter without claiming a read or control capability.""" + attributes = { + "solisDeviceId": device.device_id, + "serialVerified": device.serial_verified, + } + if device.serial is not None: + attributes["serial"] = device.serial + if device.online is not None: + attributes["online"] = device.online + if device.model is not None: + attributes["model"] = device.model + if device.has_battery is not None: + attributes["hasBattery"] = device.has_battery + if device.station_id is not None: + attributes["stationId"] = device.station_id + return { + "id": device.node_id, + "kind": "inverter", + "attributes": attributes, + "accessPaths": [_access_path(path, provider_id) for path in device.access_paths], + "capabilities": [], + } + + +def _topology_document( + provider_id, + generation, + discovery_version, + config_version, + devices, +): + """Build one deterministic provider-owned v0.3 topology fragment.""" + return { + "topologyVersion": "0.3.0", + "scope": "fragment", + "docVersion": generation, + "producer": { + "name": "Solis", + "provider": provider_id, + "authority": 0, + "inputVersions": { + "discovery": discovery_version, + "config": config_version, + }, + }, + "nodes": [_device_node(device, provider_id) for device in devices], + } + + +def _semantic_document(document): + """Return topology semantics without the derived durable generation.""" + document = _plain(document) + document.pop("docVersion", None) + return document + + +def _input_versions(document): + """Read and validate the durable two-input cursor.""" + try: + versions = document["producer"]["inputVersions"] + return ( + _input_version(versions["discovery"], "discovery_version"), + _input_version(versions["config"], "config_version"), + ) + except (KeyError, TypeError) as exc: + raise FragmentAdapterReadError( + "durable Solis fragment has no valid inputVersions cursor", + ) from exc + + +def _reference_aliases(devices): + """Publish provider-local names that never grant materialization roles.""" + aliases = [] + for device in devices: + aliases.append( + ProviderAlias( + name="device:{}".format(device.device_id), + node_id=device.node_id, + roles=frozenset((AliasRole.REFERENCE,)), + ) + ) + if device.serial is not None: + prefix = "serial" if device.serial_verified else "configured-serial" + aliases.append( + ProviderAlias( + name="{}:{}".format(prefix, device.serial), + node_id=device.node_id, + roles=frozenset((AliasRole.REFERENCE,)), + ) + ) + return tuple( + sorted( + aliases, + key=lambda item: (item.name, item.node_id), + ) + ) + + +def _identity_aliases(provider_id, devices): + """Publish provider-qualified ids and only explicitly verified serials.""" + aliases = [] + for device in devices: + aliases.append( + ProviderIdentityAlias( + kind="solis-provider-device-id", + value="{}:{}".format(provider_id, device.device_id), + node_id=device.node_id, + ) + ) + if device.serial_verified: + aliases.append( + ProviderIdentityAlias( + kind="serial", + value=device.serial, + node_id=device.node_id, + ) + ) + return tuple( + sorted( + aliases, + key=lambda item: (item.kind, item.value, item.node_id), + ) + ) + + +class SolisFragmentPublisher: + """Adapt explicit Solis 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 input_versions(self): + """Fresh-read the current discovery/config input cursor.""" + return _input_versions( + _plain(self.read_state().snapshot.topology_fragment), + ) + + 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, config, 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_snapshot( + self, + discovery_version, + config_version, + devices, + health=_HEALTH_UNCHANGED, + ): + """Publish one accepted complete discovery/config snapshot.""" + if not self._enabled: + return False + discovery_version = _input_version( + discovery_version, + "discovery_version", + ) + config_version = _input_version(config_version, "config_version") + devices = _normalize_devices(devices) + + with self._lock: + current = self._current_state() + if current is not None and current.removed: + raise FragmentAdapterRemoved( + "provider {} was removed at generation {}; Solis input " + "cannot re-enrol it".format( + self.provider_id, + current.generation, + ) + ) + next_generation = 1 if current is None else current.generation + 1 + document = _topology_document( + self.provider_id, + next_generation, + discovery_version, + config_version, + devices, + ) + 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_discovery, previous_config = _input_versions(previous) + discovery_regressed = discovery_version < previous_discovery + config_regressed = config_version < previous_config + discovery_advanced = discovery_version > previous_discovery + config_advanced = config_version > previous_config + if discovery_regressed or config_regressed: + if not discovery_advanced and not config_advanced: + return False + raise FragmentAdapterConflict( + "provider {} mixed Solis input regression from " + "discovery/config {}/{} to {}/{}".format( + self.provider_id, + previous_discovery, + previous_config, + discovery_version, + config_version, + ) + ) + same_versions = not discovery_advanced and not config_advanced + if same_versions and _semantic_document(document) != _semantic_document(previous): + raise FragmentAdapterConflict( + "provider {} reused Solis discovery/config versions " + "{}/{} for different content".format( + self.provider_id, + discovery_version, + config_version, + ) + ) + next_health = current.snapshot.health if health is _HEALTH_UNCHANGED else _provider_health(health) + if same_versions and next_health is current.snapshot.health: + return False + + snapshot = ProviderSnapshot( + provider_id=self.provider_id, + generation=next_generation, + health=next_health, + topology_fragment=document, + aliases=_reference_aliases(devices), + identity_aliases=_identity_aliases( + self.provider_id, + devices, + ), + role_assignments=(), + config_projections=(), + ) + published = self._adapter.publish( + snapshot, + "Solis inputs changed to discovery/config {}/{}".format( + discovery_version, + config_version, + ), + ) + if published: + self._seeded = True + return published + + def set_liveness(self, health): + """Publish provider liveness without changing structural inputs.""" + if not self._enabled: + return False + health = _provider_health(health) + with self._lock: + current = self._current_state() + if current is None: + raise FragmentAdapterReadError( + "Solis inputs 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, + "Solis 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( + "Solis inputs must be seeded before removal", + ) + if current.removed: + return False + return self._adapter.remove( + current.generation + 1, + "Solis integration removed", + ) diff --git a/apps/predbat/tests/test_lattice_solis_fragment.py b/apps/predbat/tests/test_lattice_solis_fragment.py new file mode 100644 index 000000000..75ed72c2a --- /dev/null +++ b/apps/predbat/tests/test_lattice_solis_fragment.py @@ -0,0 +1,571 @@ +"""Tests for the pure Solis 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_gateway_fragment import ( # noqa: E402 + GatewayRetainedTopologyFragmentPublisher, +) +from lattice_ge_cloud_fragment import ( # noqa: E402 + GECloudDeviceSnapshot, + GECloudFragmentPublisher, +) +from lattice_solax_fragment import ( # noqa: E402 + SolaxCloudFragmentPublisher, + SolaxDeviceSnapshot, + SolaxPlantSnapshot, +) +from lattice_solis_fragment import ( # noqa: E402 + SolisAccessPathSnapshot, + SolisDeviceSnapshot, + SolisFragmentPublisher, +) + + +def access_path( + path_id="solis-cloud-oauth", + channel="cloud", + protocol_version="v1", + cid_profile="tou-v2", + auth_method="oauth", + endpoint_id="solis-oauth-api", +): + """Build one explicit Solis access path.""" + return SolisAccessPathSnapshot( + path_id=path_id, + channel=channel, + protocol_version=protocol_version, + cid_profile=cid_profile, + auth_method=auth_method, + endpoint_id=endpoint_id, + ) + + +def device( + device_id="cloud-record-1", + access_paths=None, + serial="1910A12345", + serial_verified=True, + online=1, + model="S6-EH1P", + has_battery=True, + station_id="station-7", +): + """Build one explicitly grouped Solis device snapshot.""" + if access_paths is None: + access_paths = (access_path(),) + return SolisDeviceSnapshot( + device_id=device_id, + access_paths=access_paths, + serial=serial, + serial_verified=serial_verified, + online=online, + model=model, + has_battery=has_battery, + station_id=station_id, + ) + + +def publisher(enabled=True, state_store=None, provider_id="solis"): + """Build one reference publisher and its in-memory durable store.""" + if state_store is None: + state_store = InMemoryFragmentAdapterStateStore() + return ( + SolisFragmentPublisher( + provider_id, + state_store, + enabled=enabled, + ), + state_store, + ) + + +class TestSolisFragmentPublisher(unittest.TestCase): + """Solis inputs publish immutable, REFERENCE-only provider snapshots.""" + + def test_default_off_has_no_validation_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_snapshot(-1, -1, (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_snapshot_is_immutable_reference_only_and_keeps_metadata_local(self): + """Structural state carries no configuration or control authority.""" + adapter, state_store = publisher() + paths = [ + access_path(), + access_path( + path_id="solis-local-modbus", + channel="local", + protocol_version="modbus-v1", + cid_profile="GS_fb00", + auth_method=None, + endpoint_id="192.0.2.4", + ), + ] + devices = [device(access_paths=paths)] + + self.assertTrue( + adapter.ingest_snapshot( + 4, + 9, + devices, + health=True, + ) + ) + paths.append(access_path(path_id="mutated")) + devices.append(device(device_id="mutated", serial="MUTATED")) + snapshot = adapter.read_snapshot() + document = snapshot.topology_fragment + node = document["nodes"][0] + + self.assertIs(adapter.lattice_fragment_adapter(), adapter) + self.assertEqual(state_store.writes, 1) + self.assertEqual(adapter.generation, 1) + self.assertEqual(adapter.input_versions, (4, 9)) + self.assertEqual(len(adapter.semantic_fingerprint), 64) + self.assertEqual(snapshot.health, ProviderHealth.HEALTHY) + self.assertEqual(document["docVersion"], 1) + self.assertEqual( + document["producer"]["inputVersions"], + {"config": 9, "discovery": 4}, + ) + self.assertEqual(node["id"], "solis:device:cloud-record-1") + self.assertEqual(node["kind"], "inverter") + self.assertEqual(node["capabilities"], ()) + self.assertEqual( + tuple(path["id"] for path in node["accessPaths"]), + ("solis-cloud-oauth", "solis-local-modbus"), + ) + self.assertEqual(snapshot.role_assignments, ()) + self.assertEqual(snapshot.config_projections, ()) + 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", + "1910A12345", + "solis:device:cloud-record-1", + ), + ( + "solis-provider-device-id", + "solis:cloud-record-1", + "solis:device:cloud-record-1", + ), + ), + ) + identity_text = repr(snapshot.identity_aliases) + self.assertNotIn("station-7", identity_text) + self.assertNotIn("S6-EH1P", identity_text) + self.assertNotIn("tou-v2", identity_text) + self.assertNotIn("GS_fb00", identity_text) + self.assertNotIn("192.0.2.4", identity_text) + + def test_local_cloud_and_cid_ambiguity_fail_closed(self): + """Access metadata cannot silently correlate separate devices.""" + with self.assertRaisesRegex(ValueError, "requires hmac or oauth"): + access_path(auth_method=None) + with self.assertRaisesRegex(ValueError, "must not carry"): + access_path( + channel="local", + auth_method="oauth", + ) + with self.assertRaisesRegex( + FragmentAdapterConflict, + "duplicate access path", + ): + device(access_paths=(access_path(), access_path())) + + adapter, _state_store = publisher() + shared_cid = "tou-v2" + adapter.ingest_snapshot( + 1, + 1, + ( + device( + device_id="cloud-a", + serial=None, + serial_verified=False, + access_paths=( + access_path( + path_id="cloud-a", + cid_profile=shared_cid, + ), + ), + ), + device( + device_id="cloud-b", + serial=None, + serial_verified=False, + access_paths=( + access_path( + path_id="local-b", + channel="local", + protocol_version="modbus-v1", + cid_profile=shared_cid, + auth_method=None, + ), + ), + ), + ), + ) + plan = compile_auto_config((adapter.read_snapshot(),)) + + self.assertEqual(len(plan.topology["nodes"]), 2) + self.assertFalse(any(alias.kind in ("cid", "cid-profile", "protocol-version") for alias in adapter.read_snapshot().identity_aliases)) + + def test_unverified_configured_serial_never_becomes_strong_identity(self): + """A configured value is only a reference until discovery proves it.""" + adapter, _state_store = publisher() + adapter.ingest_snapshot( + 1, + 1, + ( + device( + serial="configured-only", + serial_verified=False, + ), + ), + ) + snapshot = adapter.read_snapshot() + + self.assertEqual( + tuple((alias.kind, alias.value) for alias in snapshot.identity_aliases), + ( + ( + "solis-provider-device-id", + "solis:cloud-record-1", + ), + ), + ) + self.assertEqual( + tuple(alias.name for alias in snapshot.aliases), + ( + "configured-serial:CONFIGURED-ONLY", + "device:cloud-record-1", + ), + ) + + def test_verified_serial_and_provider_ids_must_not_collide(self): + """One provider cannot assert the same durable identity twice.""" + adapter, state_store = publisher() + with self.assertRaisesRegex( + FragmentAdapterConflict, + "duplicate device", + ): + adapter.ingest_snapshot( + 1, + 1, + ( + device(), + device(serial="OTHER"), + ), + ) + with self.assertRaisesRegex( + FragmentAdapterConflict, + "verified Solis serial", + ): + adapter.ingest_snapshot( + 1, + 1, + ( + device(), + device( + device_id="cloud-record-2", + serial="1910a12345", + ), + ), + ) + self.assertEqual(state_store.writes, 0) + + def test_discovery_and_config_versions_advance_durable_generation(self): + """Either provider input can invalidate a fresh compiler read.""" + adapter, state_store = publisher() + self.assertTrue(adapter.ingest_snapshot(1, 1, (device(),))) + first = adapter.read_state() + self.assertTrue(adapter.ingest_snapshot(2, 1, (device(),))) + self.assertTrue(adapter.ingest_snapshot(2, 2, (device(),))) + self.assertFalse(adapter.ingest_snapshot(2, 2, (device(),))) + + self.assertEqual(adapter.input_versions, (2, 2)) + self.assertEqual(adapter.generation, 3) + self.assertEqual(state_store.writes, 3) + self.assertNotEqual( + first.semantic_fingerprint, + adapter.semantic_fingerprint, + ) + self.assertEqual( + adapter.read_snapshot().topology_fragment["docVersion"], + 3, + ) + + def test_stale_mixed_and_same_version_reuse_are_fail_closed(self): + """Input cursors cannot regress or be reused for changed contents.""" + adapter, state_store = publisher() + adapter.ingest_snapshot(5, 8, (device(),), health=True) + before = adapter.read_state() + + self.assertFalse( + adapter.ingest_snapshot( + 4, + 7, + (device(model="stale"),), + health=False, + ) + ) + with self.assertRaisesRegex( + FragmentAdapterConflict, + "mixed Solis input regression", + ): + adapter.ingest_snapshot(6, 7, (device(),)) + with self.assertRaisesRegex( + FragmentAdapterConflict, + "reused Solis discovery/config versions 5/8", + ): + adapter.ingest_snapshot( + 5, + 8, + (device(model="changed"),), + ) + + self.assertEqual(adapter.read_state(), before) + self.assertEqual(state_store.writes, 1) + + def test_all_changes_invalidate_common_compiler_with_fresh_reads(self): + """Discovery, config, liveness, and removal all notify composition.""" + adapter, _state_store = publisher() + adapter.ingest_snapshot(1, 1, (device(),), health=True) + invalidations = [] + adapter.subscribe_invalidation( + lambda provider_id, generation, reason, feedback_token: ( + invalidations.append( + ( + provider_id, + generation, + reason, + feedback_token, + ) + ) + ) + ) + registry = FragmentAdapterRegistry(enabled=True) + self.assertEqual(registry.discover((adapter,)), ("solis",)) + compiler = registry.create_compiler( + InMemoryCompiledLatticeStateStore(), + ) + + first = compiler.drain() + self.assertEqual(first.status, CompileStatus.FRESH) + self.assertTrue(adapter.ingest_snapshot(2, 1, (device(),))) + discovery = compiler.drain() + self.assertEqual(discovery.status, CompileStatus.FRESH) + self.assertTrue(adapter.ingest_snapshot(2, 2, (device(),))) + config = compiler.drain() + self.assertEqual(config.status, CompileStatus.FRESH) + self.assertTrue(adapter.set_liveness(False)) + offline = compiler.drain() + self.assertEqual(offline.status, CompileStatus.STALE) + self.assertTrue(offline.pending) + self.assertTrue(adapter.remove()) + removed = compiler.drain() + self.assertEqual(removed.status, CompileStatus.STALE) + + self.assertEqual( + tuple(item[:3] for item in invalidations), + ( + ( + "solis", + 2, + "Solis inputs changed to discovery/config 2/1", + ), + ( + "solis", + 3, + "Solis inputs changed to discovery/config 2/2", + ), + ( + "solis", + 4, + "Solis liveness changed to offline", + ), + ( + "solis", + 5, + "Solis integration removed", + ), + ), + ) + + def test_restart_preserves_cursor_and_removal_is_irreversible(self): + """Durable versions survive restart and a tombstone cannot be replayed.""" + adapter, state_store = publisher() + adapter.ingest_snapshot(3, 4, (device(),), health=True) + restarted = SolisFragmentPublisher( + "solis", + state_store, + enabled=True, + ) + self.assertEqual(restarted.input_versions, (3, 4)) + self.assertFalse(restarted.ingest_snapshot(3, 4, (device(),))) + self.assertTrue(restarted.remove()) + + after_removal = SolisFragmentPublisher( + "solis", + state_store, + enabled=True, + ) + self.assertTrue(after_removal.read_state().removed) + with self.assertRaises(FragmentAdapterRemoved): + after_removal.read_snapshot() + with self.assertRaisesRegex( + FragmentAdapterRemoved, + "cannot re-enrol", + ): + after_removal.ingest_snapshot(99, 99, (device(),)) + + def test_verified_serial_composes_gateway_ge_and_solax_only_explicitly(self): + """Four proven views join while unverified Solis state stays separate.""" + serial = "1910A12345" + solis, _solis_store = publisher() + solis.ingest_snapshot( + 1, + 1, + ( + device(serial=serial), + device( + device_id="configured-shadow", + serial=serial, + serial_verified=False, + ), + ), + health=True, + ) + + gateway = GatewayRetainedTopologyFragmentPublisher( + "predbat-gateway", + InMemoryFragmentAdapterStateStore(), + enabled=True, + ) + gateway.ingest_retained_topology( + { + "topologyVersion": "0.3.0", + "scope": "fragment", + "docVersion": 1, + "producer": { + "name": "PredBat Gateway", + "provider": "predbat-gateway", + "authority": 10, + }, + "nodes": [ + { + "id": "gateway-inverter", + "kind": "inverter", + "attributes": {"serial": serial}, + "accessPaths": [], + "capabilities": [], + } + ], + }, + online=True, + ) + + ge_cloud = GECloudFragmentPublisher( + "ge-cloud", + InMemoryFragmentAdapterStateStore(), + enabled=True, + ) + ge_cloud.ingest_discovery( + 1, + ( + GECloudDeviceSnapshot( + serial=serial, + kind="battery-inverter", + online=True, + ), + ), + health=True, + ) + + solax = SolaxCloudFragmentPublisher( + "solax-cloud", + InMemoryFragmentAdapterStateStore(), + enabled=True, + ) + solax.ingest_discovery( + 1, + (SolaxPlantSnapshot(plant_id="plant-1"),), + ( + SolaxDeviceSnapshot( + serial=serial, + plant_id="plant-1", + kind="inverter", + online=True, + ), + ), + health=True, + ) + + plan = compile_auto_config( + ( + gateway.read_snapshot(), + ge_cloud.read_snapshot(), + solax.read_snapshot(), + solis.read_snapshot(), + ) + ) + + self.assertEqual( + tuple(node["id"] for node in plan.topology["nodes"]), + ( + "identity:serial:1910A12345", + "identity:solax-plant-id:plant-1", + ("identity:solis-provider-device-id:" "solis:configured-shadow"), + ), + ) + self.assertEqual( + plan.provider_generations, + ( + ("ge-cloud", 1), + ("predbat-gateway", 1), + ("solax-cloud", 1), + ("solis", 1), + ), + ) + self.assertEqual(plan.role_assignments, ()) + self.assertEqual(plan.config_arguments, ()) + self.assertEqual(dict(plan.projected_config), {}) + self.assertFalse(plan.materialization_readiness.ready) + + +if __name__ == "__main__": + unittest.main()