197 lines
9.8 KiB
Python
197 lines
9.8 KiB
Python
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)
|