Pierre
54c525c843
- Add generation counters to SmartCondition and EventSensor timers. - Preserve pending changes and the retention period when timers are restarted. - Silently cancel timers that have already expired. - Add regression tests for stale callbacks and expiry during cancellation.
80 lines
3.0 KiB
Python
80 lines
3.0 KiB
Python
"""An expired timer queued before reactivation must not end the new window."""
|
|
import ast
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
import unittest
|
|
from unittest.mock import Mock
|
|
|
|
from test_continuous_condition import FakeScheduler
|
|
|
|
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 == 'EventSensor')
|
|
namespace = {'VirtualSensorBase': object, 'EventHandler': Mock()}
|
|
exec(compile(ast.Module(body=[node], type_ignores=[]), str(source), 'exec'), namespace)
|
|
EventSensor = namespace['EventSensor']
|
|
|
|
|
|
class EventSensorTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.scheduler = FakeScheduler()
|
|
self.state = 'off'
|
|
entity = Mock()
|
|
entity.exists.return_value = True
|
|
entity.set_state.side_effect = lambda *, state: setattr(self, 'state', state)
|
|
self.sensor = EventSensor()
|
|
self.sensor.ad_api = SimpleNamespace(
|
|
get_entity=lambda name: entity,
|
|
run_in=self.scheduler.run_in,
|
|
cancel_timer=self.scheduler.cancel_timer,
|
|
)
|
|
self.sensor.sensor_name = 'binary_sensor.red_shelly_home'
|
|
self.sensor.yaml_block = {'retain_time': 120}
|
|
self.sensor.initialize()
|
|
|
|
def activate(self):
|
|
self.sensor.on_activation_event('signal_strength', {}, {})
|
|
return self.sensor.cb_handle
|
|
|
|
def test_current_expiry_turns_sensor_off(self):
|
|
handle = self.activate()
|
|
self.assertEqual(self.state, 'on')
|
|
self.scheduler.expire(handle)()
|
|
self.assertEqual(self.state, 'off')
|
|
self.assertIsNone(self.sensor.cb_handle)
|
|
|
|
def test_reactivation_cancels_running_timer(self):
|
|
old = self.activate()
|
|
current = self.activate()
|
|
self.assertNotIn(old, self.scheduler.timers)
|
|
self.assertIn(current, self.scheduler.timers)
|
|
self.assertEqual(self.state, 'on')
|
|
self.assertEqual(self.scheduler.warnings, [])
|
|
|
|
def test_queued_expiry_preserves_reactivated_sensor_and_timer(self):
|
|
queued = self.scheduler.expire(self.activate())
|
|
current = self.activate()
|
|
queued()
|
|
self.assertEqual(self.state, 'on')
|
|
self.assertEqual(self.sensor.cb_handle, current)
|
|
self.assertEqual(self.scheduler.warnings, [])
|
|
self.scheduler.expire(current)()
|
|
self.assertEqual(self.state, 'off')
|
|
self.assertIsNone(self.sensor.cb_handle)
|
|
|
|
def test_multiple_queued_expiries_cannot_end_latest_window(self):
|
|
first = self.scheduler.expire(self.activate())
|
|
second = self.scheduler.expire(self.activate())
|
|
current = self.activate()
|
|
second()
|
|
first()
|
|
self.assertEqual(self.state, 'on')
|
|
self.assertEqual(self.sensor.cb_handle, current)
|
|
self.assertEqual(self.scheduler.warnings, [])
|
|
self.scheduler.expire(current)()
|
|
self.assertEqual(self.state, 'off')
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|