diff --git a/unit_tests/test_continuous_condition.py b/unit_tests/test_continuous_condition.py new file mode 100644 index 0000000..def29c4 --- /dev/null +++ b/unit_tests/test_continuous_condition.py @@ -0,0 +1,121 @@ +"""Exercise timer expiry/cancellation ordering without a running AppDaemon.""" +import ast +import unittest +from pathlib import Path +from types import SimpleNamespace +from unittest.mock import Mock + + +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 == 'ContinuousCondition') +namespace = {'VirtualSensorBase': object, 'SmartCondition': Mock()} +exec(compile(ast.Module(body=[node], type_ignores=[]), str(source), 'exec'), namespace) +ContinuousCondition = namespace['ContinuousCondition'] + + +class FakeScheduler: + def __init__(self): + self.timers = {} + self.next_id = 0 + self.warnings = [] + + def run_in(self, callback, delay, **kwargs): + self.next_id += 1 + handle = str(self.next_id) + self.timers[handle] = (callback, kwargs) + return handle + + def expire(self, handle): + # AppDaemon removes the timer while its callback can still be queued. + callback, kwargs = self.timers.pop(handle) + return lambda: callback(kwargs) + + def cancel_timer(self, handle, silent=False): + if handle in self.timers: + del self.timers[handle] + return True + if not silent: + self.warnings.append(handle) + return False + + +class ContinuousConditionTests(unittest.TestCase): + def setUp(self): + self.scheduler = FakeScheduler() + self.state = 'off' + entity = Mock() + entity.get_state.side_effect = lambda: self.state + entity.set_state.side_effect = lambda *, state: setattr(self, 'state', state) + self.sensor = ContinuousCondition() + self.sensor.ad_api = SimpleNamespace( + get_entity=lambda name: entity, + run_in=self.scheduler.run_in, + cancel_timer=self.scheduler.cancel_timer, + ) + self.sensor.yaml_block = {'time': 5, 'conditions': 'paused'} + self.sensor.sensor_name = 'binary_sensor.is_vertex2_content_idle' + self.sensor.log_info = Mock() + self.sensor.logger_interface = None + self.sensor.constants = {} + self.sensor.templates_library = None + self.sensor.initialize() + + def test_valid_expiry_sets_state_and_clears_handle(self): + self.sensor.on_condition_succeed() + self.scheduler.expire(self.sensor.cb_handle)() + self.assertEqual(self.state, 'on') + self.assertIsNone(self.sensor.cb_handle) + self.sensor.on_condition_fail() + self.assertEqual(self.state, 'off') + + def test_failure_cancels_running_timer(self): + self.sensor.on_condition_succeed() + self.sensor.on_condition_fail() + self.assertEqual(self.scheduler.timers, {}) + self.assertIsNone(self.sensor.cb_handle) + self.assertEqual(self.state, 'off') + + def test_resume_at_expiry_ignores_queued_callback_without_warning(self): + self.sensor.on_condition_succeed() + queued = self.scheduler.expire(self.sensor.cb_handle) + self.sensor.on_condition_fail() + queued() + self.assertEqual(self.state, 'off') + self.assertEqual(self.scheduler.warnings, []) + self.assertIsNone(self.sensor.cb_handle) + + def test_old_callback_preserves_new_timer_and_full_delay(self): + self.sensor.on_condition_succeed() + queued = self.scheduler.expire(self.sensor.cb_handle) + self.sensor.on_condition_fail() + self.sensor.on_condition_succeed() + new_handle = self.sensor.cb_handle + queued() + self.assertEqual(self.state, 'off') + self.assertEqual(self.sensor.cb_handle, new_handle) + self.scheduler.expire(new_handle)() + self.assertEqual(self.state, 'on') + + def test_repeated_success_keeps_existing_timer(self): + self.sensor.on_condition_succeed() + handle = self.sensor.cb_handle + self.sensor.on_condition_succeed() + self.assertEqual(self.sensor.cb_handle, handle) + self.assertEqual(len(self.scheduler.timers), 1) + + def test_inverted_condition_also_ignores_stale_expiry(self): + self.sensor.invert = True + self.state = 'on' + self.sensor.on_condition_succeed() + queued = self.scheduler.expire(self.sensor.cb_handle) + self.sensor.on_condition_fail() + queued() + self.assertEqual(self.state, 'on') + self.sensor.on_condition_succeed() + self.scheduler.expire(self.sensor.cb_handle)() + self.assertEqual(self.state, 'off') + + +if __name__ == '__main__': + unittest.main() diff --git a/unit_tests/test_ephemeral_condition.py b/unit_tests/test_ephemeral_condition.py new file mode 100644 index 0000000..94342f1 --- /dev/null +++ b/unit_tests/test_ephemeral_condition.py @@ -0,0 +1,105 @@ +"""Check initial windows, rearming and queued timer races.""" +import ast +import unittest +from unittest.mock import Mock, patch +import time + +from test_continuous_condition import ContinuousConditionTests, namespace, source + +namespace['time'] = time +node = next(node for node in ast.parse(source.read_text()).body + if isinstance(node, ast.ClassDef) and node.name == 'EphemeralCondition') +exec(compile(ast.Module(body=[node], type_ignores=[]), str(source), 'exec'), namespace) +EphemeralCondition = namespace['EphemeralCondition'] + + +class EphemeralConditionTests(unittest.TestCase): + def setUp(self): + ContinuousConditionTests.setUp(self) + self.sensor.__class__ = EphemeralCondition + self.sensor.attributes = {} + self.sensor.ad_api.get_state = Mock(return_value=None) + self.sensor.output_sensor.set_state.side_effect = lambda *, state, **kwargs: setattr(self, 'state', state) + self.sensor.initialize() + + def test_window_expires_without_restarting_while_true(self): + self.sensor.on_condition_succeed() + self.assertEqual(self.state, 'on') + handle = self.sensor.cb_handle + self.sensor.on_condition_succeed() + self.assertEqual(self.sensor.cb_handle, handle) + self.scheduler.expire(handle)() + self.assertEqual(self.state, 'off') + self.sensor.on_condition_succeed() + self.assertEqual(self.state, 'off') + self.assertIsNone(self.sensor.cb_handle) + + def test_failure_ends_window_and_rearms(self): + self.sensor.on_condition_succeed() + self.sensor.on_condition_fail() + self.assertEqual(self.state, 'off') + self.assertEqual(self.scheduler.timers, {}) + self.sensor.on_condition_succeed() + self.assertEqual(self.state, 'on') + self.scheduler.expire(self.sensor.cb_handle)() + self.assertEqual(self.state, 'off') + + def test_queued_expiry_cannot_end_new_window(self): + self.sensor.on_condition_succeed() + queued = self.scheduler.expire(self.sensor.cb_handle) + self.sensor.on_condition_fail() + self.sensor.on_condition_succeed() + handle = self.sensor.cb_handle + queued() + self.assertEqual(self.state, 'on') + self.assertEqual(self.sensor.cb_handle, handle) + self.scheduler.expire(handle)() + self.assertEqual(self.state, 'off') + + def test_restart_resumes_remaining_time(self): + self.sensor.ad_api.get_state.return_value = 100 + with patch.object(time, 'time', return_value=103): + self.sensor.initialize() + self.sensor.ad_api.run_in = Mock(return_value='restored') + self.sensor.on_condition_succeed() + self.assertEqual(self.state, 'on') + self.assertEqual(self.sensor.ad_api.run_in.call_args.args[1], 2) + self.assertEqual(self.sensor.attributes['condition_started_at'], 100) + + def test_restart_does_not_rearm_expired_window(self): + self.sensor.ad_api.get_state.return_value = 100 + with patch.object(time, 'time', return_value=110): + self.sensor.initialize() + self.sensor.on_condition_succeed() + self.assertEqual(self.state, 'off') + self.assertIsNone(self.sensor.cb_handle) + self.sensor.on_condition_fail() + self.assertIsNone(self.sensor.attributes['condition_started_at']) + self.sensor.on_condition_succeed() + self.assertEqual(self.state, 'on') + + def test_invalid_restored_timestamp_starts_fresh(self): + for value in ('invalid', float('nan'), float('inf'), -1, 200): + with self.subTest(value=value), patch.object(time, 'time', return_value=100): + self.sensor.ad_api.get_state.return_value = value + self.sensor.initialize() + self.sensor.on_condition_succeed() + self.assertEqual(self.state, 'on') + self.assertEqual(self.sensor.attributes['condition_started_at'], 100) + self.sensor.on_condition_fail() + + def test_initial_true_condition_starts_window(self): + evaluator = namespace['SmartCondition'].Evaluator + def initialize_condition(*args, **kwargs): + kwargs['on_succeed_cb']() + evaluator.side_effect = initialize_condition + try: + self.sensor.initialize() + self.assertEqual(self.state, 'on') + self.assertIsNotNone(self.sensor.cb_handle) + finally: + evaluator.side_effect = None + + +if __name__ == '__main__': + unittest.main() diff --git a/unit_tests/test_retain_condition.py b/unit_tests/test_retain_condition.py index 6b19fc5..a330a7b 100644 --- a/unit_tests/test_retain_condition.py +++ b/unit_tests/test_retain_condition.py @@ -27,6 +27,7 @@ class RetainConditionTests(unittest.TestCase): def setUp(self): self.sensor = RetainCondition() self.sensor.cb_handle = None + self.sensor._timer_generation = 0 self.sensor.retain_time = 15 self.sensor.reset_retain_time_on_new_success = True self.state = 'off' @@ -41,6 +42,54 @@ class RetainConditionTests(unittest.TestCase): def set_state(self, *, state): self.state = state + def queued_expiry(self): + call = self.sensor.ad_api.run_in.call_args + callback = call.args[0] + kwargs = dict(call.kwargs) + return lambda: callback(kwargs) + + def test_return_at_expiry_ignores_queued_callback(self): + self.sensor.on_condition_succeed() + self.sensor.on_condition_fail() + queued = self.queued_expiry() + # The scheduler has already removed the timer and queued its callback. + self.sensor.ad_api.cancel_timer.return_value = False + self.sensor.condition.evaluate.return_value = Result.Succeeded + self.sensor.on_condition_succeed() + queued() + self.assertEqual(self.state, 'on') + self.sensor.condition.evaluate.assert_not_called() + self.sensor.ad_api.cancel_timer.assert_called_once_with('timer', silent=True) + + def test_stale_expiry_cannot_shorten_new_retention(self): + self.sensor.ad_api.run_in.side_effect = ['old-timer', 'new-timer'] + self.sensor.on_condition_succeed() + self.sensor.on_condition_fail() + queued = self.queued_expiry() + self.sensor.on_condition_succeed() + self.sensor.on_condition_fail() + current_expiry = self.queued_expiry() + queued() + self.assertEqual(self.state, 'on') + self.assertEqual(self.sensor.cb_handle, 'new-timer') + current_expiry() + self.assertEqual(self.state, 'off') + self.assertIsNone(self.sensor.cb_handle) + + def test_stale_expiry_cannot_forget_timer_while_condition_disabled(self): + self.sensor.ad_api.run_in.side_effect = ['old-timer', 'new-timer'] + self.sensor.on_condition_succeed() + self.sensor.on_condition_fail() + queued = self.queued_expiry() + self.sensor.on_condition_succeed() + self.sensor.on_condition_fail() + self.sensor.condition.evaluate.return_value = Result.Disabled + queued() + self.assertEqual(self.sensor.cb_handle, 'new-timer') + self.assertEqual(self.state, 'on') + self.sensor.on_condition_succeed() + self.assertEqual(self.sensor.ad_api.cancel_timer.call_args.args, ('new-timer',)) + def test_long_lived_signal_gets_full_delay_after_loss(self): self.sensor.on_condition_succeed() self.assertEqual(self.state, 'on') @@ -48,15 +97,16 @@ class RetainConditionTests(unittest.TestCase): # 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.sensor.ad_api.run_in.assert_called_once_with( + self.sensor.on_time_elapse, 15, generation=self.sensor._timer_generation) + self.queued_expiry()() 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.sensor.ad_api.cancel_timer.assert_called_once_with('timer', silent=True) self.assertIsNone(self.sensor.cb_handle) self.assertEqual(self.state, 'on') self.sensor.on_condition_fail() @@ -79,8 +129,11 @@ class RetainConditionTests(unittest.TestCase): with self.subTest(result=result): self.state = 'on' self.sensor.condition.evaluate.return_value = result - self.sensor.on_time_elapse({}) + self.sensor.register_callback() + self.queued_expiry()() self.assertEqual(self.state, 'on') + self.assertIsNone(self.sensor.cb_handle) + self.sensor.condition.evaluate.assert_called_with() def test_explicit_no_reset_preserves_fixed_window(self): self.sensor.reset_retain_time_on_new_success = False @@ -88,7 +141,7 @@ class RetainConditionTests(unittest.TestCase): self.sensor.on_condition_fail() self.assertEqual(self.state, 'on') self.sensor.condition.evaluate.return_value = Result.Succeeded - self.sensor.on_time_elapse({}) + self.queued_expiry()() self.sensor.on_condition_fail() self.assertEqual(self.state, 'off') diff --git a/virtualsensors.py b/virtualsensors.py index e6996e4..cbb5421 100644 --- a/virtualsensors.py +++ b/virtualsensors.py @@ -406,6 +406,7 @@ class ValueSelector(VirtualSensorBase): class ContinuousCondition(VirtualSensorBase): def initialize(self): self.cb_handle = None + self._timer_generation = 0 self.time = self.yaml_block['time'] try: self.ignore_unavailable_condition = self.yaml_block['ignore_unavailable_condition'] except KeyError: self.ignore_unavailable_condition = False @@ -452,14 +453,22 @@ class ContinuousCondition(VirtualSensorBase): def register_callback(self): if self.cb_handle == None: - self.cb_handle = self.ad_api.run_in(self.on_time_elapse, self.time) + self._timer_generation += 1 + self.cb_handle = self.ad_api.run_in( + self.on_time_elapse, self.time, generation=self._timer_generation) def cancel_callback(self): - if self.cb_handle != None: - self.ad_api.cancel_timer(self.cb_handle) - self.cb_handle = None + # An expired timer may already have a callback queued in the worker. + # Invalidate it before cancellation, even if it is no longer scheduled. + self._timer_generation += 1 + handle = self.cb_handle + self.cb_handle = None + if handle is not None: + self.ad_api.cancel_timer(handle, silent=True) def on_time_elapse(self, kwargs): + if kwargs.get('generation') != self._timer_generation: + return self.cb_handle = None value = 'off' if self.invert else 'on' if self.output_sensor.get_state() != value: @@ -474,9 +483,68 @@ class ContinuousCondition(VirtualSensorBase): if self.output_sensor.get_state() != value: self.output_sensor.set_state(state = value) +class EphemeralCondition(ContinuousCondition): + """True during the first `time` seconds of each true condition. + + condition_started_at is a Unix timestamp retained after expiry, until + the condition becomes false. Reloads resume this window when the entity + attributes survive; without a timestamp, a true condition starts anew. + """ + + def initialize(self): + self._condition_active = False + if self.yaml_block.get('invert', False): + raise ValueError("ephemeral_condition does not support invert; negate conditions instead") + started_at = self.ad_api.get_state( + self.sensor_name, attribute='condition_started_at') + try: + started_at = float(started_at) + if not 0 <= started_at <= time.time(): + started_at = None + except (TypeError, ValueError): + started_at = None + self._started_at = started_at + self.attributes['condition_started_at'] = started_at + # Unknown inputs during startup must not erase a restored window. + self.yaml_block = dict(self.yaml_block) + self.yaml_block.setdefault('ignore_unavailable_condition', True) + super().initialize() + + def on_condition_succeed(self): + if self._condition_active: + return + self._condition_active = True + if self._started_at is None: + self._started_at = time.time() + self.attributes['condition_started_at'] = self._started_at + remaining = self._started_at + self.time - time.time() + self.output_sensor.set_state( + state='on' if remaining > 0 else 'off', + attributes={'condition_started_at': self._started_at}) + if remaining > 0: + self._timer_generation += 1 + self.cb_handle = self.ad_api.run_in( + self.on_time_elapse, remaining, generation=self._timer_generation) + + def on_condition_fail(self): + self._condition_active = False + self._started_at = None + self.attributes['condition_started_at'] = None + self.cancel_callback() + self.output_sensor.set_state( + state='off', attributes={'condition_started_at': None}) + + def on_time_elapse(self, kwargs): + if kwargs.get('generation') != self._timer_generation: + return + self.cb_handle = None + self.output_sensor.set_state(state='off') + + class RetainCondition(VirtualSensorBase): def initialize(self): self.cb_handle = None + self._timer_generation = 0 self.retain_time = self.yaml_block['retain_time'] try: self.reset_retain_time_on_new_success = self.yaml_block['reset_retain_time_on_new_success'] except KeyError: self.reset_retain_time_on_new_success = True @@ -510,14 +578,21 @@ class RetainCondition(VirtualSensorBase): def register_callback(self): if self.cb_handle == None: - self.cb_handle = self.ad_api.run_in(self.on_time_elapse, self.retain_time) + self._timer_generation += 1 + self.cb_handle = self.ad_api.run_in( + self.on_time_elapse, self.retain_time, generation=self._timer_generation) def cancel_callback(self): - if self.cb_handle != None: - self.ad_api.cancel_timer(self.cb_handle) - self.cb_handle = None + # Invalidate callbacks already queued by an expired timer as well. + self._timer_generation += 1 + handle = self.cb_handle + self.cb_handle = None + if handle is not None: + self.ad_api.cancel_timer(handle, silent=True) def on_time_elapse(self, kwargs): + if kwargs.get('generation') != self._timer_generation: + return self.cb_handle = None if self.condition.evaluate() == SmartCondition.Result.Failed: if self.output_sensor.get_state() != 'off': @@ -697,6 +772,8 @@ class VirtualSensors(): self.virtual_sensors[f"sensor.{splitted_sensor[1]}"] = Averager(self.ad_api,self.logger_interface,splitted_sensor[1],super_entity_id = super_entity_id,yaml_block = yaml_block['sensors'][sensor],constants = constants,templates_library = templates_library,app_name = app_name) elif splitted_sensor[0] == 'value_selector': self.virtual_sensors[f"sensor.{splitted_sensor[1]}"] = ValueSelector(self.ad_api,self.logger_interface,splitted_sensor[1],super_entity_id = super_entity_id,yaml_block = yaml_block['sensors'][sensor],constants = constants,templates_library = templates_library,app_name = app_name) + elif splitted_sensor[0] == 'ephemeral_condition': + self.virtual_sensors[f"binary_sensor.{splitted_sensor[1]}"] = EphemeralCondition(self.ad_api,self.logger_interface,splitted_sensor[1],super_entity_id = super_entity_id,yaml_block = yaml_block['sensors'][sensor],constants = constants,templates_library = templates_library,app_name = app_name) elif splitted_sensor[0] == 'retain_condition': self.virtual_sensors[f"binary_sensor.{splitted_sensor[1]}"] = RetainCondition(self.ad_api,self.logger_interface,splitted_sensor[1],super_entity_id = super_entity_id,yaml_block = yaml_block['sensors'][sensor],constants = constants,templates_library = templates_library,app_name = app_name) elif splitted_sensor[0] == 'event_sensor':