97 lines
2.8 KiB
Python
97 lines
2.8 KiB
Python
|
from __future__ import annotations
|
||
|
|
|
||
|
|
import time
|
||
|
|
from typing import TypedDict
|
||
|
|
from uuid import UUID
|
||
|
|
|
||
|
|
import httpx
|
||
|
|
import pytest
|
||
|
|
from sqlalchemy import select
|
||
|
|
|
||
|
|
from core.db.session import AsyncSessionLocal
|
||
|
|
from models.notification import Notification
|
||
|
|
from models.user_notification import UserNotification
|
||
|
|
|
||
|
|
|
||
|
|
class IdentityData(TypedDict):
|
||
|
|
email: str
|
||
|
|
code: str
|
||
|
|
|
||
|
|
|
||
|
|
async def _create_email_session(
|
||
|
|
client: httpx.AsyncClient,
|
||
|
|
*,
|
||
|
|
email: str,
|
||
|
|
code: str,
|
||
|
|
) -> dict[str, object]:
|
||
|
|
resp = await client.post(
|
||
|
|
"/api/v1/auth/email-session",
|
||
|
|
json={"email": email, "token": code},
|
||
|
|
)
|
||
|
|
resp.raise_for_status()
|
||
|
|
return resp.json()
|
||
|
|
|
||
|
|
|
||
|
|
async def _delete_user(client: httpx.AsyncClient, *, token: str) -> None:
|
||
|
|
resp = await client.delete(
|
||
|
|
"/api/v1/users/me",
|
||
|
|
headers={"Authorization": f"Bearer {token}"},
|
||
|
|
)
|
||
|
|
assert resp.status_code == 204
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_notification_target_mode_first_reg_and_reregister(
|
||
|
|
api_client: httpx.AsyncClient,
|
||
|
|
test_identity: IdentityData,
|
||
|
|
db_cleanup: list[str],
|
||
|
|
) -> None:
|
||
|
|
email = str(test_identity["email"]).strip().lower()
|
||
|
|
db_cleanup.append(email)
|
||
|
|
|
||
|
|
first = await _create_email_session(
|
||
|
|
api_client, email=email, code=str(test_identity["code"])
|
||
|
|
)
|
||
|
|
user1 = first.get("user")
|
||
|
|
assert isinstance(user1, dict)
|
||
|
|
user1_id = UUID(str(user1["id"]))
|
||
|
|
token1 = str(first["access_token"])
|
||
|
|
|
||
|
|
async with AsyncSessionLocal() as session:
|
||
|
|
result = await session.execute(
|
||
|
|
select(Notification.target_mode)
|
||
|
|
.join(UserNotification, UserNotification.notification_id == Notification.id)
|
||
|
|
.where(UserNotification.user_id == user1_id)
|
||
|
|
.order_by(Notification.target_mode)
|
||
|
|
)
|
||
|
|
first_target_modes = [str(row[0]) for row in result.all()]
|
||
|
|
|
||
|
|
assert "new_users" in first_target_modes
|
||
|
|
assert "exist_users" not in first_target_modes
|
||
|
|
|
||
|
|
await _delete_user(api_client, token=token1)
|
||
|
|
time.sleep(0.5)
|
||
|
|
|
||
|
|
second = await _create_email_session(
|
||
|
|
api_client, email=email, code=str(test_identity["code"])
|
||
|
|
)
|
||
|
|
user2 = second.get("user")
|
||
|
|
assert isinstance(user2, dict)
|
||
|
|
user2_id = UUID(str(user2["id"]))
|
||
|
|
token2 = str(second["access_token"])
|
||
|
|
|
||
|
|
async with AsyncSessionLocal() as session:
|
||
|
|
result = await session.execute(
|
||
|
|
select(Notification.target_mode)
|
||
|
|
.join(UserNotification, UserNotification.notification_id == Notification.id)
|
||
|
|
.where(UserNotification.user_id == user2_id)
|
||
|
|
.order_by(Notification.target_mode)
|
||
|
|
)
|
||
|
|
second_target_modes = [str(row[0]) for row in result.all()]
|
||
|
|
|
||
|
|
assert "new_users" not in second_target_modes
|
||
|
|
assert "all_users" not in second_target_modes
|
||
|
|
assert "exist_users" not in second_target_modes
|
||
|
|
|
||
|
|
await _delete_user(api_client, token=token2)
|