115 lines
5.7 KiB
Python
115 lines
5.7 KiB
Python
from copy import deepcopy
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
import tempfile
|
|
import unittest
|
|
from unittest.mock import patch
|
|
from urllib.parse import urlsplit
|
|
|
|
from solar_relay.cli import _run
|
|
from solar_relay.config import CollectorConfig, Config, HttpConfig, InverterConfig, UploadConfig, utc_text
|
|
from solar_relay.fronius import FroniusClient
|
|
from solar_relay.http import TransportError
|
|
from solar_relay.storage import Store
|
|
from tests.test_relay import RESPONSES, server
|
|
from tests.test_schedule import Clock
|
|
|
|
POWER_PATH = "/solar_api/v1/GetPowerFlowRealtimeData.fcgi"
|
|
METER_PATH = "/solar_api/v1/GetMeterRealtimeData.cgi"
|
|
BOILER_PATH = "/api/status/emrs"
|
|
|
|
|
|
class ScriptedHttp:
|
|
def __init__(self, clock):
|
|
self.clock = clock
|
|
self.requests = []
|
|
self.failures = {POWER_PATH: [503, 503, 503]}
|
|
|
|
def request(self, url):
|
|
path = urlsplit(url).path
|
|
self.requests.append((self.clock.now, path))
|
|
failures = self.failures.get(path, [])
|
|
if failures:
|
|
raise TransportError(f"HTTP {failures.pop(0)}")
|
|
response = deepcopy(RESPONSES[path])
|
|
if "Head" in response:
|
|
response["Head"]["Timestamp"] = utc_text(datetime.fromtimestamp(self.clock.now, timezone.utc))
|
|
return response
|
|
|
|
|
|
class RetryTests(unittest.TestCase):
|
|
def test_http_errors_retry_only_the_failed_endpoint_and_stop_on_success(self):
|
|
for retries in (1, 3):
|
|
with self.subTest(retries=retries), server() as (httpd, base):
|
|
httpd.get_failures[POWER_PATH] = [503] * retries
|
|
client = FroniusClient(InverterConfig(base_url=base, min_request_interval_seconds=0))
|
|
with self.assertLogs("solar_relay.fronius", level="WARNING") as messages:
|
|
sample = client.collect("test-house")
|
|
self.assertEqual(sample["errors"], {})
|
|
self.assertEqual(sample["readings"]["production_ac_w"], 2000)
|
|
self.assertEqual(httpd.get_counts, {POWER_PATH: retries + 1, METER_PATH: 1, BOILER_PATH: 1})
|
|
self.assertEqual(len(messages.output), retries)
|
|
|
|
def test_giving_up_after_three_retries_preserves_other_sources(self):
|
|
for failures in ([500, 502, 503, 504, 503], [404] * 5):
|
|
with self.subTest(statuses=failures), server() as (httpd, base):
|
|
httpd.get_failures[BOILER_PATH] = list(failures)
|
|
client = FroniusClient(InverterConfig(base_url=base, min_request_interval_seconds=0))
|
|
with self.assertLogs("solar_relay.fronius", level="WARNING"):
|
|
sample = client.collect("test-house")
|
|
self.assertEqual(httpd.get_counts, {POWER_PATH: 1, METER_PATH: 1, BOILER_PATH: 4})
|
|
self.assertEqual(len(httpd.get_failures[BOILER_PATH]), 1)
|
|
self.assertEqual(sample["errors"], {"boiler": f"HTTP {failures[3]}"})
|
|
self.assertIsNone(sample["readings"]["boiler_enabled"])
|
|
self.assertEqual(sample["readings"]["production_ac_w"], 2000)
|
|
self.assertEqual(sample["readings"]["grid_import_total_wh"], 50000)
|
|
|
|
def test_retry_attempts_keep_the_four_second_request_spacing(self):
|
|
clock = Clock(0)
|
|
http = ScriptedHttp(clock)
|
|
client = FroniusClient(InverterConfig(), http=http)
|
|
with patch("solar_relay.fronius.time.monotonic", side_effect=lambda: clock.now), \
|
|
patch("solar_relay.fronius.time.sleep", side_effect=clock.wait), \
|
|
self.assertLogs("solar_relay.fronius", level="WARNING"):
|
|
result = client.collect("test-house")
|
|
self.assertEqual(result["errors"], {})
|
|
self.assertEqual([stamp for stamp, _ in http.requests], [0, 4, 8, 12, 16, 20])
|
|
|
|
def test_retries_preserve_clock_boundaries_and_one_saved_observation_per_poll(self):
|
|
origin = datetime(2026, 11, 1, 12, tzinfo=timezone.utc).timestamp()
|
|
clock = Clock(origin + 180) # Start at 12:03; first observation must be at 12:05.
|
|
http = ScriptedHttp(clock)
|
|
client = FroniusClient(InverterConfig(), http=http)
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
store = Store(Path(directory) / "archive")
|
|
config = Config(CollectorConfig("test-house", store.path), InverterConfig(), UploadConfig(), HttpConfig(enabled=False))
|
|
added = []
|
|
original_add = store.add
|
|
|
|
def add(sample):
|
|
original_add(sample)
|
|
added.append(sample)
|
|
if len(added) == 3:
|
|
clock.set()
|
|
|
|
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.fronius.time.sleep", side_effect=clock.wait), \
|
|
patch("solar_relay.fronius.datetime", wraps=datetime) as dates, \
|
|
patch("solar_relay.cli.signal.signal"), \
|
|
patch("solar_relay.cli.FroniusClient", return_value=client), \
|
|
patch.object(store, "add", side_effect=add), \
|
|
self.assertLogs("solar_relay.fronius", level="WARNING"):
|
|
dates.now.side_effect = lambda zone: datetime.fromtimestamp(clock.now, zone)
|
|
self.assertEqual(_run(config, store), 0)
|
|
records = list(store.records())
|
|
self.assertEqual(len(records), 3)
|
|
self.assertEqual([datetime.fromisoformat(s["observed_at"]).timestamp() for s in records],
|
|
[origin + 300, origin + 600, origin + 900])
|
|
self.assertTrue(all(not record["errors"] for record in records))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|