A new epheleral_condition sensor type + fix issue with dangling callback

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