48 lines
1.4 KiB
Python
48 lines
1.4 KiB
Python
import pytest
|
|
|
|
from traderai.memory import MemoryStore
|
|
from traderai.scheduler import WakeScheduler
|
|
|
|
|
|
class FakeUEXNotifications:
|
|
def __init__(self):
|
|
self.calls = 0
|
|
|
|
async def get_user_notifications(self):
|
|
self.calls += 1
|
|
return {
|
|
"status": "ok",
|
|
"notifications": [
|
|
{
|
|
"id": 10,
|
|
"message": "A buyer replied to your listing.",
|
|
"redir": "/marketplace/negotiations/abc",
|
|
"code": "negotiation_reply",
|
|
"date_added": 123,
|
|
"date_read": 0,
|
|
},
|
|
{
|
|
"id": 11,
|
|
"message": "Already read.",
|
|
"date_added": 122,
|
|
"date_read": 123,
|
|
},
|
|
],
|
|
}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_poll_uex_notifications_adds_unread_once(tmp_path):
|
|
memory = MemoryStore(str(tmp_path / "memory.sqlite3"))
|
|
scheduler = WakeScheduler(memory)
|
|
scheduler.bind_uex_notifications(FakeUEXNotifications())
|
|
|
|
first = await scheduler.poll_uex_notifications()
|
|
second = await scheduler.poll_uex_notifications()
|
|
outbox = memory.inspect()["outbox"]
|
|
|
|
assert len(first) == 1
|
|
assert second == []
|
|
assert len(outbox) == 1
|
|
assert "A buyer replied to your listing." in outbox[0]["content"]
|