feat(notification): add target_mode enum constraint and merge register-notifications script
- 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
This commit is contained in:
@@ -0,0 +1,150 @@
|
||||
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
|
||||
Reference in New Issue
Block a user