mozdemir
ba249128a1
Registers one point alarm per CSV row (recipientId, lat, lng, radius) with the shared filter JSON and the existing webhook bootstrap, records status, httpStatus, registrationId and errorBody back into the CSV and recovers ambiguous failures only on an identical recipient, point and radius match.
248 lines
10 KiB
Python
248 lines
10 KiB
Python
from __future__ import annotations
|
|
|
|
import csv
|
|
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,
|
|
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, {"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")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|