qzl
c79c773d67
- Add NotificationTargetMode enum (new_users/exist_users/all_users/user_ids) - Create Alembic migrations: drop duplicate indexes, add target_mode column - Merge register-notifications.sh into dev-migrate.sh sync-notifications subcommand - Shorten notification config path: static/notification/notifications -> static/notifications - Update registration flow to dispatch notifications by target_mode - Add is_first_registration to RegisterBonusResult for first-time user detection - Remove dead code: link_published_notifications_to_user - Update welcome_points.yaml to target new_users only - Add 44 unit tests + 1 integration test, all passing
151 lines
5.2 KiB
Python
151 lines
5.2 KiB
Python
from __future__ import annotations
|
|
|
|
from datetime import datetime
|
|
from uuid import UUID, uuid4
|
|
|
|
import pytest
|
|
|
|
from schemas.enums import NotificationTargetMode
|
|
from v1.notifications.service import NotificationService
|
|
|
|
|
|
class _FakeNotification:
|
|
def __init__(
|
|
self,
|
|
*,
|
|
id: UUID,
|
|
target_mode: NotificationTargetMode = NotificationTargetMode.ALL_USERS,
|
|
status: str = "published",
|
|
deleted_at: datetime | None = None,
|
|
):
|
|
self.id = id
|
|
self.target_mode = target_mode
|
|
self.status = status
|
|
self.deleted_at = deleted_at
|
|
|
|
|
|
class _TrackingNotificationRepository:
|
|
def __init__(self, notifications: list[_FakeNotification]) -> None:
|
|
self._notifications = notifications
|
|
self.linked_notification_ids: list[list[UUID]] = []
|
|
self.linked_is_first: list[bool] = []
|
|
|
|
async def link_notifications_for_registered_user(
|
|
self, *, user_id: UUID, is_first_registration: bool
|
|
) -> int:
|
|
target_modes: list[NotificationTargetMode]
|
|
if is_first_registration:
|
|
target_modes = [
|
|
NotificationTargetMode.NEW_USERS,
|
|
NotificationTargetMode.ALL_USERS,
|
|
]
|
|
else:
|
|
target_modes = [NotificationTargetMode.ALL_USERS]
|
|
|
|
matched = [
|
|
n
|
|
for n in self._notifications
|
|
if n.status == "published"
|
|
and n.deleted_at is None
|
|
and n.target_mode in target_modes
|
|
]
|
|
self.linked_notification_ids.append([n.id for n in matched])
|
|
self.linked_is_first.append(is_first_registration)
|
|
return len(matched)
|
|
|
|
|
|
@pytest.fixture
|
|
def notification_new_users() -> _FakeNotification:
|
|
return _FakeNotification(id=uuid4(), target_mode=NotificationTargetMode.NEW_USERS)
|
|
|
|
|
|
@pytest.fixture
|
|
def notification_all_users() -> _FakeNotification:
|
|
return _FakeNotification(id=uuid4(), target_mode=NotificationTargetMode.ALL_USERS)
|
|
|
|
|
|
@pytest.fixture
|
|
def notification_exist_users() -> _FakeNotification:
|
|
return _FakeNotification(id=uuid4(), target_mode=NotificationTargetMode.EXIST_USERS)
|
|
|
|
|
|
class TestLinkNotificationsForRegisteredUser:
|
|
@pytest.mark.asyncio
|
|
async def test_first_registration_gets_new_users_and_all_users(
|
|
self,
|
|
notification_new_users: _FakeNotification,
|
|
notification_all_users: _FakeNotification,
|
|
notification_exist_users: _FakeNotification,
|
|
) -> None:
|
|
repo = _TrackingNotificationRepository(
|
|
[notification_new_users, notification_all_users, notification_exist_users]
|
|
)
|
|
service = NotificationService(repository=repo) # type: ignore[arg-type]
|
|
|
|
count = await service.link_notifications_for_registered_user(
|
|
user_id=uuid4(), is_first_registration=True
|
|
)
|
|
|
|
assert count == 2
|
|
linked_ids = repo.linked_notification_ids[0]
|
|
assert notification_new_users.id in linked_ids
|
|
assert notification_all_users.id in linked_ids
|
|
assert notification_exist_users.id not in linked_ids
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_reregistered_user_only_gets_all_users(
|
|
self,
|
|
notification_new_users: _FakeNotification,
|
|
notification_all_users: _FakeNotification,
|
|
notification_exist_users: _FakeNotification,
|
|
) -> None:
|
|
repo = _TrackingNotificationRepository(
|
|
[notification_new_users, notification_all_users, notification_exist_users]
|
|
)
|
|
service = NotificationService(repository=repo) # type: ignore[arg-type]
|
|
|
|
count = await service.link_notifications_for_registered_user(
|
|
user_id=uuid4(), is_first_registration=False
|
|
)
|
|
|
|
assert count == 1
|
|
linked_ids = repo.linked_notification_ids[0]
|
|
assert notification_new_users.id not in linked_ids
|
|
assert notification_all_users.id in linked_ids
|
|
assert notification_exist_users.id not in linked_ids
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_no_published_notifications_returns_zero(self) -> None:
|
|
repo = _TrackingNotificationRepository([])
|
|
service = NotificationService(repository=repo) # type: ignore[arg-type]
|
|
|
|
count = await service.link_notifications_for_registered_user(
|
|
user_id=uuid4(), is_first_registration=True
|
|
)
|
|
|
|
assert count == 0
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_only_new_users_notification_first_registration(self) -> None:
|
|
n = _FakeNotification(id=uuid4(), target_mode=NotificationTargetMode.NEW_USERS)
|
|
repo = _TrackingNotificationRepository([n])
|
|
service = NotificationService(repository=repo) # type: ignore[arg-type]
|
|
|
|
count = await service.link_notifications_for_registered_user(
|
|
user_id=uuid4(), is_first_registration=True
|
|
)
|
|
|
|
assert count == 1
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_only_new_users_notification_reregistered(self) -> None:
|
|
n = _FakeNotification(id=uuid4(), target_mode=NotificationTargetMode.NEW_USERS)
|
|
repo = _TrackingNotificationRepository([n])
|
|
service = NotificationService(repository=repo) # type: ignore[arg-type]
|
|
|
|
count = await service.link_notifications_for_registered_user(
|
|
user_id=uuid4(), is_first_registration=False
|
|
)
|
|
|
|
assert count == 0
|