387 lines
17 KiB
Python
387 lines
17 KiB
Python
from contextlib import contextmanager
|
|
from copy import deepcopy
|
|
from datetime import datetime
|
|
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import tempfile
|
|
import threading
|
|
import unittest
|
|
import subprocess
|
|
import sys
|
|
from unittest.mock import patch
|
|
from urllib.parse import urlsplit
|
|
|
|
from solar_relay.config import InverterConfig, UploadConfig, date_bound, load_config
|
|
from solar_relay.fronius import FroniusClient, powerflow_readings
|
|
from solar_relay.http import JsonHttp, TransportError
|
|
from solar_relay.storage import Store
|
|
from solar_relay.upload import Uploader, UploadError
|
|
|
|
|
|
def api(data):
|
|
return {
|
|
"Body": {"Data": data},
|
|
"Head": {"Status": {"Code": 0}, "Timestamp": "2026-11-01T12:00:00+00:00"},
|
|
}
|
|
|
|
|
|
POWER = api({
|
|
"Site": {"P_PV": 2100, "P_Load": -2500, "P_Grid": 500, "E_Total": 100000},
|
|
"Inverters": {"1": {"P": 2000, "E_Total": 100000}},
|
|
})
|
|
METER = api({"0": {
|
|
"Enable": 1, "Meter_Location_Current": 0,
|
|
"EnergyReal_WAC_Sum_Consumed": 50000, "EnergyReal_WAC_Sum_Produced": 80000,
|
|
}})
|
|
BOILER = {"pin1": {"state": True, "reason": "PSTR_EMRS_REASON_3"}}
|
|
RESPONSES = {
|
|
"/solar_api/v1/GetPowerFlowRealtimeData.fcgi": POWER,
|
|
"/solar_api/v1/GetMeterRealtimeData.cgi": METER,
|
|
"/api/status/emrs": BOILER,
|
|
}
|
|
|
|
|
|
class FakeHttp:
|
|
def __init__(self, responses=None):
|
|
self.responses = deepcopy(RESPONSES if responses is None else responses)
|
|
|
|
def request(self, url):
|
|
result = self.responses[urlsplit(url).path]
|
|
if isinstance(result, Exception):
|
|
raise result
|
|
return deepcopy(result)
|
|
|
|
|
|
def sample():
|
|
return FroniusClient(InverterConfig(min_request_interval_seconds=0), http=FakeHttp()).collect("test-house")
|
|
|
|
|
|
@contextmanager
|
|
def server():
|
|
class Handler(BaseHTTPRequestHandler):
|
|
def log_message(self, *args):
|
|
pass
|
|
|
|
def respond(self, status, data):
|
|
body = json.dumps(data).encode()
|
|
self.send_response(status)
|
|
self.send_header("Content-Type", "application/json")
|
|
self.send_header("Content-Length", str(len(body)))
|
|
self.end_headers()
|
|
self.wfile.write(body)
|
|
|
|
def do_GET(self):
|
|
path = urlsplit(self.path).path
|
|
self.server.get_counts[path] = self.server.get_counts.get(path, 0) + 1
|
|
failures = self.server.get_failures.get(path, [])
|
|
if failures:
|
|
self.respond(failures.pop(0), {"error": "temporary"})
|
|
return
|
|
if self.path == "/redirect":
|
|
self.send_response(302)
|
|
self.send_header("Location", "/unexpected")
|
|
self.send_header("Content-Length", "0")
|
|
self.end_headers()
|
|
elif self.path == "/unexpected":
|
|
self.server.redirect_followed = True
|
|
self.respond(200, {})
|
|
elif self.path == "/nan":
|
|
self.respond(200, {"reading": float("nan")})
|
|
else:
|
|
path = urlsplit(self.path).path
|
|
self.respond(200, RESPONSES[path])
|
|
|
|
def do_POST(self):
|
|
self.server.authorizations.append(self.headers.get("Authorization"))
|
|
data = json.loads(self.rfile.read(int(self.headers["Content-Length"])))
|
|
self.server.requests.append(data)
|
|
ids = [s["id"] for s in data["samples"]]
|
|
# Emulate an idempotent receiver that committed before its response failed.
|
|
self.server.received_ids.update(ids)
|
|
if self.server.fail_once:
|
|
self.server.fail_once = False
|
|
self.respond(503, {"error": "temporary"})
|
|
elif self.server.ack_override is not None:
|
|
self.respond(200, self.server.ack_override)
|
|
else:
|
|
self.respond(200, {"accepted_ids": ids})
|
|
|
|
httpd = ThreadingHTTPServer(("127.0.0.1", 0), Handler)
|
|
httpd.fail_once = False
|
|
httpd.ack_override = None
|
|
httpd.requests = []
|
|
httpd.get_counts = {}
|
|
httpd.get_failures = {}
|
|
httpd.authorizations = []
|
|
httpd.received_ids = set()
|
|
httpd.redirect_followed = False
|
|
thread = threading.Thread(target=httpd.serve_forever, daemon=True)
|
|
thread.start()
|
|
try:
|
|
yield httpd, f"http://127.0.0.1:{httpd.server_port}"
|
|
finally:
|
|
httpd.shutdown()
|
|
httpd.server_close()
|
|
thread.join(timeout=2)
|
|
|
|
|
|
class NormalizationTests(unittest.TestCase):
|
|
def test_dc_ac_and_meter_directions_stay_distinct(self):
|
|
result = sample()
|
|
readings = result["readings"]
|
|
self.assertEqual(readings["production_w"], 2100)
|
|
self.assertEqual(readings["production_ac_w"], 2000)
|
|
self.assertEqual(readings["consumption_w"], 2500)
|
|
self.assertEqual(readings["grid_import_w"], 500)
|
|
self.assertEqual(readings["grid_export_w"], 0)
|
|
self.assertEqual(readings["grid_import_total_wh"], 50000)
|
|
self.assertTrue(readings["boiler_enabled"])
|
|
self.assertEqual(result["errors"], {})
|
|
self.assertEqual(result["raw"]["powerflow"], POWER)
|
|
self.assertEqual(result["source_timestamps"]["powerflow"], "2026-11-01T12:00:00.000Z")
|
|
|
|
def test_export_direction_and_nighttime_nulls(self):
|
|
response = deepcopy(POWER)
|
|
response["Body"]["Data"]["Site"].update(P_Grid=-1200, P_PV=None)
|
|
response["Body"]["Data"]["Inverters"]["1"]["P"] = None
|
|
readings, _ = powerflow_readings(response)
|
|
self.assertEqual(readings["grid_import_w"], 0)
|
|
self.assertEqual(readings["grid_export_w"], 1200)
|
|
self.assertIsNone(readings["production_w"])
|
|
self.assertIsNone(readings["production_ac_w"])
|
|
|
|
def test_boiler_failure_is_unknown_and_does_not_lose_energy(self):
|
|
responses = deepcopy(RESPONSES)
|
|
responses["/api/status/emrs"] = TransportError("HTTP 404")
|
|
result = FroniusClient(InverterConfig(min_request_interval_seconds=0), http=FakeHttp(responses)).collect("test")
|
|
self.assertIsNone(result["readings"]["boiler_enabled"])
|
|
self.assertEqual(result["readings"]["production_w"], 2100)
|
|
self.assertEqual(result["errors"], {"boiler": "HTTP 404"})
|
|
|
|
def test_false_is_a_valid_boiler_state(self):
|
|
responses = deepcopy(RESPONSES)
|
|
responses["/api/status/emrs"]["pin1"]["state"] = False
|
|
result = FroniusClient(InverterConfig(min_request_interval_seconds=0), http=FakeHttp(responses)).collect("test")
|
|
self.assertIs(result["readings"]["boiler_enabled"], False)
|
|
self.assertNotIn("boiler", result["errors"])
|
|
|
|
def test_solar_api_application_error_does_not_accept_old_values(self):
|
|
responses = deepcopy(RESPONSES)
|
|
responses["/solar_api/v1/GetPowerFlowRealtimeData.fcgi"]["Head"]["Status"]["Code"] = 12
|
|
result = FroniusClient(InverterConfig(min_request_interval_seconds=0), http=FakeHttp(responses)).collect("test")
|
|
self.assertIsNone(result["readings"]["production_w"])
|
|
self.assertIn("powerflow", result["errors"])
|
|
self.assertTrue(result["readings"]["boiler_enabled"])
|
|
|
|
def test_load_branch_meter_cannot_be_mislabeled_as_grid_energy(self):
|
|
responses = deepcopy(RESPONSES)
|
|
responses["/solar_api/v1/GetMeterRealtimeData.cgi"]["Body"]["Data"]["0"]["Meter_Location_Current"] = 1
|
|
result = FroniusClient(InverterConfig(min_request_interval_seconds=0), http=FakeHttp(responses)).collect("test")
|
|
self.assertIsNone(result["readings"]["grid_import_total_wh"])
|
|
self.assertIn("meter", result["errors"])
|
|
|
|
def test_meter_measurement_timestamp_is_preserved_instead_of_request_time(self):
|
|
responses = deepcopy(RESPONSES)
|
|
responses["/solar_api/v1/GetMeterRealtimeData.cgi"]["Body"]["Data"]["0"]["TimeStamp"] = 1793534100
|
|
result = FroniusClient(InverterConfig(min_request_interval_seconds=0), http=FakeHttp(responses)).collect("test")
|
|
self.assertEqual(result["source_timestamps"]["meter"], "2026-11-01T11:55:00.000Z")
|
|
|
|
def test_missing_power_fields_are_a_source_error(self):
|
|
responses = deepcopy(RESPONSES)
|
|
responses["/solar_api/v1/GetPowerFlowRealtimeData.fcgi"]["Body"]["Data"]["Site"] = {}
|
|
result = FroniusClient(InverterConfig(min_request_interval_seconds=0), http=FakeHttp(responses)).collect("test")
|
|
self.assertIn("powerflow", result["errors"])
|
|
self.assertIsNone(result["readings"]["production_w"])
|
|
|
|
|
|
class ArchiveTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.directory = tempfile.TemporaryDirectory()
|
|
self.addCleanup(self.directory.cleanup)
|
|
self.path = Path(self.directory.name) / "archive"
|
|
self.store = Store(self.path)
|
|
|
|
def test_restart_and_acknowledgement_preserve_archive(self):
|
|
first = sample()
|
|
self.store.add(first)
|
|
Store(self.path).acknowledge([first["id"]])
|
|
reopened = Store(self.path)
|
|
self.assertEqual(reopened.pending(100), [])
|
|
self.assertEqual(list(reopened.records()), [first])
|
|
self.assertEqual(reopened.status()["samples"], 1)
|
|
if os.name == "posix":
|
|
self.assertEqual(self.path.stat().st_mode & 0o777, 0o700)
|
|
self.assertEqual(next(self.path.glob("date=*/samples.parquet")).stat().st_mode & 0o777, 0o600)
|
|
|
|
def test_local_date_export_and_replay_across_dst(self):
|
|
start = date_bound("2026-10-25", "Europe/Zurich")
|
|
until = date_bound("2026-10-26", "Europe/Zurich")
|
|
self.assertEqual((datetime.fromisoformat(until) - datetime.fromisoformat(start)).total_seconds(), 25 * 3600)
|
|
items = []
|
|
for timestamp in ["2026-10-24T21:59:59.000Z", start, "2026-10-25T12:00:00.000Z", until]:
|
|
item = sample()
|
|
item["observed_at"] = timestamp
|
|
items.append(item)
|
|
self.store.add(item)
|
|
self.store.acknowledge([item["id"] for item in items])
|
|
self.assertEqual(list(self.store.records(start, until)), items[1:3])
|
|
self.assertEqual(self.store.replay(start, until), 2)
|
|
self.assertEqual(self.store.pending(100), items[1:3])
|
|
self.assertEqual(self.store.status()["samples"], 4)
|
|
|
|
|
|
class DeliveryTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.directory = tempfile.TemporaryDirectory()
|
|
self.addCleanup(self.directory.cleanup)
|
|
self.store = Store(Path(self.directory.name) / "archive")
|
|
|
|
def test_real_http_collection_delivery_and_lost_ack_retry(self):
|
|
with server() as (httpd, base):
|
|
client = FroniusClient(InverterConfig(base_url=base, min_request_interval_seconds=0))
|
|
reading = client.collect("test")
|
|
self.assertEqual(reading["errors"], {})
|
|
self.store.add(reading)
|
|
uploader = Uploader(UploadConfig(url=base + "/ingest"), self.store, token="test-token")
|
|
httpd.fail_once = True
|
|
with self.assertRaises(UploadError):
|
|
uploader.upload_once()
|
|
self.assertEqual(self.store.status()["pending"], 1)
|
|
self.assertEqual(uploader.upload_once(), 1)
|
|
self.assertEqual(self.store.status()["pending"], 0)
|
|
self.assertEqual(self.store.status()["samples"], 1)
|
|
self.assertEqual(httpd.requests[0], httpd.requests[1])
|
|
self.assertEqual(httpd.received_ids, {reading["id"]})
|
|
self.assertEqual(httpd.authorizations, ["Bearer test-token", "Bearer test-token"])
|
|
|
|
def test_plain_http_success_is_not_an_acknowledgement(self):
|
|
with server() as (httpd, base):
|
|
self.store.add(sample())
|
|
uploader = Uploader(UploadConfig(url=base), self.store, token="test-token")
|
|
httpd.ack_override = {"success": True}
|
|
with self.assertRaises(UploadError):
|
|
uploader.upload_once()
|
|
self.assertEqual(self.store.status()["pending"], 1)
|
|
|
|
def test_partial_ack_only_marks_the_accepted_sample(self):
|
|
with server() as (httpd, base):
|
|
first, second = sample(), sample()
|
|
self.store.add(first)
|
|
self.store.add(second)
|
|
httpd.ack_override = {"accepted_ids": [first["id"]]}
|
|
with self.assertRaises(UploadError):
|
|
Uploader(UploadConfig(url=base), self.store, token="test-token").upload_once()
|
|
self.assertEqual(self.store.pending(100), [second])
|
|
self.assertEqual(self.store.status()["samples"], 2)
|
|
|
|
def test_unknown_ack_cannot_clear_the_queue(self):
|
|
with server() as (httpd, base):
|
|
self.store.add(sample())
|
|
httpd.ack_override = {"accepted_ids": ["unknown-id"]}
|
|
with self.assertRaises(UploadError):
|
|
Uploader(UploadConfig(url=base), self.store, token="test-token").upload_once()
|
|
self.assertEqual(self.store.status()["pending"], 1)
|
|
|
|
def test_slow_upload_does_not_lock_recording(self):
|
|
entered, release = threading.Event(), threading.Event()
|
|
|
|
class BlockingHttp:
|
|
def request(self, url, *, payload, token):
|
|
entered.set()
|
|
release.wait(timeout=3)
|
|
return {"accepted_ids": [s["id"] for s in payload["samples"]]}
|
|
|
|
self.store.add(sample())
|
|
uploader = Uploader(UploadConfig(url="https://example.test/ingest"), self.store, token="test-token", http=BlockingHttp())
|
|
result = []
|
|
worker = threading.Thread(target=lambda: result.append(uploader.upload_once()))
|
|
worker.start()
|
|
try:
|
|
self.assertTrue(entered.wait(timeout=2))
|
|
second = sample()
|
|
self.store.add(second)
|
|
self.assertEqual(self.store.status()["samples"], 2)
|
|
finally:
|
|
release.set()
|
|
worker.join(timeout=4)
|
|
self.assertEqual(result, [1])
|
|
self.assertEqual(self.store.pending(100), [second])
|
|
|
|
def test_redirects_and_invalid_json_numbers_are_rejected(self):
|
|
with server() as (httpd, base):
|
|
http = JsonHttp(timeout=2)
|
|
with self.assertRaises(TransportError):
|
|
http.request(base + "/redirect", token="test-token")
|
|
self.assertFalse(httpd.redirect_followed)
|
|
with self.assertRaises(TransportError):
|
|
http.request(base + "/nan")
|
|
|
|
def test_missing_upload_token_fails_before_delivery(self):
|
|
with patch.dict(os.environ, {}, clear=True):
|
|
with self.assertRaisesRegex(ValueError, "SOLAR_RELAY_TOKEN"):
|
|
Uploader(UploadConfig(url="https://example.test/ingest"), self.store)
|
|
|
|
def test_cli_sample_and_export_use_the_persistent_archive(self):
|
|
with server() as (_, base):
|
|
config = Path(self.directory.name) / "config.toml"
|
|
config.write_text(
|
|
'[collector]\nsource_id="cli-test"\narchive="cli-archive"\n'
|
|
f'\n[inverter]\nbase_url="{base}"\n'
|
|
)
|
|
command = [sys.executable, "-m", "solar_relay", "--config", str(config)]
|
|
cwd = Path(__file__).resolve().parents[1]
|
|
result = subprocess.run(command + ["sample"], cwd=cwd, capture_output=True, text=True, timeout=20)
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
normalized = json.loads(result.stdout)
|
|
result = subprocess.run(command + ["export"], cwd=cwd, capture_output=True, text=True, timeout=5)
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
archived = json.loads(result.stdout)
|
|
self.assertEqual(normalized["id"], archived["id"])
|
|
self.assertEqual(archived["raw"]["powerflow"], POWER)
|
|
|
|
|
|
class ConfigTests(unittest.TestCase):
|
|
def load(self, extra=""):
|
|
with tempfile.TemporaryDirectory() as name:
|
|
path = Path(name) / "config.toml"
|
|
path.write_text('[collector]\nsource_id="home"\narchive="state/archive"\n' + extra)
|
|
return load_config(path), path.parent
|
|
|
|
def test_archive_is_relative_to_config(self):
|
|
config, directory = self.load()
|
|
self.assertEqual(config.collector.archive, directory / "state/archive")
|
|
self.assertEqual(config.collector.poll_interval_seconds, 300)
|
|
self.assertEqual(config.collector.retention_days, 30)
|
|
self.assertEqual(config.upload.url, "")
|
|
|
|
def test_remote_upload_requires_https(self):
|
|
with self.assertRaisesRegex(ValueError, "HTTPS"):
|
|
self.load('\n[upload]\nurl="http://example.test/ingest"\n')
|
|
config, _ = self.load('\n[upload]\nurl="http://127.0.0.1:8000/ingest"\n')
|
|
self.assertTrue(config.upload.url)
|
|
|
|
def test_mistyped_configuration_is_rejected(self):
|
|
with self.assertRaisesRegex(ValueError, "Unknown inverter"):
|
|
self.load('\n[inverter]\nboiler_pni="pin2"\n')
|
|
with self.assertRaises(ValueError):
|
|
self.load('\n[inverter]\nboiler_pin="pin9"\n')
|
|
|
|
def test_status_listener_defaults_and_loopback_binding(self):
|
|
config, _ = self.load()
|
|
self.assertEqual(config.http.port, 8765)
|
|
self.assertEqual(config.http.bind, "127.0.0.1")
|
|
with self.assertRaisesRegex(ValueError, "127.0.0.1"):
|
|
self.load('\n[http]\nbind="0.0.0.0"\n')
|
|
|
|
def test_status_settings_can_disable_http_and_reject_invalid_paths(self):
|
|
config, _ = self.load('\n[http]\nenabled=false\n')
|
|
self.assertFalse(config.http.enabled)
|
|
with self.assertRaises(ValueError):
|
|
self.load('\n[http]\nbase_path="/../config"\n')
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|