48 lines
1.5 KiB
Python
48 lines
1.5 KiB
Python
import asyncio
|
|
from datetime import UTC, datetime, timedelta
|
|
from types import SimpleNamespace
|
|
|
|
from schemas.enums import NotificationType
|
|
from services.notifications import collect_subscription_notifications
|
|
|
|
|
|
class FakeUnitOfWork:
|
|
def __init__(self, subscriptions_by_period):
|
|
self.session = self
|
|
self.subscriptions_by_period = iter(subscriptions_by_period)
|
|
self.executed = []
|
|
self.committed = False
|
|
|
|
async def scalars(self, _statement):
|
|
return next(self.subscriptions_by_period)
|
|
|
|
async def execute(self, statement):
|
|
self.executed.append(statement)
|
|
return SimpleNamespace(rowcount=1)
|
|
|
|
async def commit(self):
|
|
self.committed = True
|
|
|
|
|
|
def test_collects_notification_for_each_expiry_period():
|
|
now = datetime(2026, 8, 20, tzinfo=UTC)
|
|
uow = FakeUnitOfWork(
|
|
[
|
|
[SimpleNamespace(id=1, expires_at=now + timedelta(days=6))],
|
|
[SimpleNamespace(id=2, expires_at=now + timedelta(days=2))],
|
|
[SimpleNamespace(id=3, expires_at=now + timedelta(hours=12))],
|
|
[SimpleNamespace(id=4, expires_at=now - timedelta(hours=1))],
|
|
]
|
|
)
|
|
|
|
created = asyncio.run(collect_subscription_notifications(uow, now))
|
|
|
|
assert created == len(uow.executed)
|
|
assert uow.committed
|
|
assert [statement.compile().params["notify_type_m0"] for statement in uow.executed] == [
|
|
NotificationType.SEVEN_DAYS,
|
|
NotificationType.THREE_DAYS,
|
|
NotificationType.ONE_DAY,
|
|
NotificationType.EXPIRED,
|
|
]
|