import json import sqlite3 from edge.events import Event, EventType, Severity from edge.queue import OfflineQueue from edge.alert import SendResult def _event(n=1): return Event(EventType.FALL, Severity.HIGH, "cam-01", n, None, float(n), 3.0, {}, None) class _Client: def __init__(self, fail=False): self.fail = fail self.sent = [] def send(self, event, snapshots): if self.fail: return False self.sent.append(event.track_id) return True def test_try_send_success_not_queued(tmp_path): q = OfflineQueue(str(tmp_path / "q.db"), _Client()) assert q.try_send(_event(1), [b"x"]) is True assert q.pending() == 0 def test_failure_is_queued(tmp_path): q = OfflineQueue(str(tmp_path / "q.db"), _Client(fail=True)) assert q.try_send(_event(1), [b"x"]) is False assert q.pending() == 1 def test_flush_resends_when_recovered(tmp_path): client = _Client(fail=True) q = OfflineQueue(str(tmp_path / "q.db"), client) q.try_send(_event(1), [b"x"]) q.try_send(_event(2), [b"y"]) assert q.pending() == 2 client.fail = False assert q.flush() == 2 assert q.pending() == 0 assert client.sent == [1, 2] class _PoisonClient: """Rejects one specific event forever with a definite 4xx the server will never accept.""" def __init__(self, poison_id): self.poison = poison_id self.sent = [] def send(self, event, snapshots): if event.track_id == self.poison: return SendResult.REJECTED self.sent.append(event.track_id) return SendResult.DELIVERED def test_queue_is_bounded_and_keeps_newest(tmp_path): # An extended outage (or a permanent misconfig) must not fill the SD card. The queue caps # itself, evicting the OLDEST — a fresh fall alert matters more than a stale one. q = OfflineQueue(str(tmp_path / "q.db"), _Client(fail=True), max_rows=5) for n in range(20): q.try_send(_event(n), [b"x"]) assert q.pending() == 5 q.client.fail = False q.flush() assert q.client.sent == [15, 16, 17, 18, 19] # the newest 5 survived def test_event_uuid_survives_the_queue_roundtrip(tmp_path): # A re-sent event (failed once -> queued -> flushed) must carry the SAME event_uuid so the # server dedups it instead of storing a duplicate alert. This is the whole point of the key. class _CaptureClient: def __init__(self): self.fail = True self.uuids = [] def send(self, event, snapshots): self.uuids.append(event.event_uuid) return not self.fail client = _CaptureClient() q = OfflineQueue(str(tmp_path / "q.db"), client) ev = _event(1) q.try_send(ev, [b"x"]) # fails -> queued (1st send attempt) client.fail = False assert q.flush() == 1 # re-sent from the queue (2nd send attempt) assert len(client.uuids) == 2 assert client.uuids[0] == client.uuids[1] == ev.event_uuid def test_poison_head_is_dropped_and_unblocks_the_queue(tmp_path): # A permanently-rejected event at the head must not wedge the whole queue forever; after # max_send_attempts it is dropped so the events behind it can drain. client = _PoisonClient(99) q = OfflineQueue(str(tmp_path / "q.db"), client, max_send_attempts=3) q.enqueue(_event(99), [b"x"]) # poison, at the head q.enqueue(_event(2), [b"y"]) # good, behind it for _ in range(3): q.flush() assert client.sent == [2] # good event delivered assert q.pending() == 0 # poison dropped, queue drained def test_sustained_outage_never_drops_events(tmp_path): # A transient outage (network down / 5xx) must NEVER poison-drop queued events, no matter how # long it lasts — a real fall alert has to survive an overnight outage. Only a definite 4xx # rejection counts toward the drop cap (contrast test_poison_head_is_dropped above). class _Outage: up = False sent = [] def send(self, event, snapshots): if not _Outage.up: return SendResult.TRANSIENT _Outage.sent.append(event.track_id) return SendResult.DELIVERED client = _Outage() q = OfflineQueue(str(tmp_path / "q.db"), client, max_send_attempts=3) q.enqueue(_event(1), [b"x"]) for _ in range(10): # far beyond max_send_attempts q.flush() assert q.pending() == 1 # still queued — a transient failure is never poison-dropped _Outage.up = True q.flush() assert q.pending() == 0 and _Outage.sent == [1] # drains once connectivity returns def test_undecodable_row_is_dropped_and_unblocks_the_queue(tmp_path): # A corrupt row (torn SD-card write / schema drift) can never be decoded, so it must not wedge # every queued safety event behind it forever. flush() drops the single bad row and drains the # rest — losing one unrecoverable event beats freezing the whole queue. db = str(tmp_path / "q.db") con = sqlite3.connect(db) con.execute("CREATE TABLE queue (id INTEGER PRIMARY KEY AUTOINCREMENT, payload TEXT, " "snaps TEXT, attempts INTEGER DEFAULT 0)") con.execute("INSERT INTO queue (payload, snaps) VALUES (?, ?)", ("{not valid json", json.dumps([]))) # poison row at the head good = {"event_type": "fall", "severity": "high", "camera_id": "cam-01", "track_id": 7, "zone_id": None, "occurred_at": 100.0, "duration_s": 3.0, "metrics": {}, "spatial": None, "event_uuid": "u7"} con.execute("INSERT INTO queue (payload, snaps) VALUES (?, ?)", (json.dumps(good), json.dumps([]))) # good row behind it con.commit(); con.close() client = _Client() q = OfflineQueue(db, client) assert q.flush() == 1 # the good row drained past the dropped poison row assert client.sent == [7] assert q.pending() == 0 # both rows gone: poison dropped, good delivered def test_legacy_row_without_uuid_reuses_minted_uuid_across_retries(tmp_path): # A row queued before event_uuid existed has none stored. On first flush a fresh uuid is # minted; it MUST be persisted so every retry reuses the SAME key — else each failed-then- # retried delivery sends a different uuid and the server stores duplicates (the P1-9 bug). db = str(tmp_path / "q.db") con = sqlite3.connect(db) con.execute("CREATE TABLE queue (id INTEGER PRIMARY KEY AUTOINCREMENT, payload TEXT, " "snaps TEXT, attempts INTEGER DEFAULT 0)") legacy = {"event_type": "fall", "severity": "high", "camera_id": "cam-01", "track_id": 1, "zone_id": None, "occurred_at": 100.0, "duration_s": 3.0, "metrics": {}, "spatial": None} con.execute("INSERT INTO queue (payload, snaps) VALUES (?, ?)", (json.dumps(legacy), json.dumps([]))) con.commit(); con.close() class _Capture: up = False uuids = [] def send(self, event, snapshots): _Capture.uuids.append(event.event_uuid) return SendResult.DELIVERED if _Capture.up else SendResult.TRANSIENT q = OfflineQueue(db, _Capture()) q.flush() # transient -> row stays; uuid minted and persisted _Capture.up = True q.flush() # delivered assert len(_Capture.uuids) == 2 assert _Capture.uuids[0] == _Capture.uuids[1] # same key reused across retries def test_flush_drains_in_bounded_slices(tmp_path): # flush() runs inside the single-threaded detection loop, so an unbounded drain after a long # outage would stall fall detection for minutes at the worst possible moment. Each call must # send at most max_flush_rows and leave the rest for the next tick. q = OfflineQueue(str(tmp_path / "q.db"), _Client(), max_flush_rows=3) for n in range(10): q.enqueue(_event(n), []) assert q.pending() == 10 assert q.flush() == 3 # one slice, not the whole queue assert q.pending() == 7 q.flush(); q.flush() assert q.pending() == 1 # keeps draining across ticks q.flush() assert q.pending() == 0 def test_flush_slice_preserves_fifo_order(tmp_path): # Bounding the slice must not reorder delivery: the oldest queued alert still goes first. client = _Client() q = OfflineQueue(str(tmp_path / "q.db"), client, max_flush_rows=2) for n in range(4): q.enqueue(_event(n), []) q.flush(); q.flush() assert client.sent == [0, 1, 2, 3]