60 lines
1.5 KiB
Python
60 lines
1.5 KiB
Python
import os
|
|
import tempfile
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
import pytest
|
|
|
|
from app.models.event import EventType, OutboundEvent
|
|
from app.repositories.event_store import EventStore
|
|
|
|
|
|
@pytest.fixture
|
|
def store():
|
|
fd, path = tempfile.mkstemp(suffix=".db")
|
|
os.close(fd)
|
|
store = EventStore(path)
|
|
yield store
|
|
os.unlink(path)
|
|
|
|
|
|
def test_save_and_fetch_pending(store):
|
|
event = OutboundEvent(
|
|
event_type=EventType.HEARTBEAT,
|
|
account_id="bot-001",
|
|
tenant_code="rescue",
|
|
data={"online_status": "online"},
|
|
)
|
|
store.save(event)
|
|
pending = store.fetch_pending(limit=10)
|
|
assert len(pending) == 1
|
|
assert pending[0].event_id == event.event_id
|
|
|
|
|
|
def test_mark_delivered(store):
|
|
event = OutboundEvent(
|
|
event_type=EventType.HEARTBEAT,
|
|
account_id="bot-001",
|
|
tenant_code="rescue",
|
|
data={},
|
|
)
|
|
store.save(event)
|
|
store.mark_delivered(event.event_id)
|
|
pending = store.fetch_pending(limit=10)
|
|
assert len(pending) == 0
|
|
|
|
|
|
def test_cleanup_old_delivered_events(store):
|
|
old_event = OutboundEvent(EventType.HEARTBEAT, "bot-001", "rescue", {})
|
|
store.save(old_event)
|
|
store.mark_delivered(old_event.event_id)
|
|
|
|
# Modify created_at to be 11 days ago.
|
|
with store._connection() as conn:
|
|
conn.execute(
|
|
"UPDATE outbound_events SET created_at = ? WHERE id = ?",
|
|
(datetime.now(timezone.utc) - timedelta(days=11), old_event.event_id),
|
|
)
|
|
|
|
deleted = store.cleanup(10)
|
|
assert deleted == 1
|