from __future__ import annotations import csv import tempfile import unittest from datetime import datetime from pathlib import Path from common import ApiError, ConfigurationError from common.client import ApiResponse from point_alarm import ( PointStore, fetch_pages, flatten_registration, load_point_alarm_filter, lookup_rows, read_id_list, mask_secrets, parse_polygon, timestamped_path, write_records, ) def registration(index: int, recipient: str = "r1") -> dict: return { "registrationId": f"id-{index}", "recipientId": recipient, "boundary": {"point": {"lat": 39.9, "lng": 32.8}, "radius": 1000}, "filter": { "lightning": {"type": "PULSE_IN_CLOUD", "peakCurrent": 5}, "thunderstorm": {"intersectsCellPolygon": True, "severityThreshold": "LOW"}, "precipitation": {"intensities": ["LIGHT", "HEAVY"]}, }, "webhook": {"accountId": "acc", "url": "https://x", "authentication": {"password": "p"}}, } class FakeService: def __init__(self, registrations: list[dict], forbid_account: bool = False) -> None: self.registrations = registrations self.forbid_account = forbid_account self.calls: list[dict] = [] def page(self, *, page_number, page_size, account_id=None, **_): self.calls.append({"page": page_number, "account": account_id}) if self.forbid_account and account_id: raise ApiError("forbidden", status_code=403) start = page_number * page_size return { "registrations": self.registrations[start : start + page_size], "totalRecords": len(self.registrations), } class PointAlarmTest(unittest.TestCase): def setUp(self) -> None: self.tmp = tempfile.TemporaryDirectory() self.addCleanup(self.tmp.cleanup) self.dir = Path(self.tmp.name) self.store = PointStore(self.dir / "t.db") self.addCleanup(self.store.close) def test_flatten(self) -> None: flat = flatten_registration(registration(1)) self.assertEqual(flat["precipitationIntensities"], "LIGHT|HEAVY") self.assertEqual(flat["intersectsCellPolygon"], "true") self.assertIsNone(flat["speedThreshold"]) def test_mask(self) -> None: masked = mask_secrets(registration(1)["webhook"]) self.assertEqual(masked["authentication"]["password"], "***") self.assertEqual(masked["url"], "https://x") def test_fetch_all_pages_upserts_without_duplicates(self) -> None: service = FakeService([registration(i) for i in range(5)]) summary = fetch_pages(service, self.store, account_id="acc", page_size=2) self.assertEqual(summary.distinct, 5) self.assertEqual(summary.missing, 0) fetch_pages(service, self.store, account_id="acc", page_size=2) self.assertEqual(self.store.count(), 5) self.assertEqual(self.store.webhooks("acc")[0]["authentication"]["password"], "***") def test_single_page(self) -> None: service = FakeService([registration(i) for i in range(5)]) summary = fetch_pages(service, self.store, account_id="acc", page_size=2, page_number=1) self.assertEqual(summary.received, 2) self.assertEqual(len(service.calls), 1) def test_forbidden_account_retries_without_it(self) -> None: service = FakeService([registration(1)], forbid_account=True) summary = fetch_pages(service, self.store, account_id="acc") self.assertEqual(summary.distinct, 1) self.assertIsNone(service.calls[-1]["account"]) def test_csv_and_txt_output(self) -> None: fetch_pages(FakeService([registration(1)]), self.store, account_id="acc") records = self.store.records("acc") write_records(self.dir / "o.csv", records, self.store.webhooks("acc")) with (self.dir / "o.csv").open(encoding="utf-8-sig", newline="") as handle: rows = list(csv.reader(handle, delimiter=";")) self.assertEqual(rows[0][0], "#webhook") self.assertEqual(rows[1][0], "registrationId") self.assertEqual(rows[2][0], "id-1") write_records(self.dir / "o.txt", records, []) self.assertIn('"registrationId": "id-1"', (self.dir / "o.txt").read_text(encoding="utf-8")) with self.assertRaises(ConfigurationError): write_records(self.dir / "o.json", records, []) def test_timestamped_path(self) -> None: path = timestamped_path(Path("a/out.csv"), datetime(2026, 10, 23, 9, 5, 7)) self.assertEqual(path, Path("a/out-20261023-090507.csv")) def test_polygon(self) -> None: ring = parse_polygon("0,0;1,0;1,1;0,0") self.assertEqual(ring[0], [0.0, 0.0]) with self.assertRaises(ConfigurationError): parse_polygon("0,0;1,0;1,1") def test_filter_validation(self) -> None: good = load_point_alarm_filter(Path("nowcast-point-alarm-filter.example.json")) self.assertIn("lightning", good) bad = self.dir / "bad.json" bad.write_text('{"thunderstorm": {"severityThreshold": "X"}}', encoding="utf-8") with self.assertRaises(ConfigurationError): load_point_alarm_filter(bad) def test_read_id_list(self) -> None: txt = self.dir / "ids.txt" txt.write_text("# c\n\na\nb\na\n", encoding="utf-8") self.assertEqual(read_id_list(txt, "registrationId"), ["a", "b"]) csv_file = self.dir / "ids.csv" csv_file.write_text("x;recipientId\n1;r1\n2;r2\n", encoding="utf-8") self.assertEqual(read_id_list(csv_file, "recipientId"), ["r1", "r2"]) with self.assertRaises(ConfigurationError): read_id_list(csv_file, "registrationId") def test_lookup_rows(self) -> None: class Svc: def get_by_id(self, value): if value == "missing": raise ApiError("nf", status_code=404, response_data={"m": 1}) if value == "boom": raise ApiError("e", status_code=500, response_data="bad") return ApiResponse(200, registration(1)) def get_by_recipient_id(self, value): return ApiResponse(200, [registration(1), registration(2)]) rows, _ = lookup_rows(Svc(), self.store, ["ok", "missing", "boom"], "id", None) self.assertEqual([r["status"] for r in rows], ["SUCCESS", "NOT_FOUND", "FAILED"]) self.assertEqual(rows[2]["errorBody"], '"bad"') rows, _ = lookup_rows(Svc(), self.store, ["r1"], "recipient", None) self.assertEqual(len(rows), 2) self.assertEqual(self.store.count(), 2) if __name__ == "__main__": unittest.main()