From a387a13fa27f06a8982bfa84ed8692bb7c4fb6a0 Mon Sep 17 00:00:00 2001 From: Pierre Gironde Date: Sat, 26 Sep 2026 16:43:08 +0200 Subject: [PATCH] fix retain_contiion behavior which use to start the retain timer when the condition is true instead of when its false --- unit_tests/test_retain_condition.py | 97 +++++++++++++++++++++++++++++ virtualsensors.py | 17 +++-- 2 files changed, 109 insertions(+), 5 deletions(-) create mode 100644 unit_tests/test_retain_condition.py diff --git a/unit_tests/test_retain_condition.py b/unit_tests/test_retain_condition.py new file mode 100644 index 0000000..6b19fc5 --- /dev/null +++ b/unit_tests/test_retain_condition.py @@ -0,0 +1,97 @@ +"""Exercise retain timing without requiring a running AppDaemon instance.""" +import ast +import unittest +from enum import Enum +from pathlib import Path +from types import SimpleNamespace +from unittest.mock import Mock + + +class Result(Enum): + Failed = 0 + Succeeded = 1 + Disabled = 2 + + +# Load the actual class while avoiding the module's AppDaemon dependencies. +source = Path(__file__).resolve().parents[1] / 'virtualsensors.py' +node = next(node for node in ast.parse(source.read_text()).body + if isinstance(node, ast.ClassDef) and node.name == 'RetainCondition') +namespace = {'VirtualSensorBase': object, + 'SmartCondition': SimpleNamespace(Result=Result)} +exec(compile(ast.Module(body=[node], type_ignores=[]), str(source), 'exec'), namespace) +RetainCondition = namespace['RetainCondition'] + + +class RetainConditionTests(unittest.TestCase): + def setUp(self): + self.sensor = RetainCondition() + self.sensor.cb_handle = None + self.sensor.retain_time = 15 + self.sensor.reset_retain_time_on_new_success = True + self.state = 'off' + self.sensor.output_sensor = Mock() + self.sensor.output_sensor.get_state.side_effect = lambda: self.state + self.sensor.output_sensor.set_state.side_effect = self.set_state + self.sensor.ad_api = Mock() + self.sensor.ad_api.run_in.return_value = 'timer' + self.sensor.condition = Mock() + self.sensor.condition.evaluate.return_value = Result.Failed + + def set_state(self, *, state): + self.state = state + + def test_long_lived_signal_gets_full_delay_after_loss(self): + self.sensor.on_condition_succeed() + self.assertEqual(self.state, 'on') + self.sensor.ad_api.run_in.assert_not_called() + # No timer can expire while the signal stays valid. + self.sensor.on_condition_fail() + self.assertEqual(self.state, 'on') + self.sensor.ad_api.run_in.assert_called_once_with(self.sensor.on_time_elapse, 15) + self.sensor.on_time_elapse({}) + self.assertEqual(self.state, 'off') + + def test_returning_signal_cancels_off_and_next_loss_gets_new_delay(self): + self.sensor.on_condition_succeed() + self.sensor.on_condition_fail() + self.sensor.on_condition_succeed() + self.sensor.ad_api.cancel_timer.assert_called_once_with('timer') + self.assertIsNone(self.sensor.cb_handle) + self.assertEqual(self.state, 'on') + self.sensor.on_condition_fail() + self.assertEqual(self.sensor.ad_api.run_in.call_count, 2) + self.assertEqual(self.state, 'on') + + def test_repeated_failure_does_not_extend_delay(self): + self.sensor.on_condition_succeed() + self.sensor.on_condition_fail() + self.sensor.on_condition_fail() + self.assertEqual(self.sensor.ad_api.run_in.call_count, 1) + + def test_initially_off_failure_does_not_schedule_timer(self): + self.sensor.on_condition_fail() + self.sensor.ad_api.run_in.assert_not_called() + self.assertEqual(self.state, 'off') + + def test_expiry_rechecks_condition_before_turning_off(self): + for result in (Result.Succeeded, Result.Disabled): + with self.subTest(result=result): + self.state = 'on' + self.sensor.condition.evaluate.return_value = result + self.sensor.on_time_elapse({}) + self.assertEqual(self.state, 'on') + + def test_explicit_no_reset_preserves_fixed_window(self): + self.sensor.reset_retain_time_on_new_success = False + self.sensor.on_condition_succeed() + self.sensor.on_condition_fail() + self.assertEqual(self.state, 'on') + self.sensor.condition.evaluate.return_value = Result.Succeeded + self.sensor.on_time_elapse({}) + self.sensor.on_condition_fail() + self.assertEqual(self.state, 'off') + + +if __name__ == '__main__': + unittest.main() diff --git a/virtualsensors.py b/virtualsensors.py index 7bf54c5..d0cd076 100644 --- a/virtualsensors.py +++ b/virtualsensors.py @@ -480,13 +480,20 @@ class RetainCondition(VirtualSensorBase): def on_condition_succeed(self): if self.output_sensor.get_state() != 'on': self.output_sensor.set_state(state = 'on') - if self.reset_retain_time_on_new_success: - self.cancel_callback() - self.register_callback() + if self.reset_retain_time_on_new_success: + # A returning signal cancels the pending off transition. + self.cancel_callback() + else: + # Preserve the fixed window when resetting is explicitly disabled. + self.register_callback() def on_condition_fail(self): - # we don't want to set the sensor to off during the timer - if self.cb_handle == None: + if self.reset_retain_time_on_new_success: + # Retain for the full delay after the condition becomes false, + # even when it has been true for longer than retain_time. + if self.output_sensor.get_state() == 'on': + self.register_callback() + elif self.cb_handle == None: if self.output_sensor.get_state() != 'off': self.output_sensor.set_state(state = 'off')