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

@@ -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()