"""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._timer_generation = 0 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 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') 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, 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', silent=True) 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.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 self.sensor.on_condition_succeed() self.sensor.on_condition_fail() self.assertEqual(self.state, 'on') self.sensor.condition.evaluate.return_value = Result.Succeeded self.queued_expiry()() self.sensor.on_condition_fail() self.assertEqual(self.state, 'off') if __name__ == '__main__': unittest.main()