106 lines
4.4 KiB
Python
106 lines
4.4 KiB
Python
"""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()
|