123 lines
5.8 KiB
Python
123 lines
5.8 KiB
Python
from copy import deepcopy
|
|
from datetime import datetime, timedelta, timezone
|
|
from pathlib import Path
|
|
import tempfile
|
|
import unittest
|
|
import pyarrow.parquet as pq
|
|
from urllib.parse import parse_qs, urlsplit
|
|
|
|
from solar_relay.config import HistoryConfig, utc_text
|
|
from solar_relay.history import CHANNELS, HistoryRecovery, history_sample
|
|
from solar_relay.http import TransportError
|
|
from solar_relay.storage import Store
|
|
from tests.test_status import observed
|
|
|
|
SYSTEM = "73733052-6f10-47a5-9746-7f7e76ebcb8c"
|
|
|
|
|
|
def entry(when):
|
|
values = {"production": 100, "consumption": 120, "self_consumption": 80, "bought": 40, "sold": 20}
|
|
return {"logDateTime": utc_text(when), "logDuration": 300,
|
|
"channels": [{"channelName": channel, "channelType": "Energy", "unit": "Wh", "value": values[field]}
|
|
for field, channel in CHANNELS.items()]}
|
|
|
|
|
|
class FakeHistoryHttp:
|
|
def __init__(self, entries):
|
|
self.entries, self.requests = entries, []
|
|
|
|
def request(self, url, **kwargs):
|
|
self.requests.append((url, kwargs))
|
|
return {"pvSystemId": SYSTEM, "data": deepcopy(self.entries)}
|
|
|
|
|
|
class HistoryTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.directory = tempfile.TemporaryDirectory()
|
|
self.addCleanup(self.directory.cleanup)
|
|
self.store = Store(Path(self.directory.name) / "archive")
|
|
self.when = datetime(2026, 10, 30, 10, tzinfo=timezone.utc)
|
|
self.source = "test-house"
|
|
self.http = FakeHistoryHttp([entry(self.when + timedelta(minutes=offset)) for offset in [0, 5, 10]])
|
|
self.config = HistoryConfig(enabled=True, pv_system_id=SYSTEM)
|
|
self.recovery = HistoryRecovery(self.config, self.source, self.store, http=self.http, credentials=("id-test", "value-test"))
|
|
|
|
def recover(self):
|
|
return self.recovery.recover(utc_text(self.when), utc_text(self.when + timedelta(minutes=15)), now=self.when + timedelta(hours=4))
|
|
|
|
def test_wifi_gap_recovers_only_missing_energy_without_inventing_boiler(self):
|
|
for offset in [0, 10]:
|
|
self.store.add(observed(self.when + timedelta(minutes=offset)))
|
|
self.assertEqual(self.recover(), 1)
|
|
records = list(self.store.records())
|
|
recovered = [s for s in records if "history_interval" in s]
|
|
self.assertEqual(len(recovered), 1)
|
|
self.assertEqual(recovered[0]["history_interval"]["energy_wh"]["production"], 100)
|
|
self.assertIsNone(recovered[0]["readings"]["boiler_enabled"])
|
|
self.assertNotIn("boiler", recovered[0]["source_timestamps"])
|
|
self.assertIn("boiler", recovered[0]["errors"])
|
|
url, kwargs = self.http.requests[0]
|
|
self.assertEqual(urlsplit(url).hostname, "swqapi.solarweb.com")
|
|
self.assertEqual(kwargs["extra_headers"], {"AccessKeyId": "id-test", "AccessKeyValue": "value-test"})
|
|
self.assertNotIn("value-test", url)
|
|
self.assertEqual(self.recover(), 0)
|
|
self.assertEqual(len(self.http.requests), 1)
|
|
|
|
def test_cloud_energy_does_not_mark_boiler_or_meter_history_complete(self):
|
|
self.assertEqual(self.recover(), 3)
|
|
rows = self.store.coverage(self.source, "UTC", 1, self.when + timedelta(days=1))
|
|
day = rows[1]
|
|
self.assertEqual(day["energy_complete"], 3)
|
|
self.assertEqual(day["boiler_complete"], 0)
|
|
self.assertEqual(day["meter_complete"], 0)
|
|
self.assertEqual(day["complete"], 0)
|
|
row = pq.ParquetFile(next(self.store.path.glob("date=*/samples.parquet"))).read().to_pylist()[0]
|
|
self.assertEqual(row["record_kind"], "history")
|
|
self.assertEqual(row["interval_energy_wh"]["production"], 100)
|
|
|
|
def test_partial_cloud_records_remain_missing_for_retry(self):
|
|
self.http.entries[1]["channels"][0]["value"] = None
|
|
self.assertEqual(self.recover(), 2)
|
|
self.assertEqual(self.recover(), 0)
|
|
self.http.entries[1] = entry(self.when + timedelta(minutes=5))
|
|
self.assertEqual(self.recover(), 1)
|
|
|
|
def test_recovered_identifiers_are_stable_for_duplicate_delivery(self):
|
|
first = history_sample(entry(self.when), self.source, SYSTEM)
|
|
second = history_sample(entry(self.when), self.source, SYSTEM)
|
|
self.assertEqual(first, second)
|
|
|
|
def test_non_five_minute_and_bad_units_are_rejected(self):
|
|
for change in [{"logDuration": 599}, {"logDateTime": utc_text(self.when + timedelta(seconds=1))}]:
|
|
value = entry(self.when)
|
|
value.update(change)
|
|
with self.assertRaises(ValueError):
|
|
history_sample(value, self.source, SYSTEM)
|
|
value = entry(self.when)
|
|
value["channels"][0]["unit"] = "W"
|
|
with self.assertRaises(ValueError):
|
|
history_sample(value, self.source, SYSTEM)
|
|
|
|
def test_missing_private_credentials_fail_before_requests(self):
|
|
with self.assertRaisesRegex(ValueError, "SOLARWEB_ACCESS_KEY"):
|
|
HistoryRecovery(self.config, self.source, self.store, credentials=("", ""))
|
|
|
|
def test_requests_are_split_at_utc_midnight_even_across_dst(self):
|
|
self.http.entries = []
|
|
self.recovery.recover("2026-10-24T22:00:00Z", "2026-10-26T00:00:00Z", now=datetime(2026, 10, 27, tzinfo=timezone.utc))
|
|
self.assertEqual(len(self.http.requests), 2)
|
|
for url, _ in self.http.requests:
|
|
query = parse_qs(urlsplit(url).query)
|
|
start, end = [datetime.fromisoformat(query[name][0]) for name in ("from", "to")]
|
|
self.assertLessEqual((end - start).total_seconds(), 86400)
|
|
|
|
def test_recent_intervals_wait_for_the_hourly_cloud_upload(self):
|
|
self.http.entries = []
|
|
self.recovery.recover(utc_text(self.when), utc_text(self.when + timedelta(hours=4)), now=self.when + timedelta(hours=4))
|
|
query = parse_qs(urlsplit(self.http.requests[0][0]).query)
|
|
self.assertEqual(query["to"], [utc_text(self.when + timedelta(hours=2))])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|