102 lines
4.8 KiB
Python
102 lines
4.8 KiB
Python
"""Time-period boundaries, source updates and timer lifecycle without AppDaemon."""
|
|
import ast
|
|
import datetime
|
|
import re
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest.mock import Mock
|
|
|
|
source = Path(__file__).resolve().parents[1] / 'virtualsensors.py'
|
|
nodes = [node for node in ast.parse(source.read_text()).body
|
|
if isinstance(node, ast.ClassDef) and node.name in ('VirtualSensorBase', 'TimePeriod', 'VirtualSensors')]
|
|
from kwargsparser import kwargsParser
|
|
namespace = {'datetime': datetime, 're': re, 'kwargsParser': kwargsParser, 'ParsingException': ValueError}
|
|
exec(compile(ast.Module(body=nodes, type_ignores=[]), str(source), 'exec'), namespace)
|
|
TimePeriod = namespace['TimePeriod']
|
|
|
|
|
|
class TimePeriodTests(unittest.TestCase):
|
|
def make_sensor(self, start='08:00:00', end='18:00:00', now='2026-10-25 12:00:00'):
|
|
self.states = {'input_datetime.start': start, 'input_datetime.end': end}
|
|
self.api = Mock()
|
|
self.api.datetime.return_value = datetime.datetime.fromisoformat(now)
|
|
self.api.get_state.side_effect = self.states.get
|
|
self.api.entity_exists.return_value = True
|
|
self.api.get_entity.return_value.get_state.return_value = 'unknown'
|
|
self.sensor = TimePeriod(self.api, Mock(), virtual_sensor_name='period',
|
|
yaml_block={'start': 'input_datetime.start', 'end': 'input_datetime.end'})
|
|
self.sensor._initialize()
|
|
return self.sensor
|
|
|
|
def assert_state(self, expected):
|
|
self.assertEqual(self.sensor.entity.set_state.call_args.kwargs['state'], expected)
|
|
|
|
def test_factory_registration(self):
|
|
self.make_sensor()
|
|
factory = namespace['VirtualSensors'](ad_api=self.api, logger_interface=Mock(),
|
|
yaml_block={'sensors': {'time_period.example': {
|
|
'start': 'input_datetime.start', 'end': 'input_datetime.end'}}})
|
|
self.assertIsInstance(factory.virtual_sensors['binary_sensor.example'], TimePeriod)
|
|
|
|
def test_day_boundaries_and_timer(self):
|
|
self.make_sensor(now='2026-10-25 07:59:50')
|
|
self.assert_state('off')
|
|
self.assertEqual(self.api.run_in.call_args.args[1], 10)
|
|
for hour, expected in [('08:00:00', 'on'), ('17:59:59', 'on'), ('18:00:00', 'off')]:
|
|
self.api.datetime.return_value = datetime.datetime.fromisoformat('2026-10-25 ' + hour)
|
|
self.sensor.on_time_elapsed({'generation': self.sensor._timer_generation})
|
|
self.assert_state(expected)
|
|
|
|
def test_overnight_and_equal_boundaries(self):
|
|
for hour, expected in [('23:00:00', 'on'), ('02:00:00', 'on'), ('06:00:00', 'off'), ('12:00:00', 'off')]:
|
|
self.make_sensor('22:00:00', '06:00:00', '2026-10-25 ' + hour)
|
|
self.assert_state(expected)
|
|
self.make_sensor('08:00:00', '08:00:00')
|
|
self.assert_state('off')
|
|
|
|
def test_source_change_cancels_timer_and_ignores_stale_callback(self):
|
|
self.make_sensor()
|
|
self.assert_state('on')
|
|
previous_generation = self.sensor._timer_generation
|
|
previous_handle = self.sensor.cb_handle
|
|
self.states['input_datetime.end'] = '10:00:00'
|
|
self.sensor.on_input_change('input_datetime.end', 'state', '18:00:00', '10:00:00', {})
|
|
self.assert_state('off')
|
|
self.api.cancel_timer.assert_called_with(previous_handle, silent=True)
|
|
count = self.api.run_in.call_count
|
|
self.sensor.on_time_elapsed({'generation': previous_generation})
|
|
self.assertEqual(self.api.run_in.call_count, count)
|
|
self.assertEqual(self.api.listen_state.call_count, 2)
|
|
|
|
def test_unavailable_and_recovery(self):
|
|
for invalid in [None, 'unknown', 'unavailable', '25:00:00', '2026-10-25']:
|
|
self.make_sensor(start=invalid)
|
|
self.assert_state('unavailable')
|
|
self.states['input_datetime.start'] = '08:00:00'
|
|
self.sensor.update_period()
|
|
self.assert_state('on')
|
|
|
|
def test_dated_periods(self):
|
|
for start, end, expected in [
|
|
('2026-10-24 22:00:00', '2026-10-26 06:00:00', 'on'),
|
|
('2026-10-24 22:00:00', '2026-10-25 12:00:00', 'off'),
|
|
('2026-10-26 22:00:00', '2026-10-24 06:00:00', 'off'),
|
|
('2026-10-24', '2026-10-26', 'on'),
|
|
]:
|
|
self.make_sensor(start, end)
|
|
self.assert_state(expected)
|
|
|
|
def test_dependencies_constants_and_validation(self):
|
|
self.make_sensor()
|
|
self.assertEqual(self.sensor.sensor_name, 'binary_sensor.period')
|
|
self.sensor.constants['begin'] = 'input_datetime.start'
|
|
self.sensor.yaml_block['start'] = 'begin'
|
|
self.assertEqual(self.sensor.generate_dependencies_list(), list(self.states))
|
|
self.sensor.yaml_block['end'] = 'sensor.invalid'
|
|
with self.assertRaises(ValueError):
|
|
self.sensor.generate_dependencies_list()
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|