leocrm-bot
119 lines
4.0 KiB
Python
119 lines
4.0 KiB
Python
"""Notification tests — ACs 24-26: list, mark read, unread count."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import UTC, datetime
|
|
|
|
import pytest
|
|
from httpx import AsyncClient
|
|
|
|
from app.core.notifications import create_notification
|
|
from app.models.notification import Notification
|
|
from tests.conftest import ORIGIN_HEADER, login_client, seed_tenant_and_users
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
class TestNotifications:
|
|
"""ACs 24-26: Notification list, mark read, unread count."""
|
|
|
|
async def test_list_notifications_returns_200(self, client: AsyncClient, db_session):
|
|
"""AC 24: GET /api/v1/notifications -> 200 + unread first."""
|
|
seed = await seed_tenant_and_users(db_session)
|
|
# Create some notifications for admin_a
|
|
await create_notification(
|
|
db_session,
|
|
seed["tenant_a"].id,
|
|
seed["admin_a"].id,
|
|
"info",
|
|
"Read Notif",
|
|
"Already read",
|
|
)
|
|
await db_session.flush()
|
|
# Mark the first one as read
|
|
from sqlalchemy import select
|
|
|
|
q = select(Notification).where(Notification.title == "Read Notif")
|
|
result = await db_session.execute(q)
|
|
first_notif = result.scalar_one()
|
|
first_notif.read_at = datetime.now(UTC)
|
|
await db_session.flush()
|
|
|
|
# Create an unread one
|
|
await create_notification(
|
|
db_session,
|
|
seed["tenant_a"].id,
|
|
seed["admin_a"].id,
|
|
"info",
|
|
"Unread Notif",
|
|
"Not read yet",
|
|
)
|
|
await db_session.commit()
|
|
|
|
await login_client(client, "admin@tenanta.com")
|
|
resp = await client.get("/api/v1/notifications")
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert "items" in data
|
|
assert "total" in data
|
|
# Unread should come first
|
|
items = data["items"]
|
|
if len(items) >= 2:
|
|
# Find the unread and read ones
|
|
unread = [i for i in items if i["read_at"] is None]
|
|
read = [i for i in items if i["read_at"] is not None]
|
|
if unread and read:
|
|
# Unread should appear before read in the list
|
|
unread_idx = items.index(unread[0])
|
|
read_idx = items.index(read[0])
|
|
assert unread_idx < read_idx
|
|
|
|
async def test_mark_notification_read_returns_200(self, client: AsyncClient, db_session):
|
|
"""AC 25: PATCH /api/v1/notifications/{id}/read -> 200."""
|
|
seed = await seed_tenant_and_users(db_session)
|
|
notif = await create_notification(
|
|
db_session,
|
|
seed["tenant_a"].id,
|
|
seed["admin_a"].id,
|
|
"info",
|
|
"Test Notif",
|
|
"Test body",
|
|
)
|
|
await db_session.commit()
|
|
|
|
await login_client(client, "admin@tenanta.com")
|
|
resp = await client.patch(
|
|
f"/api/v1/notifications/{notif.id}/read",
|
|
headers=ORIGIN_HEADER,
|
|
)
|
|
assert resp.status_code == 200
|
|
assert resp.json()["read_at"] is not None
|
|
|
|
async def test_unread_count_returns_200_with_integer(self, client: AsyncClient, db_session):
|
|
"""AC 26: GET /api/v1/notifications/unread-count -> 200 + integer count."""
|
|
seed = await seed_tenant_and_users(db_session)
|
|
await create_notification(
|
|
db_session,
|
|
seed["tenant_a"].id,
|
|
seed["admin_a"].id,
|
|
"info",
|
|
"Unread 1",
|
|
"Body 1",
|
|
)
|
|
await create_notification(
|
|
db_session,
|
|
seed["tenant_a"].id,
|
|
seed["admin_a"].id,
|
|
"info",
|
|
"Unread 2",
|
|
"Body 2",
|
|
)
|
|
await db_session.commit()
|
|
|
|
await login_client(client, "admin@tenanta.com")
|
|
resp = await client.get("/api/v1/notifications/unread-count")
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert "count" in data
|
|
assert isinstance(data["count"], int)
|
|
assert data["count"] >= 2
|