From ac661f13c90a2ad251820e93657103d1955a59d3 Mon Sep 17 00:00:00 2001 From: Mark Gascoyne Date: Tue, 13 Oct 2026 02:14:36 +0100 Subject: [PATCH 1/3] fix(gateway): send direct battery charge windows as paired slots --- apps/predbat/gateway.py | 29 +++ apps/predbat/inverter.py | 80 +++++++- apps/predbat/tests/test_gateway.py | 2 + .../tests/test_gateway_charge_window.py | 193 ++++++++++++++++++ 4 files changed, 301 insertions(+), 3 deletions(-) create mode 100644 apps/predbat/tests/test_gateway_charge_window.py diff --git a/apps/predbat/gateway.py b/apps/predbat/gateway.py index d561501d4..255f76e06 100644 --- a/apps/predbat/gateway.py +++ b/apps/predbat/gateway.py @@ -2644,6 +2644,35 @@ async def select_event(self, entity_id, value): # Read current charge slot times to send both start and end await self._update_charge_slot(entity_id, hhmm, serial=serial) + async def update_charge_window(self, start_entity, end_entity, start, end): + """Send an engine charge window as one full-serial command, not two partial slots. + + True means publication was attempted, not that the inverter applied it. The + caller must verify both endpoint entities, which only an applied ACK or + physical telemetry updates. None preserves the existing coordinator path: + those devices can stage a disabled slot and apply it only on enable. + Manual single-endpoint selects keep their path. + """ + if getattr(self.base, "set_read_only", False): + return False + if not isinstance(start_entity, str) or not start_entity.endswith("_charge_slot1_start"): + return False + if end_entity != start_entity.removesuffix("_start") + "_end": + return False + serial = self._serial_from_entity_id(start_entity) + if not serial or self._serial_from_entity_id(end_entity) != serial: + return False + record = getattr(self, "_discovery", {}).get(serial.upper(), {}) + if record.get("type") in GATEWAY_COORDINATOR_TYPES: + return None + try: + times = [datetime.datetime.strptime(value, "%H:%M:%S") for value in (start, end)] + except (ValueError, TypeError): + return False + schedule = {field: value.hour * 100 + value.minute for field, value in zip(("start", "end"), times)} + await self._send_control(start_entity, "set_charge_slot", schedule_json=json.dumps(schedule), serial=serial) + return True + async def _update_charge_slot(self, entity_id, hhmm, serial): """Send set_charge_slot command with updated start or end time.""" # Determine which field changed diff --git a/apps/predbat/inverter.py b/apps/predbat/inverter.py index 604ab2317..6219a1ba0 100644 --- a/apps/predbat/inverter.py +++ b/apps/predbat/inverter.py @@ -3485,6 +3485,74 @@ def adjust_export_immediate(self, target_soc, freeze=False): if not self.call_service_template("discharge_stop_service", service_data_stop, domain="discharge"): self.call_service_template("charge_stop_service", service_data_stop, domain="discharge") + def write_and_poll_gateway_charge_window(self, new_start, new_end): + """Verify a complete Hub slot together, or return None for another provider. + + A start equal to the old end cannot be sent alone while charging is enabled: + it temporarily disables the slot and the Hub correctly refuses it. Sending + both bounds also handles swapping the bounds of a midnight-spanning slot. + """ + from ha import run_async + + entities = [self.base.get_arg(name, indirect=False, index=self.id) for name in ("charge_start_time", "charge_end_time")] + owners = [getattr(self.base, "dashboard_index_app", {}).get(entity) for entity in entities] + if "gateway" not in owners: + return None + components = getattr(self.base, "components", None) + gateway = components.get_component("gateway") if components else None + names = ("charge_start_time", "charge_end_time") + target = (new_start, new_end) + if owners != ["gateway", "gateway"] or gateway is None: + self.base.record_status(f"Warn: Inverter {self.id} Hub charge window has inconsistent entity ownership", had_errors=True) + return False + if not all(self.check_write_entity("write_and_poll_gateway_charge_window", name, entity, value) for name, entity, value in zip(names, entities, target)): + return False + + def read_pair(): + """Read both endpoint entities without substituting desired values.""" + return tuple(self.base.get_state_wrapper(entity, refresh=True) for entity in entities) + + before = read_pair() + ledger = self.base.control_ledger + if ledger is not None: + for name, entity, value, wanted in zip(names, entities, before, target): + self._ledger_observe(ledger, name, entity, value) + if value != wanted: + ledger.note_write_attempt(entity) + key = "|".join(entities) + verified = before == target + for _retry in range(0 if verified else self._write_attempts("charge_window", key, target)): + try: + published = run_async(gateway.update_charge_window(*entities, *target)) + except Exception as error: + self.base.log(f"Warn: Inverter {self.id} Hub charge window publication failed: {error}") + break + if published is None: + return None # Coordinator slot staging must precede its enable write. + if not published: + break + self._poll_after_write(entities[0], lambda state: state == new_start and self.base.get_state_wrapper(entities[1], refresh=True) == new_end) + verified = read_pair() == target + if verified: + self.count_register_writes += 1 + self.registers_moved += sum(old != wanted for old, wanted in zip(before, target)) + break + self._write_backoff_result("charge_window", key, target, verified) + if ledger is not None: + for name, entity, value in zip(names, entities, target): + if verified: + if before == target: + ledger.record_ownership_from_read(entity, name, value, now=time.time(), generation=self._ledger_generation(entity)) + else: + ledger.record_write(entity, name, value, now=time.time(), generation=self._ledger_generation(entity)) + else: + ledger.clear(entity) + if not verified: + message = f"Warn: Inverter {self.id} write to Hub charge window failed: wanted {target}, got {read_pair()}" + self.base.log(message) + self.base.record_status(message, had_errors=True) + return verified + def adjust_charge_window(self, charge_start_time, charge_end_time, minutes_now): """ Adjust the charging window times (start and end) in GivTCP @@ -3556,8 +3624,14 @@ def adjust_charge_window(self, charge_start_time, charge_end_time, minutes_now): # Some inverters have an idle time setting self.adjust_idle_time(charge_start=new_start, charge_end=new_end) + gateway_window = None + if new_start != old_start or new_end != old_end: + gateway_window = self.write_and_poll_gateway_charge_window(new_start, new_end) + if gateway_window is False: + return + # Disable charging if required to avoid a blip while the start/end entities are written - if not in_new_window and ((new_start != old_start) or (new_end != old_end)) and self.inv_has_charge_enable_time: + if gateway_window is None and not in_new_window and ((new_start != old_start) or (new_end != old_end)) and self.inv_has_charge_enable_time: self.disable_charge_window(notify=False) have_disabled = True @@ -3571,7 +3645,7 @@ def adjust_charge_window(self, charge_start_time, charge_end_time, minutes_now): # Registers this cycle's writes actually moved, for commit_needed() below moved_before = self.registers_moved - if new_start != old_start or (self.inv_charge_time_format in ["H M", "H:M-H:M"]): + if gateway_window is None and (new_start != old_start or (self.inv_charge_time_format in ["H M", "H:M-H:M"])): if "charge_start_time" in self.base.args: # Always write to this as it is the GE default entity_id_start = self.base.get_arg("charge_start_time", indirect=False, index=self.id) @@ -3589,7 +3663,7 @@ def adjust_charge_window(self, charge_start_time, charge_end_time, minutes_now): self.log("Warn: Inverter {} unable write charge window start as neither REST or charge_start_time are set".format(self.id)) # Program end slot - if new_end != old_end or (self.inv_charge_time_format in ["H M", "H:M-H:M"]): + if gateway_window is None and (new_end != old_end or (self.inv_charge_time_format in ["H M", "H:M-H:M"])): if "charge_end_time" in self.base.args: # Always write to this as it is the GE default entity_id_end = self.base.get_arg("charge_end_time", indirect=False, index=self.id) diff --git a/apps/predbat/tests/test_gateway.py b/apps/predbat/tests/test_gateway.py index f83d2b08b..cfffa4461 100644 --- a/apps/predbat/tests/test_gateway.py +++ b/apps/predbat/tests/test_gateway.py @@ -7404,8 +7404,10 @@ def test_counters_restart_at_local_midnight(self): def run_gateway_tests(my_predbat=None): """Run all GatewayMQTT tests. Returns True on failure, False on success.""" from tests.test_gateway_token_refresh import TestIsAuthFailure, TestApplyRefreshResponse, TestMaybeRefreshOnAuthError + from tests.test_gateway_charge_window import TestGatewayChargeWindow test_classes = [ + TestGatewayChargeWindow, TestProtobufDecode, TestPlanSerialization, TestCommandFormat, diff --git a/apps/predbat/tests/test_gateway_charge_window.py b/apps/predbat/tests/test_gateway_charge_window.py new file mode 100644 index 000000000..1a959b9a4 --- /dev/null +++ b/apps/predbat/tests/test_gateway_charge_window.py @@ -0,0 +1,193 @@ +"""Joined engine window, Hub JSON and applied-ACK regressions.""" + +import asyncio +import datetime +import json +import types +import unittest +from unittest.mock import MagicMock + +from gateway import GatewayMQTT +from inverter import Inverter +from tests.test_gateway import TestCommandAck + + +class WindowBase: + """Predbat state/service boundary, backed by actual Hub ACK entity updates.""" + + def __init__(self, gateway): + """Configure two owned endpoint entities and capture logs and errors.""" + self.gateway = gateway + self.args = {"charge_start_time": TestCommandAck.SLOT_START, "charge_end_time": TestCommandAck.SLOT_END} + self.dashboard_index_app = {entity: "gateway" for entity in self.args.values()} + self.components = types.SimpleNamespace(get_component=lambda name: gateway if name == "gateway" else None) + self.control_ledger = None + self.set_read_only = False + self.set_inverter_notify = False + self.inverter_clock_skew_start = self.inverter_clock_skew_end = 0 + self.logs = [] + self.errors = [] + + def get_arg(self, name, default=None, index=None, indirect=True): + """Resolve configured endpoint ids or their currently observed values.""" + if name == "scheduled_charge_enable": + return "on" + entity = self.args.get(name) + return self.gateway.cache.get(entity, default) if indirect else entity + + def get_state_wrapper(self, entity_id, **kwargs): + """Expose the actual ACK cache to the engine polling method.""" + return self.gateway.cache.get(entity_id) + + def log(self, message): + """Capture human-readable execution logs.""" + self.logs.append(message) + + def record_status(self, message, **kwargs): + """Capture reported execution errors.""" + self.errors.append(message) + + +class TestGatewayChargeWindow(unittest.TestCase): + """Run real engine methods through serialization and real ACK handling.""" + + def fixture(self, start="00:30:00", end="01:30:00", outcome="applied"): + """Use synthetic MQTT transport while preserving production command/ACK code.""" + gateway = TestCommandAck()._make_gateway() + base = WindowBase(gateway) + gateway.base = base + gateway.topic_command = "predbat/devices/pbgw_test/command" + gateway.publish_command = types.MethodType(GatewayMQTT.publish_command, gateway) + gateway.cache.update({TestCommandAck.SLOT_START: start, TestCommandAck.SLOT_END: end}) + wire = [] + + async def transport(topic, payload, retain=False): + """Capture exact wire and simulate applied, staged, refused or absent ACKs.""" + command = json.loads(payload) + wire.append(command) + self.assertFalse(retain) + if outcome == "timeout": + return + if outcome == "exception": + raise OSError("synthetic MQTT failure") + applied = json.loads(command["schedule_json"]) + if outcome == "partial": + applied["end"] = 130 + if outcome == "staged": + applied["staged"] = True + ack = {"command_id": command["command_id"], "command": command["command"], "ok": outcome != "refused", "applied": applied} + if outcome == "refused": + ack["error"] = "modbus_write_failed" + gateway._process_ack(json.dumps(ack)) + + gateway._publish_raw = transport + inverter = Inverter.__new__(Inverter) + inverter.base = base + inverter.log = base.log + inverter.id = 0 + inverter.count_register_writes = inverter.registers_moved = 0 + inverter.inv_write_backoff = False + inverter.inv_write_max_retry = 3 + inverter.inv_write_and_poll_sleep = 0.01 + inverter.inv_has_charge_enable_time = True + inverter.inv_charge_time_format = "H:M" + inverter.inv_time_button_press = False + inverter.commit_pending = {} + inverter.last_committed = {} + inverter.sleep = lambda seconds: None + inverter.adjust_idle_time = lambda **kwargs: None + inverter.disable_charge_window = MagicMock() + inverter.write_and_poll_option = MagicMock(return_value=True) + inverter.write_and_poll_switch = MagicMock(return_value=True) + return inverter, gateway, base, wire + + def apply(self, inverter, start="01:30:00", end="02:00:00"): + """Run the actual engine charge-window update at 01:30 local time.""" + parse = lambda value: datetime.datetime.strptime("2026-10-13 " + value, "%Y-%m-%d %H:%M:%S") + inverter.adjust_charge_window(parse(start), parse(end), 90) + + def test_zero_length_intermediate_and_midnight_window_use_one_full_pair(self): + """The two live failure shapes now reach the requested complete slot.""" + for old_start in ("00:30:00", "23:30:00"): + inverter, gateway, base, wire = self.fixture(start=old_start) + self.apply(inverter) + self.assertEqual(len(wire), 1) + self.assertEqual(json.loads(wire[0]["schedule_json"]), {"start": 130, "end": 200}) + self.assertEqual(wire[0]["serial"], "CE123456789") + self.assertEqual(gateway.cache[TestCommandAck.SLOT_START], "01:30:00") + self.assertEqual(gateway.cache[TestCommandAck.SLOT_END], "02:00:00") + inverter.write_and_poll_option.assert_not_called() + inverter.disable_charge_window.assert_not_called() + self.assertEqual(base.errors, []) + + def test_swap_of_endpoints_does_not_need_an_invalid_intermediate(self): + """Either single-endpoint ordering would disable this slot temporarily.""" + inverter, gateway, base, wire = self.fixture(start="02:00:00", end="01:30:00") + self.apply(inverter) + self.assertEqual(json.loads(wire[0]["schedule_json"]), {"start": 130, "end": 200}) + self.assertEqual(base.errors, []) + + def test_unverified_pair_never_logs_success_or_enables_window(self): + """Refused, staged, wrong-end and missing ACKs cannot confirm the pair.""" + for outcome in ("refused", "staged", "partial", "timeout", "exception"): + inverter, gateway, base, wire = self.fixture(outcome=outcome) + base.control_ledger = MagicMock() + self.apply(inverter) + self.assertTrue(base.errors, outcome) + self.assertFalse(any("Updated start and end" in line for line in base.logs), outcome) + inverter.write_and_poll_switch.assert_not_called() + inverter.write_and_poll_option.assert_not_called() + self.assertEqual(base.control_ledger.clear.call_count, 2, outcome) + base.control_ledger.record_write.assert_not_called() + self.assertEqual(len(wire), 1, outcome) + + def test_success_records_both_controls_and_no_redundant_second_command(self): + """Applied pair owns both endpoints and the next unchanged cycle is quiet.""" + inverter, gateway, base, wire = self.fixture() + base.control_ledger = MagicMock() + self.apply(inverter) + self.assertEqual(base.control_ledger.record_write.call_count, 2) + self.apply(inverter) + self.assertEqual(len(wire), 1) + + def test_read_only_or_mismatched_entity_ownership_refuses_without_publication(self): + """Never route a Cloud endpoint or user read-only window to the Hub.""" + for reason in ("read_only", "owner", "serial"): + inverter, gateway, base, wire = self.fixture() + if reason == "read_only": + base.set_read_only = True + elif reason == "owner": + base.dashboard_index_app[TestCommandAck.SLOT_END] = "gecloud" + else: + base.args["charge_end_time"] = TestCommandAck.SLOT_END.replace("456789", "654321") + base.dashboard_index_app[base.args["charge_end_time"]] = "gateway" + self.apply(inverter) + self.assertEqual(wire, [], reason) + self.assertTrue(base.errors, reason) + + def test_other_provider_keeps_individual_entity_writes(self): + """Cloud and other integrations retain their existing update path.""" + inverter, gateway, base, wire = self.fixture() + base.dashboard_index_app = {entity: "gecloud" for entity in base.args.values()} + self.apply(inverter) + self.assertEqual(wire, []) + self.assertEqual(inverter.write_and_poll_option.call_count, 2) + + def test_coordinator_slot_staging_keeps_existing_enable_sequence(self): + """EMS and Gateway coordinators keep their staged-slot update path.""" + from gateway import GATEWAY_COORDINATOR_TYPES + + for kind in GATEWAY_COORDINATOR_TYPES: + inverter, gateway, base, wire = self.fixture() + gateway._discovery = {"CE123456789": {"type": kind}} + self.apply(inverter) + self.assertEqual(wire, []) + self.assertEqual(inverter.write_and_poll_option.call_count, 2) + self.assertEqual(base.errors, []) + + def test_invalid_time_or_pair_is_rejected_before_command(self): + """The public paired method validates inputs before calling the command sink.""" + inverter, gateway, base, wire = self.fixture() + for start, end in (("25:00:00", "02:00:00"), (None, "02:00:00")): + self.assertFalse(asyncio.run(gateway.update_charge_window(TestCommandAck.SLOT_START, TestCommandAck.SLOT_END, start, end))) + self.assertEqual(wire, []) From 08d96daf1a071e67eda05f2283e7fa67d00eab3d Mon Sep 17 00:00:00 2001 From: Mark Gascoyne Date: Tue, 13 Oct 2026 02:18:53 +0100 Subject: [PATCH 2/3] fix(gateway): separate coordinator staging from async publication failure --- apps/predbat/gateway.py | 17 ++++++++++++---- apps/predbat/inverter.py | 6 +++--- .../tests/test_gateway_charge_window.py | 20 ++++++++++++++++++- 3 files changed, 35 insertions(+), 8 deletions(-) diff --git a/apps/predbat/gateway.py b/apps/predbat/gateway.py index 255f76e06..16909ec46 100644 --- a/apps/predbat/gateway.py +++ b/apps/predbat/gateway.py @@ -2644,12 +2644,22 @@ async def select_event(self, entity_id, value): # Read current charge slot times to send both start and end await self._update_charge_slot(entity_id, hhmm, serial=serial) + def charge_window_uses_staging(self, start_entity, end_entity): + """Identify a coordinator synchronously, before crossing an async bridge.""" + if not isinstance(start_entity, str) or not start_entity.endswith("_charge_slot1_start"): + return False + if end_entity != start_entity.removesuffix("_start") + "_end": + return False + serial = self._serial_from_entity_id(start_entity) + record = getattr(self, "_discovery", {}).get((serial or "").upper(), {}) + return bool(serial) and record.get("type") in GATEWAY_COORDINATOR_TYPES + async def update_charge_window(self, start_entity, end_entity, start, end): """Send an engine charge window as one full-serial command, not two partial slots. True means publication was attempted, not that the inverter applied it. The caller must verify both endpoint entities, which only an applied ACK or - physical telemetry updates. None preserves the existing coordinator path: + physical telemetry updates. Coordinators use their existing separate path: those devices can stage a disabled slot and apply it only on enable. Manual single-endpoint selects keep their path. """ @@ -2662,9 +2672,8 @@ async def update_charge_window(self, start_entity, end_entity, start, end): serial = self._serial_from_entity_id(start_entity) if not serial or self._serial_from_entity_id(end_entity) != serial: return False - record = getattr(self, "_discovery", {}).get(serial.upper(), {}) - if record.get("type") in GATEWAY_COORDINATOR_TYPES: - return None + if self.charge_window_uses_staging(start_entity, end_entity): + return False try: times = [datetime.datetime.strptime(value, "%H:%M:%S") for value in (start, end)] except (ValueError, TypeError): diff --git a/apps/predbat/inverter.py b/apps/predbat/inverter.py index 6219a1ba0..c05172fcf 100644 --- a/apps/predbat/inverter.py +++ b/apps/predbat/inverter.py @@ -3507,6 +3507,8 @@ def write_and_poll_gateway_charge_window(self, new_start, new_end): return False if not all(self.check_write_entity("write_and_poll_gateway_charge_window", name, entity, value) for name, entity, value in zip(names, entities, target)): return False + if gateway.charge_window_uses_staging(*entities): + return None # Coordinator slot staging must precede its enable write. def read_pair(): """Read both endpoint entities without substituting desired values.""" @@ -3527,9 +3529,7 @@ def read_pair(): except Exception as error: self.base.log(f"Warn: Inverter {self.id} Hub charge window publication failed: {error}") break - if published is None: - return None # Coordinator slot staging must precede its enable write. - if not published: + if published is not True: break self._poll_after_write(entities[0], lambda state: state == new_start and self.base.get_state_wrapper(entities[1], refresh=True) == new_end) verified = read_pair() == target diff --git a/apps/predbat/tests/test_gateway_charge_window.py b/apps/predbat/tests/test_gateway_charge_window.py index 1a959b9a4..7bb473571 100644 --- a/apps/predbat/tests/test_gateway_charge_window.py +++ b/apps/predbat/tests/test_gateway_charge_window.py @@ -1,11 +1,12 @@ """Joined engine window, Hub JSON and applied-ACK regressions.""" +# cspell:ignore excepthook import asyncio import datetime import json import types import unittest -from unittest.mock import MagicMock +from unittest.mock import MagicMock, patch from gateway import GatewayMQTT from inverter import Inverter @@ -150,6 +151,23 @@ def test_success_records_both_controls_and_no_redundant_second_command(self): self.apply(inverter) self.assertEqual(len(wire), 1) + def test_async_timer_publication_exception_never_becomes_coordinator_fallback(self): + """RunThread returns None on error; that must remain a failed direct write.""" + inverter, gateway, base, wire = self.fixture(outcome="exception") + + async def timer(): + """Match the standalone timer calling synchronous execution on its loop.""" + self.apply(inverter) + + with patch("threading.excepthook") as thread_error: + asyncio.run(timer()) + thread_error.assert_called_once() + self.assertEqual(len(wire), 1) + self.assertTrue(base.errors) + inverter.write_and_poll_option.assert_not_called() + inverter.write_and_poll_switch.assert_not_called() + self.assertFalse(any("Updated start and end" in line for line in base.logs)) + def test_read_only_or_mismatched_entity_ownership_refuses_without_publication(self): """Never route a Cloud endpoint or user read-only window to the Hub.""" for reason in ("read_only", "owner", "serial"): From 1993cfc3abf4b5aa61ab35681f371c3c8bf82b3e Mon Sep 17 00:00:00 2001 From: CI Date: Wed, 14 Oct 2026 08:10:09 +0100 Subject: [PATCH 3/3] fix(gateway): detect coordinator staging before charge ownership checks A coordinator whose charge end entity was not (yet) owned by the gateway was refused with an ownership error, so its charge window was never written, instead of keeping its staged separate-endpoint sequence. Identify coordinators first, as that decision only needs the entity names and discovery record. Same ordering as the discharge side in #5532. Co-Authored-By: Claude Opus 5.5 --- apps/predbat/inverter.py | 4 ++-- apps/predbat/tests/test_gateway_charge_window.py | 13 +++++++++++++ 2 files changed, 15 insertions(+), 2 deletions(-) diff --git a/apps/predbat/inverter.py b/apps/predbat/inverter.py index e1fc66c0e..6f2422fc1 100644 --- a/apps/predbat/inverter.py +++ b/apps/predbat/inverter.py @@ -3638,13 +3638,13 @@ def write_and_poll_gateway_charge_window(self, new_start, new_end): gateway = components.get_component("gateway") if components else None names = ("charge_start_time", "charge_end_time") target = (new_start, new_end) + if gateway is not None and gateway.charge_window_uses_staging(*entities): + return None # Coordinator slot staging must precede its enable write. if owners != ["gateway", "gateway"] or gateway is None: self.base.record_status(f"Warn: Inverter {self.id} Hub charge window has inconsistent entity ownership", had_errors=True) return False if not all(self.check_write_entity("write_and_poll_gateway_charge_window", name, entity, value) for name, entity, value in zip(names, entities, target)): return False - if gateway.charge_window_uses_staging(*entities): - return None # Coordinator slot staging must precede its enable write. def read_pair(): """Read both endpoint entities without substituting desired values.""" diff --git a/apps/predbat/tests/test_gateway_charge_window.py b/apps/predbat/tests/test_gateway_charge_window.py index 7bb473571..6472d2980 100644 --- a/apps/predbat/tests/test_gateway_charge_window.py +++ b/apps/predbat/tests/test_gateway_charge_window.py @@ -203,6 +203,19 @@ def test_coordinator_slot_staging_keeps_existing_enable_sequence(self): self.assertEqual(inverter.write_and_poll_option.call_count, 2) self.assertEqual(base.errors, []) + def test_coordinator_with_partial_ownership_keeps_staged_sequence(self): + """A coordinator is identified before ownership checks, so it is never refused by them.""" + from gateway import GATEWAY_COORDINATOR_TYPES + + for kind in GATEWAY_COORDINATOR_TYPES: + inverter, gateway, base, wire = self.fixture() + gateway._discovery = {"CE123456789": {"type": kind}} + base.dashboard_index_app[TestCommandAck.SLOT_END] = None + self.apply(inverter) + self.assertFalse(any("inconsistent entity ownership" in error for error in base.errors), kind) + self.assertEqual(wire, [], kind) + self.assertEqual(inverter.write_and_poll_option.call_count, 2, kind) + def test_invalid_time_or_pair_is_rejected_before_command(self): """The public paired method validates inputs before calling the command sink.""" inverter, gateway, base, wire = self.fixture()