92 lines
3.3 KiB
Python
92 lines
3.3 KiB
Python
"""Exercise the real recording loop with an advancing clock instead of real waits."""
|
|
|
|
from pathlib import Path
|
|
import tempfile
|
|
import unittest
|
|
from unittest.mock import Mock, patch
|
|
|
|
from solar_relay.cli import _run
|
|
from solar_relay.config import CollectorConfig, Config, HttpConfig, InverterConfig, UploadConfig, load_config
|
|
|
|
|
|
class Clock:
|
|
def __init__(self, now):
|
|
self.now = now
|
|
self.stopped = False
|
|
self.waits = []
|
|
|
|
def is_set(self):
|
|
return self.stopped
|
|
|
|
def set(self):
|
|
self.stopped = True
|
|
|
|
def wait(self, delay):
|
|
self.waits.append((self.now, delay))
|
|
if not self.stopped:
|
|
self.now += delay
|
|
return self.stopped
|
|
|
|
|
|
class ScheduleTests(unittest.TestCase):
|
|
def run_collector(self, start, count=3, duration=8, cancel=False):
|
|
clock = Clock(start)
|
|
recorded = []
|
|
config = Config(CollectorConfig("test-house", Path("unused")), InverterConfig(), UploadConfig(), HttpConfig(enabled=False))
|
|
store = Mock()
|
|
store.prune.return_value = 0
|
|
|
|
def record(*args):
|
|
recorded.append(clock.now)
|
|
clock.now += duration
|
|
if len(recorded) == count:
|
|
clock.set()
|
|
|
|
def cancel_wait(delay):
|
|
clock.set()
|
|
return True
|
|
|
|
if cancel:
|
|
clock.wait = cancel_wait
|
|
with patch("solar_relay.cli.threading.Event", return_value=clock), \
|
|
patch("solar_relay.cli.time.time", side_effect=lambda: clock.now), \
|
|
patch("solar_relay.cli.time.monotonic", side_effect=lambda: clock.now), \
|
|
patch("solar_relay.cli.signal.signal"), \
|
|
patch("solar_relay.cli.FroniusClient"), \
|
|
patch("solar_relay.cli._record", side_effect=record):
|
|
self.assertEqual(_run(config, store), 0)
|
|
return recorded, clock
|
|
|
|
def test_startup_waits_then_records_once_on_each_boundary(self):
|
|
recorded, clock = self.run_collector(301)
|
|
self.assertEqual(recorded, [600, 900, 1200])
|
|
self.assertEqual(clock.waits[0], (301, 299))
|
|
self.assertTrue(all(stamp % 300 == 0 for stamp in recorded))
|
|
|
|
def test_restart_in_the_same_interval_adds_no_extra_observation(self):
|
|
before, _ = self.run_collector(590, count=1)
|
|
after, _ = self.run_collector(615, count=2)
|
|
self.assertEqual(before + after, [600, 900, 1200])
|
|
|
|
def test_slow_collection_skips_missed_boundaries_without_catching_up(self):
|
|
recorded, _ = self.run_collector(301, count=2, duration=370)
|
|
self.assertEqual(recorded, [600, 1200])
|
|
|
|
def test_shutdown_while_waiting_does_not_take_a_final_observation(self):
|
|
recorded, _ = self.run_collector(301, cancel=True)
|
|
self.assertEqual(recorded, [])
|
|
|
|
def test_configuration_requires_exactly_five_minutes(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
path = Path(directory) / "config.toml"
|
|
for interval in (30, 60, 299, 301, 600):
|
|
path.write_text(f"[collector]\npoll_interval_seconds = {interval}\n")
|
|
with self.assertRaisesRegex(ValueError, "must be 300"):
|
|
load_config(path)
|
|
path.write_text("[collector]\npoll_interval_seconds = 300\n")
|
|
self.assertEqual(load_config(path).collector.poll_interval_seconds, 300)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|