move_detect/tests/test_queue.py

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]