feat: add nowcast point alarm update client with dry-run

Patches registrations from CSV rows, sending only non-blank cells, validating lat/lng pairing and ranges locally, warning about the speedThreshold PATCH gap (GAIA-476) and recording status, httpStatus and errorBody back into the CSV.
This commit is contained in:
2026-10-23 19:13:35 +03:00
parent ba249128a1
commit 8587669ea5
4 changed files with 290 additions and 0 deletions
+51
View File
@@ -15,6 +15,8 @@ from common import ApiError, ConfigurationError
from common.client import ApiResponse
from point_alarm import (
PointRegistrationProcessor,
PointUpdateProcessor,
build_update_payload,
PointStore,
fetch_pages,
flatten_registration,
@@ -243,5 +245,54 @@ class RegisterTest(unittest.TestCase):
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"))
if __name__ == "__main__":
unittest.main()