from __future__ import annotations import csv import io import json import os import tempfile import unittest from datetime import datetime from pathlib import Path from unittest.mock import patch from common.security import TokenStore from common import ApiError, ConfigurationError from common.client import ApiResponse from point_alarm import ( PointRegistrationProcessor, PointUnregisterProcessor, PointUpdateProcessor, build_update_payload, 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_gives_clear_error(self) -> None: service = FakeService([registration(1)], forbid_account=True) with self.assertRaisesRegex(ConfigurationError, "--account-id vermeyin"): fetch_pages(service, self.store, account_id="acc") summary = fetch_pages(service, self.store, account_id=None) self.assertEqual(summary.distinct, 1) 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, {"registrations": [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) REG_ID = "f7587d9e-2481-4b4c-818d-c8d1946851b7" class FakeRegisterService: def __init__(self, env_file: Path, results: list) -> None: self.client = type("C", (), {"token_store": TokenStore(env_file)})() self.results = list(results) self.payloads: list[dict] = [] self.recipient_hits: list = [] def get_account_id(self): return "acc" def register(self, payload, key): self.payloads.append(payload) result = self.results.pop(0) if isinstance(result, Exception): raise result return result def get_by_recipient_id(self, recipient_id): return ApiResponse(200, {"registrations": self.recipient_hits}) class RegisterTest(unittest.TestCase): def setUp(self) -> None: self.tmp = tempfile.TemporaryDirectory() self.addCleanup(self.tmp.cleanup) self.dir = Path(self.tmp.name) (self.dir / ".env").write_text("", encoding="utf-8") self.filter = self.dir / "filter.json" self.filter.write_text(json.dumps({"lightning": {"type": "PULSE_IN_CLOUD"}}), encoding="utf-8") self.webhook = self.dir / "webhook.json" self.webhook.write_text("{}", encoding="utf-8") self.csv = self.dir / "r.csv" def run_processor(self, rows: str | None, service) -> list[dict]: if rows is not None: self.csv.write_text("recipientId;lat;lng;radius\n" + rows, encoding="utf-8") processor = PointRegistrationProcessor( service, csv_path=self.csv, webhook_config_path=self.webhook, filter_config_path=self.filter ) with patch.dict(os.environ, {"IKLIM_WEBHOOK_INITIALIZED": "true"}): processor.run() with self.csv.open(encoding="utf-8-sig", newline="") as handle: return list(csv.DictReader(handle, delimiter=";")) def test_success_failure_and_skip(self) -> None: service = FakeRegisterService( self.dir / ".env", [ ApiResponse(201, {"registrationId": REG_ID}), ApiError("bad", status_code=400, response_data={"message": "duplicate"}), ], ) rows = self.run_processor("a;39.9;32.8;1000\nb;39.9;32.8;1000\nc;99;32.8;1000\n", service) self.assertEqual([r["status"] for r in rows], ["SUCCESS", "FAILED", "FAILED"]) self.assertEqual(rows[0]["registrationId"], REG_ID) self.assertIn("duplicate", rows[1]["errorBody"]) self.assertEqual(rows[2]["httpStatus"], "NOT_SENT") self.assertEqual(service.payloads[0]["webhook"], {"accountId": "acc"}) self.assertEqual(service.payloads[0]["filter"]["lightning"]["type"], "PULSE_IN_CLOUD") # Re-run: SUCCESS row is skipped, the failed one is retried. service2 = FakeRegisterService( self.dir / ".env", [ApiResponse(201, {"registrationId": REG_ID})] * 2 ) rows = self.run_processor(None, service2) self.assertEqual(len(service2.payloads), 1) self.assertEqual(rows[0]["registrationId"], REG_ID) def test_recovery_requires_matching_point(self) -> None: service = FakeRegisterService(self.dir / ".env", [ApiError("t", status_code=503)]) service.recipient_hits = [dict(registration(1, "a"), registrationId=REG_ID)] rows = self.run_processor("a;39.9;32.8;1000\n", service) self.assertEqual(rows[0]["status"], "SUCCESS") self.assertEqual(rows[0]["registrationId"], REG_ID) other = FakeRegisterService(self.dir / ".env", [ApiError("t", status_code=503)]) other.recipient_hits = [dict(registration(1, "a"), registrationId=REG_ID)] rows = self.run_processor("a;40.0;32.8;1000\n", other) self.assertEqual(rows[0]["status"], "FAILED") class UpdateTest(unittest.TestCase): def test_payload_skips_blank_cells(self) -> None: payload = build_update_payload( {"registrationId": REG_ID, "lat": "39.9", "lng": "32.8", "radius": "", "severityThreshold": "high", "precipitationIntensities": "light|heavy"} ) self.assertEqual(payload["boundary"], {"point": {"lat": 39.9, "lng": 32.8}}) self.assertEqual(payload["filter"]["thunderstorm"], {"severityThreshold": "HIGH"}) self.assertEqual(payload["filter"]["precipitation"]["intensities"], ["LIGHT", "HEAVY"]) def test_payload_errors(self) -> None: for row in ( {"registrationId": REG_ID}, {"registrationId": REG_ID, "lat": "39"}, {"registrationId": "x", "radius": "5"}, {"registrationId": REG_ID, "radius": "60000"}, {"registrationId": REG_ID, "lightningType": "BAD"}, ): with self.assertRaises(ConfigurationError, msg=row): build_update_payload(row) def test_dry_run_does_not_call_or_write(self) -> None: with tempfile.TemporaryDirectory() as tmp: path = Path(tmp) / "u.csv" text = f"registrationId;radius\n{REG_ID};500\n" path.write_text(text, encoding="utf-8") class Boom: def update(self, *_): raise AssertionError("must not be called") PointUpdateProcessor(Boom(), csv_path=path, dry_run=True).run() self.assertEqual(path.read_text(encoding="utf-8"), text) def test_update_records_result(self) -> None: with tempfile.TemporaryDirectory() as tmp: path = Path(tmp) / "u.csv" path.write_text(f"registrationId;radius\n{REG_ID};500\n", encoding="utf-8") class Svc: def update(self, payload, key): return ApiResponse(200, {}) PointUpdateProcessor(Svc(), csv_path=path).run() with path.open(encoding="utf-8-sig", newline="") as handle: row = next(csv.DictReader(handle, delimiter=";")) self.assertEqual((row["status"], row["httpStatus"]), ("SUCCESS", "HTTP_200")) class UnregisterTest(unittest.TestCase): class Svc: def __init__(self) -> None: self.deleted: list[str] = [] def get_by_id(self, value): return ApiResponse(200, dict(registration(1), registrationId=value)) def unregister(self, value): self.deleted.append(value) return ApiResponse(204, None) def make(self, svc, answers, env="test", backup=None): queue = list(answers) return PointUnregisterProcessor( svc, ids=[REG_ID], environment=env, base_url="https://x", backup_path=backup, input_fn=lambda _: queue.pop(0), out=io.StringIO(), ) def test_wrong_confirmation_deletes_nothing(self) -> None: svc = self.Svc() self.assertEqual(self.make(svc, ["sil"]).run(), []) self.assertEqual(svc.deleted, []) def test_prod_needs_environment_name(self) -> None: svc = self.Svc() self.assertEqual(self.make(svc, ["test"], env="prod").run(), []) self.assertEqual(svc.deleted, []) self.assertEqual(len(self.make(svc, ["prod", "SİL"], env="prod").run()), 1) def test_backup_written_before_delete_and_failure_blocks(self) -> None: with tempfile.TemporaryDirectory() as tmp: svc = self.Svc() backup = Path(tmp) / "b.csv" results = self.make(svc, ["SİL"], backup=backup).run() self.assertEqual(results[0]["status"], "SUCCESS") self.assertIn(REG_ID, backup.read_text(encoding="utf-8-sig")) svc2 = self.Svc() with self.assertRaises(OSError): self.make(svc2, ["SİL"], backup=Path(tmp) / "no" / "b.csv").run() self.assertEqual(svc2.deleted, []) if __name__ == "__main__": unittest.main()