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()