from dataclasses import replace from datetime import datetime, timedelta, timezone from pathlib import Path import json import socket import subprocess import sys import tempfile import time import unittest from urllib.error import HTTPError, URLError from urllib.request import ProxyHandler, Request, build_opener from zoneinfo import ZoneInfo from solar_relay.config import Config, CollectorConfig, HttpConfig, InverterConfig, UploadConfig, utc_text from solar_relay.coverage import daily_coverage, sample_coverage from solar_relay.http import NoRedirects from solar_relay.status import StatusServer from solar_relay.storage import Store from tests.test_relay import sample, server def observed(when): value = sample() value["observed_at"] = utc_text(when) value["source_timestamps"] = {source: value["observed_at"] for source in ("powerflow", "meter", "boiler")} return value def rows_for_day(day): start = datetime.fromisoformat(day).replace(tzinfo=ZoneInfo("Europe/Zurich")) end = start + timedelta(days=1) first, last = round(start.timestamp() * 1000), round(end.timestamp() * 1000) return [(stamp, stamp, 1, stamp, 1, stamp, 1) for stamp in range(first, last + 1, 300_000)], end class CoverageTests(unittest.TestCase): def test_full_day_includes_all_288_intervals(self): rows, end = rows_for_day("2026-10-30") result = daily_coverage(rows, "Europe/Zurich", 1, end + timedelta(minutes=3)) yesterday = result[1] self.assertEqual(yesterday["expected"], 288) self.assertEqual(yesterday["complete"], 288) self.assertEqual(yesterday["status"], "complete") self.assertEqual(result[0]["elapsed"], 0) self.assertTrue(all(mask == -1 for mask in result[0]["slots"])) def test_dst_days_expect_276_and_300_intervals(self): for date, expected in [("2026-03-29", 276), ("2026-10-25", 300)]: with self.subTest(date=date): rows, end = rows_for_day(date) yesterday = daily_coverage(rows, "Europe/Zurich", 1, end + timedelta(minutes=3))[1] self.assertEqual(yesterday["expected"], expected) self.assertEqual(yesterday["complete"], expected) def test_missing_five_minute_extractions_remain_missing(self): rows, end = rows_for_day("2026-10-30") del rows[10:13] yesterday = daily_coverage(rows, "Europe/Zurich", 1, end + timedelta(minutes=3))[1] self.assertEqual(yesterday["complete"], 285) self.assertEqual(yesterday["missing"], 3) self.assertEqual(yesterday["status"], "partial") def test_energy_failure_does_not_erase_meter_or_boiler_coverage(self): rows, end = rows_for_day("2026-10-30") stamp = rows[1][0] rows[1] = (stamp, stamp, 0, stamp, 1, stamp, 1) yesterday = daily_coverage(rows, "Europe/Zurich", 1, end + timedelta(minutes=3))[1] self.assertEqual(yesterday["slots"][1], 6) self.assertEqual(yesterday["energy_complete"], 287) self.assertEqual(yesterday["meter_complete"], 288) self.assertEqual(yesterday["boiler_complete"], 288) self.assertEqual(yesterday["missing"], 1) def test_source_errors_stale_readings_and_false_boiler_state(self): when = datetime(2026, 10, 30, tzinfo=timezone.utc) value = observed(when) value["readings"]["boiler_enabled"] = False self.assertEqual(sample_coverage(value)[-1], 1) value["errors"] = {"boiler": "unavailable"} self.assertEqual(sample_coverage(value)[-1], 0) value["source_timestamps"]["powerflow"] = utc_text(when - timedelta(seconds=120)) self.assertEqual(sample_coverage(value)[4], 0) def test_parquet_coverage_uses_the_selected_source_and_stored_extractions(self): with tempfile.TemporaryDirectory() as directory: archive = Store(Path(directory) / "archive") start = datetime(2026, 10, 30, tzinfo=ZoneInfo("Europe/Zurich")) for seconds in range(0, 301, 30): value = observed(start + timedelta(seconds=seconds)) archive.add(value) result = archive.coverage(value["source_id"], "Europe/Zurich", 1, start + timedelta(days=1, minutes=3)) self.assertEqual(result[1]["complete"], 2) other = archive.coverage("another-source", "Europe/Zurich", 1, start + timedelta(days=1, minutes=3)) self.assertEqual(other[1]["complete"], 0) class StatusHttpTests(unittest.TestCase): def setUp(self): self.directory = tempfile.TemporaryDirectory() self.addCleanup(self.directory.cleanup) self.store = Store(Path(self.directory.name) / "archive") self.config = Config(CollectorConfig("test-house", self.store.path), InverterConfig(), UploadConfig(), HttpConfig(port=0)) self.status = StatusServer(self.config, self.store) self.status.start() self.addCleanup(self.status.close) self.base = "http://127.0.0.1:" + str(self.status.address[1]) self.http = build_opener(ProxyHandler({}), NoRedirects()) def get(self, path, *, method="GET"): try: result = self.http.open(Request(self.base + path, method=method), timeout=3) except HTTPError as error: result = error with result: return result.code, result.headers, result.read() def test_page_assets_prefix_and_health_routes(self): for path in ["/", "/solar-relay/", "/assets/status.css", "/solar-relay/assets/status.js"]: status, _, body = self.get(path) self.assertEqual(status, 200, path) self.assertTrue(body) status, headers, body = self.get("/solar-relay") self.assertEqual(status, 308) self.assertEqual(headers["Location"], "/solar-relay/") self.assertEqual(json.loads(self.get("/solar-relay/healthz")[2])["status"], "ok") def test_status_reports_extraction_and_backlog_without_raw_data_or_secrets(self): value = observed(datetime.now(timezone.utc)) value["readings"]["boiler_enabled"] = False self.store.add(value) status, headers, body = self.get("/solar-relay/api/status?days=7") result = json.loads(body) self.assertEqual(status, 200) self.assertEqual(headers["Cache-Control"], "no-store") self.assertEqual(result["status"], "ok") self.assertEqual(result["collector"]["last_attempt"], value["observed_at"]) self.assertEqual(result["collector"]["last_success"], value["observed_at"]) self.assertEqual(result["archive"]["pending"], 1) self.assertEqual(len(result["days"]), 8) for forbidden in ["production_w", "raw", "database", self.directory.name]: self.assertNotIn(forbidden, body.decode()) def test_source_errors_do_not_expose_exception_messages(self): value = observed(datetime.now(timezone.utc)) value["errors"] = {"boiler": "secret canary must not appear in HTTP"} self.store.add(value) result = json.loads(self.get("/api/status")[2]) self.assertEqual(result["status"], "partial") self.assertEqual(result["collector"]["source_errors"], ["boiler"]) self.assertNotIn("secret canary", json.dumps(result)) def test_only_explicit_read_only_routes_are_available(self): for path in ["/config.toml", "/.env", "/data/archive/date=2026-11-01/samples.parquet", "/../config.toml", "/solar-relay/.git/config"]: self.assertEqual(self.get(path)[0], 404, path) self.assertEqual(self.get("/api/status?days=91")[0], 400) self.assertEqual(self.get("/api/status?days=1&days=2")[0], 400) self.assertEqual(self.get("/api/status", method="POST")[0], 405) status, headers, body = self.get("/solar-relay/", method="HEAD") self.assertEqual(status, 200) self.assertGreater(int(headers["Content-Length"]), 0) self.assertEqual(body, b"") def test_run_command_starts_and_stops_the_http_listener(self): with server() as (_, inverter): with socket.socket() as free: free.bind(("127.0.0.1", 0)) port = free.getsockname()[1] path = Path(self.directory.name) / "config.toml" path.write_text(f'[collector]\nsource_id="run-test"\narchive="run-archive"\n[inverter]\nbase_url="{inverter}"\n[http]\nport={port}\n') # Keep the first scheduled poll in the future regardless of the real # clock. Scheduling itself is exercised with a simulated clock separately. entry = 'import sys; from solar_relay import cli; cli.next_poll_at = lambda now, interval: now + 300; sys.exit(cli.main())' process = subprocess.Popen([sys.executable, "-c", entry, "--config", str(path), "run"], cwd=Path(__file__).resolve().parents[1], stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True) try: for attempt in range(30): try: with self.http.open(f"http://127.0.0.1:{port}/healthz", timeout=1) as response: self.assertEqual(response.code, 200) break except URLError: time.sleep(0.1) else: self.fail("Status listener did not start") with self.http.open(f"http://127.0.0.1:{port}/api/status", timeout=1) as response: snapshot = json.load(response) self.assertEqual(snapshot["status"], "waiting") self.assertEqual(snapshot["archive"]["samples"], 0) process.terminate() _, stderr = process.communicate(timeout=5) self.assertEqual(process.returncode, 0, stderr) self.assertEqual(Store(path.parent / "run-archive").status()["samples"], 0) finally: if process.poll() is None: process.kill() process.communicate(timeout=5)