Skip to content

fix(predbat): don't read db_primary off a None ha_interface in update_time_loop #5137

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
springfall2008 merged 1 commit into from
Sep 19, 2026
Merged

fix(predbat): don't read db_primary off a None ha_interface in update_time_loop #5137

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 6 additions & 1 deletion apps/predbat/predbat.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2098,7 +2098,12 @@ def update_time_loop(self, cb_args):
Called every 15 seconds
"""
if not self.ha_interface or (not self.ha_interface.websocket_active and not self.ha_interface.db_primary):
self.log("Error: HA interface not active and db_primary is {}".format(self.ha_interface.db_primary))
# Only report db_primary when there is an interface to read it from - when ha_interface is None
# reading it here raised AttributeError before fatal_error could be set below (#5135)
if self.ha_interface:

@springfall2008 springfall2008 Sep 19, 2026

Copy link
Copy Markdown
Owner Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This fix now leaves three hand-maintained variants of the same guard semantics: this two-branch log here, run_time_loop's shorter copy at predbat.py:2175 (different wording, no db_primary), and the None-branch message chosen to match it. The PR body documents why it didn't dedupe, and the new test pins both copies so divergence is caught — but any future change to the guard's contract (new attribute in the message, different flag, changed exception) still has to be made in two places. If another guard tweak lands, consider extracting one _check_ha_interface() helper returning a reason string that both loops log and raise from.

self.log("Error: HA interface not active and db_primary is {}".format(self.ha_interface.db_primary))
else:
self.log("Error: HA interface not active")
self.fatal_error = True
raise Exception("HA interface not active")

Expand Down
56 changes: 56 additions & 0 deletions apps/predbat/tests/test_time_loop_guard.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,56 @@
# -----------------------------------------------------------------------------
# Predbat Home Battery System
# Copyright Trefor Southwell 2026 - All Rights Reserved
# This application maybe used for personal use only and not for commercial use
# -----------------------------------------------------------------------------
# fmt off
# pylint: disable=consider-using-f-string
# pylint: disable=line-too-long
# pylint: disable=attribute-defined-outside-init


def _run_guard(my_predbat, loop):

@springfall2008 springfall2008 Sep 19, 2026

Copy link
Copy Markdown
Owner Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Minor: my_predbat is unused here — the loops are already passed as bound methods, so def _run_guard(loop) does the same job. (Same shape as _run_loop_expecting_failure in test_active_flag.py, so this is only worth tidying if that helper is touched too.)

"""Run a run-loop callback that is expected to refuse to run, returning the exception it raised."""
try:
loop(None)
except Exception as e:
return e
return None


def test_time_loop_guard(my_predbat):
"""
Verify the run-loop HA interface guards report the missing interface and set fatal_error when
ha_interface is None. update_time_loop's log line used to read ha_interface.db_primary
unconditionally, so the one case the guard exists for raised AttributeError before
fatal_error could be set, leaving Predbat running with no way to stop it (#5135).
"""
print("*** Running test: Run-loop guard with no HA interface")
failed = 0

for loop_name in ("update_time_loop", "run_time_loop"):
# Each test gets its own PredBat, so clearing the interface here cannot reach another test
my_predbat.ha_interface = None

@springfall2008 springfall2008 Sep 19, 2026

Copy link
Copy Markdown
Owner Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Coverage gap: this test only drives the ha_interface is None case, but the only guard branch reachable in production is the other one — interface present, websocket_active False and db_primary False (the websocket disconnect paths set websocket_active = False at ha.py:667/785, and db-primary defaults off). No existing test covers that branch either: test_active_flag sets websocket_active = True so the guard passes, and test_predheat's timer_tick drive (test_predheat.py:136) runs PredHeat's same-named update_time_loop, not PredBat's. A future edit to the condition itself (e.g. dropping the db_primary term, or swapping and/or so a db-primary install with the websocket down now trips the guard) would pass this suite. Worth one more loop pass with ha_interface kept, websocket_active = False, db_primary = False asserting the same raise + fatal_error.

my_predbat.fatal_error = False

error = _run_guard(my_predbat, getattr(my_predbat, loop_name))

if isinstance(error, AttributeError):
print("ERROR: {} raised AttributeError ({}) rather than reporting the missing HA interface".format(loop_name, error))
failed = 1
elif error is None:
print("ERROR: {} did not raise when ha_interface is None".format(loop_name))
failed = 1
elif "HA interface not active" not in str(error):
print("ERROR: {} raised an unexpected exception: {}".format(loop_name, error))
failed = 1
else:
print("OK: {} reported the missing HA interface".format(loop_name))

if not my_predbat.fatal_error:
print("ERROR: {} did not set fatal_error when ha_interface is None".format(loop_name))
failed = 1
else:
print("OK: {} set fatal_error when ha_interface is None".format(loop_name))

return failed
2 changes: 2 additions & 0 deletions apps/predbat/unit_test.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -66,6 +66,7 @@
from tests.test_battery_accuracy import run_battery_accuracy_tests
from tests.test_plugin_startup import test_plugin_startup_order
from tests.test_active_flag import test_active_flag
from tests.test_time_loop_guard import test_time_loop_guard
from tests.test_component_health_status import test_component_health_status, test_record_status_state_clamped
from tests.test_optimise_levels import run_optimise_levels_tests
from tests.test_trim_export import run_trim_export_tests
Expand Down Expand Up @@ -502,6 +503,7 @@ def main():
("octopus_slots_change", test_octopus_slots_change, "Octopus slots change-detection signature tests (in-progress re-clock vs genuine change)", False),
("plugin_startup", test_plugin_startup_order, "Plugin startup order tests", False),
("active_flag", test_active_flag, "Active flag cleared on exception tests", False),
("time_loop_guard", test_time_loop_guard, "Run-loop guard reports a missing HA interface and sets fatal_error (#5135)", False),
("component_health_status", test_component_health_status, "Component errors fail the recorded run status tests", False),
("record_status_state_clamped", test_record_status_state_clamped, "Status sensor state is clamped at the 255 characters Home Assistant accepts", False),
("dynamic_load_car", test_dynamic_load_car_slot_cancellation, "Dynamic load car slot cancellation tests", False),
Expand Down
Loading