From c3225ccfb05736fc6cc9160cd7541584c59d2b73 Mon Sep 17 00:00:00 2001 From: Mark Gascoyne Date: Mon, 12 Oct 2026 22:24:00 +0100 Subject: [PATCH] feat: observe sealed destination transport readiness --- apps/predbat/control_handoff.py | 4 + apps/predbat/control_readiness.py | 227 +++++++++++++++++++ apps/predbat/gateway.py | 71 ++++++ apps/predbat/tests/test_control_readiness.py | 224 ++++++++++++++++++ 4 files changed, 526 insertions(+) create mode 100644 apps/predbat/control_readiness.py create mode 100644 apps/predbat/tests/test_control_readiness.py diff --git a/apps/predbat/control_handoff.py b/apps/predbat/control_handoff.py index 2940953de..468cea7b1 100644 --- a/apps/predbat/control_handoff.py +++ b/apps/predbat/control_handoff.py @@ -393,6 +393,10 @@ def start_control_handoff(base): service.stop() raise ControlAttachFailed("Controller startup cannot replace an attached gate") config = base.args.get("control_handoff") + if base.args.get("control_readiness") is not None or isinstance(config, dict) and "readiness" in config: + from control_readiness import observe_loaded_target + + return observe_loaded_target(base) if config is None or isinstance(config, dict) and config.get("enabled", False) is False: return None boot_id = str(uuid.uuid4()) diff --git a/apps/predbat/control_readiness.py b/apps/predbat/control_readiness.py new file mode 100644 index 000000000..fa67bafa0 --- /dev/null +++ b/apps/predbat/control_readiness.py @@ -0,0 +1,227 @@ +"""Sealed, one-shot destination transport observations, never activation authority.""" + +import hashlib +import asyncio +import math +import os +import re +import struct +import time +import uuid +from datetime import datetime, timezone +from urllib.parse import urlsplit + +import requests + +from control_handoff import ControlAttachFailed, ControlWriteGate + +BATTERY_ARGUMENTS = ( + "user_id", + "ge_cloud_direct", + "ge_cloud_automatic", + "ge_cloud_automatic_split_pv", + "ge_cloud_automatic_split_shared", + "ge_cloud_automatic_ems", + "ge_cloud_serial", + "ge_cloud_data", + "gateway_device_id", + "gateway_mqtt_host", + "gateway_mqtt_port", + "gateway_mqtt_auth_method", + "gateway_inverter_serial", + "gateway_display_only", + "gateway_shared_ct", + "gateway_integrate_power", + "inverter_type", + "num_inverters", + "givtcp_rest", + "auto_restart", + "set_read_only", +) + + +def canonical_battery_value(value): + """Encode exactly the versioned TypeScript manifest byte representation.""" + if value is None: + return "n" + if value is True: + return "t" + if value is False: + return "f" + if type(value) in (int, float): + if not math.isfinite(value) or abs(value) > 9007199254740991: + raise ValueError("Invalid manifest number") + return "d" + struct.pack(">d", 0 if value == 0 else value).hex() + if isinstance(value, str): + encoded = value.encode("utf-8", errors="strict") + # TypeScript measures UTF-16 code units for this bounded text field. + if len(value.encode("utf-16-le")) // 2 > 4096: + raise ValueError("Invalid manifest text") + return "s{}:{}".format(len(encoded), value) + if isinstance(value, list): + if len(value) > 256: + raise ValueError("Oversized manifest array") + return "a{}:".format(len(value)) + "".join(canonical_battery_value(item) for item in value) + if type(value) is dict: + if len(value) > 256 or any(not isinstance(key, str) or not re.fullmatch(r"[\x20-\x7e]{1,128}", key) for key in value): + raise ValueError("Invalid manifest keys") + return "o{}:".format(len(value)) + "".join(canonical_battery_value(key) + canonical_battery_value(value[key]) for key in sorted(value)) + raise ValueError("Unsupported manifest value") + + +def loaded_battery_manifest(args, issued_manifest): + """Project loaded transport args while retaining server-issued route references.""" + if type(issued_manifest) is not dict or set(issued_manifest) != {"schema", "provider", "routes", "credential_ref", "battery_args"} or type(issued_manifest.get("schema")) is not int or issued_manifest.get("schema") != 1: + raise ValueError("Invalid issued manifest") + return {"schema": 1, "provider": issued_manifest["provider"], "routes": issued_manifest["routes"], "credential_ref": issued_manifest["credential_ref"], "battery_args": {key: args[key] for key in BATTERY_ARGUMENTS if key in args}} + + +def battery_manifest_digest(manifest): + """Hash only projected non-secret arguments and explicit credential references.""" + return hashlib.sha256(("predbat-battery-config-v1\n" + canonical_battery_value(manifest)).encode("utf-8")).hexdigest() + + +def _uuid(value): + """Require a canonical, lowercase UUID string without coercion.""" + if not isinstance(value, str): + return False + try: + parsed = uuid.UUID(value) + return str(parsed) == value and parsed.version in range(1, 9) + except ValueError: + return False + + +def _fresh_timestamp(value, max_age=300): + """Require an explicitly zoned recent sample, including bounded future skew.""" + if not isinstance(value, str) or not re.search(r"(Z|[+-]\d{2}:\d{2})$", value): + raise ValueError("Missing inverter timestamp") + stamp = datetime.fromisoformat(value.replace("Z", "+00:00")).timestamp() + age = time.time() - stamp + if not math.isfinite(stamp) or age > max_age or age < -30: + raise ValueError("Stale inverter timestamp") + return datetime.fromtimestamp(stamp, timezone.utc).isoformat() + + +def fresh_givenergy_observation(args, routes): + """Read the actual loaded credential directly, with no cache or optimistic state.""" + api_key = args.get("ge_cloud_key") + if args.get("ge_cloud_direct") is not True or not isinstance(api_key, str) or not api_key or args.get("givtcp_rest"): + raise ValueError("Unsupported target transport") + ref = routes[0]["gateway_ref"] + if ref.get("kind") != "instance_secret" or ref.get("instance_id") != args.get("user_id") or ref.get("integration") != "givenergy" or ref.get("path") != "api_key": + raise ValueError("Unsupported target credential reference") + + def read(path): + """Perform only one read and never follow a credential-bearing redirect.""" + response = requests.get("https://api.givenergy.cloud/v1/" + path, headers={"Authorization": "Bearer " + api_key, "Accept": "application/json"}, timeout=5, allow_redirects=False) + if response.status_code != 200: + raise ValueError("Target read unavailable") + value = response.json() + if type(value) is not dict: + raise ValueError("Invalid target response") + return value + + expected = sorted(route["serial"] for route in routes) + if not 1 <= len(expected) <= 16 or len(set(expected)) != len(expected) or any(route.get("manufacturer") != "givenergy" or route["gateway_ref"] != ref for route in routes): + raise ValueError("Unsupported physical routes") + discovery = read("communication-device") + if type(discovery.get("data")) is not list or discovery.get("meta", {}).get("last_page") != 1: + raise ValueError("Incomplete target discovery") + serials = [] + for device in discovery["data"]: + inverter = device.get("inverter") + if type(inverter) is not dict or type(inverter.get("connections", {}).get("batteries")) is not list: + raise ValueError("Incomplete target discovery") + if inverter["connections"]["batteries"]: + if any(name in inverter.get("info", {}).get("model", "").lower() for name in ("gateway", "plant ems", "gw2")): + raise ValueError("Aggregate target adapter unavailable") + serials.append(inverter["serial"].upper()) + if sorted(serials) != expected: + raise ValueError("Target physical identities differ") + # The current independent server adapter deliberately supports one battery. + # Do not claim the later runtime dispatch map for a multi-inverter account. + if len(expected) != 1: + raise ValueError("Multi-inverter target adapter unavailable") + data = read("inverter/{}/system-data/latest".format(expected[0])).get("data") + if type(data) is not dict or data.get("serial", expected[0]).upper() != expected[0]: + raise ValueError("Target telemetry identity differs") + percent = data.get("battery", {}).get("percent") + if type(percent) not in (float, int) or not math.isfinite(percent) or not 0 <= percent <= 100: + raise ValueError("Target telemetry unavailable") + return {"provider": "givenergy", "serials": expected, "sampled_at": _fresh_timestamp(data.get("time")), "checked_at": datetime.now(timezone.utc).isoformat(), "sealed": True, "pending": 0, "uncertain": 0} + + +def observe_loaded_target(base): + """Seal before any component, claim the loaded binding, observe, then stop startup.""" + boot_id = str(uuid.uuid4()) + gate = ControlWriteGate("target-readiness", boot_id, 1, time.monotonic() + 1) + gate.seal() + base.control_write_gate = gate + try: + args = base.args + envelope = args.get("control_readiness") + if args.get("control_handoff") != {"enabled": True, "readiness": True} or type(envelope) is not dict: + raise ValueError("Invalid readiness mode") + if envelope.get("state") != "challenge" or envelope.get("activation_available") is not False or envelope.get("boot_id") is not None: + raise ValueError("Invalid readiness envelope") + for name in ("challenge_id", "transition_id", "target_epoch"): + if not _uuid(envelope.get(name)): + raise ValueError("Invalid readiness identity") + version = args.get("saas_config_version") + if type(version) is not int or not 1 <= version <= 2147483647 or type(envelope.get("config_version")) is not int or version != envelope["config_version"]: + raise ValueError("Invalid loaded readiness revision") + if not isinstance(envelope.get("nonce"), str) or not re.fullmatch(r"[a-f0-9]{64}", envelope["nonce"]): + raise ValueError("Invalid readiness nonce") + manifest = loaded_battery_manifest(args, envelope["manifest"]) + digest = battery_manifest_digest(manifest) + if manifest != envelope["manifest"] or digest != envelope.get("manifest_digest") or manifest["provider"] != envelope.get("provider") or manifest["routes"] != envelope.get("routes"): + raise ValueError("Loaded readiness projection differs") + if args.get("auto_restart") or manifest["provider"] not in ("givenergy", "predbat_gateway"): + raise ValueError("Target transport observation unavailable") + expiry = datetime.fromisoformat(envelope["expires_at"].replace("Z", "+00:00")).timestamp() + if expiry <= time.time() or expiry > time.time() + 120: + raise ValueError("Invalid readiness deadline") + url = os.environ.get("SUPABASE_URL", "") + key = os.environ.get("SUPABASE_KEY", "") + parsed = urlsplit(url) + if not key or parsed.scheme != "https" or not parsed.netloc or parsed.username or parsed.password or parsed.path not in ("", "/") or parsed.query or parsed.fragment: + raise ValueError("Readiness service credentials unavailable") + binding = { + "transition_id": envelope["transition_id"], + "challenge_id": envelope["challenge_id"], + "target_epoch": envelope["target_epoch"], + "nonce": envelope["nonce"], + "loaded_config_version": version, + "manifest": manifest, + "manifest_digest": digest, + } + + def send(action, observation=None): + """Post one bounded observation request, preserving permanent local sealing.""" + if time.time() >= expiry: + raise ValueError("Readiness deadline expired") + payload = {"action": action, "challenge_id": envelope["challenge_id"], "boot_id": boot_id, "binding": binding, "loaded_credential": args.get("ge_cloud_key") if manifest["provider"] == "givenergy" else args.get("gateway_mqtt_token")} + if observation is not None: + payload["observation"] = observation + response = requests.post(url.rstrip("/") + "/functions/v1/engine-battery-readiness", json=payload, headers={"Authorization": "Bearer " + key}, timeout=min(10, expiry - time.time()), allow_redirects=False) + result = response.json() + if response.status_code != 200 or type(result) is not dict or result.get("state") != ("attached" if action == "attach" else "observed") or result.get("sealed") is not True or result.get("activation_available") is not False: + raise ValueError("Readiness service refused") + if result.get("target_epoch") != envelope["target_epoch"] or result.get("boot_id") != boot_id or result.get("challenge_id") != envelope["challenge_id"] or time.time() >= expiry: + raise ValueError("Readiness response binding differs") + return result + + attached = send("attach") + if manifest["provider"] == "givenergy": + observation = fresh_givenergy_observation(args, manifest["routes"]) + else: + from gateway import GatewayMQTT + + observation = asyncio.run(GatewayMQTT.observe_readonly_transport(base, attached.get("transport_wire"), attached.get("transport_control"), min(10, expiry - time.time()))) + send("observe", observation) + except Exception: + raise ControlAttachFailed("Sealed target observation unavailable; activation unavailable") from None + # This process never constructs components or runs the planner. The recorded + # observation is a bounded step towards readiness, not permission to control. + raise ControlAttachFailed("Sealed target observation recorded; activation unavailable") diff --git a/apps/predbat/gateway.py b/apps/predbat/gateway.py index a7b832cbc..000df205b 100644 --- a/apps/predbat/gateway.py +++ b/apps/predbat/gateway.py @@ -228,6 +228,77 @@ class GatewayMQTT(ComponentBase): # control writes (engine thread) and by acks arriving on the MQTT listener's loop. _command_lock = threading.Lock() + @staticmethod + async def observe_readonly_transport(base, wire, control, timeout_s): + """Round-trip one signed probe through the loaded MQTT credential. + + This separate one-shot path never starts ordinary component lifecycle, + schedules, preference sync or EV control. Only the service-issued probe + retry is publishable. The backend independently requires authenticated + device ACK ingress; a message here is not physical release authority. + """ + from control_readiness import canonical_battery_value + + args = base.args + gate = control_gate_for(base) + snapshot = gate.snapshot() if gate else {} + if not snapshot.get("closed") or snapshot.get("pending") != 0 or snapshot.get("uncertain") != 0: + raise ValueError("Target write admission must remain sealed") + if not HAS_AIOMQTT or type(wire) is not dict or set(wire) != {"device_id", "command_id", "command", "expires_at", "control_grant"} or type(control) is not dict or wire.get("command") != "probe_battery_control": + raise ValueError("Invalid read-only transport challenge") + device_id = args.get("gateway_device_id") + command_id = wire.get("command_id") + if not isinstance(device_id, str) or wire.get("device_id") != device_id or not isinstance(command_id, str) or str(uuid.UUID(command_id)) != command_id: + raise ValueError("Target MQTT challenge identity differs") + if type(wire.get("expires_at")) is not int or wire.get("expires_at") != control.get("expires_at") or control.get("domain") != "battery" or type(control.get("protocol")) is not int or control.get("protocol") != 1 or not wire.get("control_grant"): + raise ValueError("Target MQTT challenge binding differs") + if args.get("gateway_display_only", False) is not False or not args.get("gateway_mqtt_token") or not args.get("gateway_mqtt_host") or type(args.get("gateway_mqtt_port", 8883)) is not int: + raise ValueError("Target MQTT configuration unsupported") + topic = "predbat/devices/" + device_id + expected = {**control, "phase": "probe_ready", "persisted": True, "pending": 0, "uncertain": 0} + async with asyncio.timeout(timeout_s): + async with aiomqtt.Client( + hostname=args["gateway_mqtt_host"], + port=args.get("gateway_mqtt_port", 8883), + username=device_id, + password=args["gateway_mqtt_token"], + tls_context=ssl.create_default_context(), + identifier="predbat-readiness-" + uuid.uuid4().hex, + clean_session=True, + ) as client: + ack_topic = topic + "/ack/" + command_id + result = await client.subscribe(ack_topic, qos=0) + if not result or any(int(getattr(code, "value", code)) >= 0x80 for code in result): + raise ValueError("Target MQTT subscription refused") + # No unsigned preference command or executable retained plan can + # pass this path, even while all normal write sinks stay sealed. + await client.publish(topic + "/command", json.dumps(wire, separators=(",", ":")), qos=1, retain=False) + async for message in client.messages: + if str(message.topic) != ack_topic or getattr(message, "retain", True) is not False: + continue + if len(message.payload) > 2048: + raise ValueError("Oversized target MQTT receipt") + value = json.loads(message.payload) + if type(value) is not dict or value.get("command_id") != command_id or value.get("command") != "probe_battery_control" or value.get("ok") is not True: + raise ValueError("Target MQTT probe refused") + receipt = value.get("applied", {}).get("control") + if canonical_battery_value(receipt) != canonical_battery_value(expected): + raise ValueError("Target MQTT receipt binding differs") + snapshot = gate.snapshot() + if not snapshot["closed"] or snapshot["pending"] or snapshot["uncertain"]: + raise ValueError("Target write admission changed") + return { + "provider": "predbat_gateway", + "serials": receipt["inverter_serials"], + "command_id": command_id, + "receipt": receipt, + "checked_at": datetime.datetime.now(datetime.timezone.utc).isoformat(), + "sealed": True, + "pending": 0, + "uncertain": 0, + } + raise ValueError("Target MQTT receipt unavailable") + # Defaults for an instance built without initialize() gateway_shared_ct = True gateway_integrate_power = False diff --git a/apps/predbat/tests/test_control_readiness.py b/apps/predbat/tests/test_control_readiness.py new file mode 100644 index 000000000..ab2ad915f --- /dev/null +++ b/apps/predbat/tests/test_control_readiness.py @@ -0,0 +1,224 @@ +"""Test protected loaded projection and actual sealed startup without real endpoints.""" + +import copy +import asyncio +import json +import os +import time +import unittest +from datetime import datetime, timezone +from types import SimpleNamespace +from unittest.mock import patch + +from control_handoff import ControlAttachFailed, ControlWriteBlocked, start_control_handoff +from control_readiness import battery_manifest_digest, canonical_battery_value, loaded_battery_manifest +from control_handoff import ControlWriteGate +from gateway import GatewayMQTT + + +class ControlReadinessTests(unittest.TestCase): + """Observe fresh physical identity while no component or battery writer can start.""" + + def setUp(self): + """Build a synthetic service-issued envelope over real loaded arguments.""" + self.args = {"user_id": "10000000-0000-4000-8000-000000000001", "ge_cloud_direct": True, "ge_cloud_key": "synthetic-key", "num_inverters": 1} + ref = {"kind": "instance_secret", "instance_id": self.args["user_id"], "integration": "givenergy", "path": "api_key", "config_version": 7} + routes = [{"manufacturer": "givenergy", "serial": "FD123", "gateway_ref": ref}] + manifest = loaded_battery_manifest(self.args, {"schema": 1, "provider": "givenergy", "routes": routes, "credential_ref": ref, "battery_args": {}}) + self.envelope = { + "state": "challenge", + "activation_available": False, + "challenge_id": "20000000-0000-4000-8000-000000000001", + "transition_id": "30000000-0000-4000-8000-000000000001", + "target_epoch": "40000000-0000-4000-8000-000000000001", + "nonce": "ab" * 32, + "config_version": 7, + "provider": "givenergy", + "routes": routes, + "manifest": manifest, + "manifest_digest": battery_manifest_digest(manifest), + "boot_id": None, + "expires_at": datetime.fromtimestamp(time.time() + 60, timezone.utc).isoformat(), + } + self.args.update(control_readiness=self.envelope, control_handoff={"enabled": True, "readiness": True}, saas_config_version=7) + self.base = SimpleNamespace(args=self.args) + self.env = {"SUPABASE_URL": "https://synthetic.invalid", "SUPABASE_KEY": "synthetic-service-key"} + + def response(self, value, status=200): + """Return a real requests-shaped response with synthetic contents.""" + return SimpleNamespace(status_code=status, json=lambda: value) + + def post(self, _url, **kwargs): + """Echo only immutable service response fields and inspect local admission.""" + self.assertTrue(self.base.control_write_gate.snapshot()["closed"]) + with self.assertRaises(ControlWriteBlocked): + self.base.control_write_gate.admit("givenergy") + body = kwargs["json"] + return self.response( + {"state": "attached" if body["action"] == "attach" else "observed", "challenge_id": body["challenge_id"], "target_epoch": body["binding"]["target_epoch"], "boot_id": body["boot_id"], "sealed": True, "activation_available": False} + ) + + def get(self, url, **kwargs): + """Serve fresh account and telemetry reads using the actual loaded key.""" + self.assertEqual(kwargs["headers"]["Authorization"], "Bearer synthetic-key") + self.assertFalse(kwargs["allow_redirects"]) + if url.endswith("communication-device"): + return self.response({"data": [{"inverter": {"serial": "fd123", "connections": {"batteries": [{}]}}}], "meta": {"last_page": 1}}) + return self.response({"data": {"time": datetime.now(timezone.utc).isoformat(), "battery": {"percent": 45}}}) + + def test_cross_language_canonical_bytes(self): + """Match the exact TypeScript fixture including Unicode and IEEE-754 values.""" + self.assertEqual(canonical_battery_value({"z": -0.0, "a": [None, True, False, "é😀", 1.5]}), "o2:s1:aa5:ntfs6:é😀d3ff8000000000000s1:zd0000000000000000") + self.assertEqual(canonical_battery_value(9007199254740991), "d433fffffffffffff") + for value in (float("nan"), float("inf"), 9007199254740992, "\ud800", {"é": 1}, [None] * 257): + with self.subTest(value=repr(value)), self.assertRaises((ValueError, UnicodeError)): + canonical_battery_value(value) + + def test_actual_startup_records_observation_and_never_constructs_components(self): + """Run PredBat.initialize, not merely the isolated helper function.""" + # Import the actual orchestrator without its unrelated release updater. + with patch("download.check_install", return_value=(True, False)): + import predbat as module + + self.base.log = lambda *_: None + self.base.update_time = lambda **_kwargs: None + self.base.now_utc_real = datetime.now(timezone.utc) + self.base.reset = lambda: None + self.base.record_status = lambda *_args, **_kwargs: None + with patch.dict(os.environ, self.env), patch("control_readiness.requests.post", side_effect=self.post) as post, patch("control_readiness.requests.get", side_effect=self.get) as get, patch.object(module, "Components") as components: + with self.assertRaisesRegex(ControlAttachFailed, "observation recorded"): + module.PredBat.initialize(self.base) + components.assert_not_called() + self.assertEqual([call.kwargs["json"]["action"] for call in post.call_args_list], ["attach", "observe"]) + self.assertEqual(get.call_count, 2) + self.assertEqual(post.call_args_list[1].kwargs["json"]["observation"]["serials"], ["FD123"]) + + def test_changed_loaded_arguments_or_revision_never_claim_boot(self): + """A corrected boot is possible because stale local values never attach.""" + for key, value in (("num_inverters", 2), ("saas_config_version", "7"), ("saas_config_version", True), ("auto_restart", True)): + with self.subTest(key=key, value=value): + args = copy.deepcopy(self.args) + args[key] = value + base = SimpleNamespace(args=args) + with patch.dict(os.environ, self.env), patch("control_readiness.requests.post") as post, patch("control_readiness.requests.get") as get, self.assertRaises(ControlAttachFailed): + start_control_handoff(base) + post.assert_not_called() + get.assert_not_called() + self.assertTrue(base.control_write_gate.snapshot()["closed"]) + + def test_stale_or_wrong_physical_identity_never_reports_observation(self): + """Successful account access cannot replace current exact inverter evidence.""" + for telemetry in ({"time": "2000-01-01T00:00:00Z", "battery": {"percent": 45}}, {"time": datetime.now(timezone.utc).isoformat(), "serial": "OTHER", "battery": {"percent": 45}}): + self.base = SimpleNamespace(args=copy.deepcopy(self.args)) + + def get(url, **kwargs): + """Substitute only the fresh endpoint, keeping real discovery parsing.""" + return self.get(url, **kwargs) if url.endswith("communication-device") else self.response({"data": telemetry}) + + with patch.dict(os.environ, self.env), patch("control_readiness.requests.post", side_effect=self.post) as post, patch("control_readiness.requests.get", side_effect=get), self.assertRaises(ControlAttachFailed): + start_control_handoff(self.base) + self.assertEqual(post.call_count, 1) + self.assertTrue(self.base.control_write_gate.snapshot()["closed"]) + + +class HubReadinessTransportTests(unittest.TestCase): + """Race actual GatewayMQTT one-shot publishing and receipt validation.""" + + def setUp(self): + """Prepare one sealed process and a bounded signed-probe-shaped fixture.""" + self.base = SimpleNamespace(args={"gateway_device_id": "pbgw_test", "gateway_mqtt_host": "synthetic.invalid", "gateway_mqtt_port": 8883, "gateway_mqtt_token": "synthetic-token"}) + self.base.control_write_gate = ControlWriteGate("target", "epoch", 7, time.monotonic() + 60) + self.base.control_write_gate.seal() + self.command_id = "20000000-0000-4000-8000-000000000001" + self.control = { + "protocol": 1, + "domain": "battery", + "transition_id": "transition", + "controller_epoch": "epoch", + "boot_id": "current-boot", + "generation": 4, + "config_version": 7, + "inverter_serials": ["FD123", "FD456"], + "for_s": 120, + "expires_at": int(time.time()) + 60, + } + self.wire = {"device_id": "pbgw_test", "command_id": self.command_id, "command": "probe_battery_control", "expires_at": self.control["expires_at"], "control_grant": {"payload": "synthetic", "signature": "synthetic"}} + self.receipt = {**self.control, "phase": "probe_ready", "persisted": True, "pending": 0, "uncertain": 0} + self.ack = {"command_id": self.command_id, "command": "probe_battery_control", "ok": True, "applied": {"control": self.receipt}} + self.calls = [] + + def client(self, messages, **kwargs): + """Fake the socket boundary while running the actual GatewayMQTT method.""" + self.assertEqual(kwargs["password"], "synthetic-token") + self.assertTrue(kwargs["clean_session"]) + calls = self.calls + + class Client: + """Bounded broker adapter with ordered subscription and publish.""" + + async def __aenter__(self): + """Open the synthetic fresh connection.""" + return self + + async def __aexit__(self, *_args): + """Close without retaining or resending any wire payload.""" + + async def subscribe(self, topic, **options): + """Record the exact command-specific ACK subscription.""" + calls.append(("subscribe", topic, options)) + return [0] + + async def publish(self, topic, payload, **options): + """Record the sole permitted signed probe publication.""" + calls.append(("publish", topic, json.loads(payload), options)) + + @property + def messages(self): + """Provide only packets delivered on this synthetic connection.""" + + async def stream(): + """Yield without creating cache or retained discovery authority.""" + for message in messages: + yield message + + return stream() + + return Client() + + def observe(self, ack=None, retained=False, wire=None): + """Execute the actual one-shot transport with a current command reply.""" + message = SimpleNamespace(topic="predbat/devices/pbgw_test/ack/" + self.command_id, payload=json.dumps(ack or self.ack).encode(), retain=retained) + with patch("gateway.aiomqtt.Client", side_effect=lambda **kwargs: self.client([message], **kwargs)): + return asyncio.run(GatewayMQTT.observe_readonly_transport(self.base, wire or self.wire, self.control, 1)) + + def test_exact_probe_receipt_is_derived_from_actual_received_ack(self): + """Subscribe before publishing, with no retained plans or control writes.""" + result = self.observe() + self.assertEqual(result["serials"], ["FD123", "FD456"]) + self.assertEqual(result["receipt"], self.receipt) + self.assertEqual(self.calls[0][0], "subscribe") + self.assertEqual(self.calls[1], ("publish", "predbat/devices/pbgw_test/command", self.wire, {"qos": 1, "retain": False})) + self.assertTrue(self.base.control_write_gate.snapshot()["closed"]) + + def test_retained_replayed_wrong_boot_or_pending_receipts_never_pass(self): + """Neither cached MQTT nor physical uncertainty can become readiness.""" + with self.assertRaises(ValueError): + self.observe(retained=True) + for change in ({"boot_id": "old-boot"}, {"pending": 1}, {"uncertain": 1}, {"inverter_serials": ["FD123"]}, {"expires_at": self.control["expires_at"] + 1}): + ack = copy.deepcopy(self.ack) + ack["applied"]["control"].update(change) + with self.subTest(change=change), self.assertRaises(ValueError): + self.observe(ack=ack) + + def test_mutation_command_or_open_gate_cannot_use_readonly_path(self): + """The dedicated probe route cannot bypass ordinary mutation admission.""" + with self.assertRaises(ValueError): + self.observe(wire={**self.wire, "command": "set_charge_rate"}) + self.base.control_write_gate = ControlWriteGate("target", "epoch", 7, time.monotonic() + 60) + with self.assertRaises(ValueError): + self.observe() + self.assertEqual(self.calls, []) + + +if __name__ == "__main__": + unittest.main()