36 lines
1.1 KiB
Python
36 lines
1.1 KiB
Python
from edge.events import Event, EventType, Severity
|
|
from edge.queue import OfflineQueue
|
|
|
|
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]
|