feat: add nowcast point alarm unregister client with confirmation and backup
Shows environment, count and an irreversibility warning, requires typing the confirmation word (plus the environment name on prod), backs every registration up through get-by-id before deleting and aborts the whole run if the backup cannot be completed.
This commit is contained in:
@@ -1,6 +1,7 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import csv
|
||||
import io
|
||||
import json
|
||||
import os
|
||||
import tempfile
|
||||
@@ -15,6 +16,7 @@ from common import ApiError, ConfigurationError
|
||||
from common.client import ApiResponse
|
||||
from point_alarm import (
|
||||
PointRegistrationProcessor,
|
||||
PointUnregisterProcessor,
|
||||
PointUpdateProcessor,
|
||||
build_update_payload,
|
||||
PointStore,
|
||||
@@ -294,5 +296,48 @@ class UpdateTest(unittest.TestCase):
|
||||
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()
|
||||
|
||||
Reference in New Issue
Block a user