leocrm-bot
3ab4925783
- 10 models: tenants, users, user_tenants, roles, sessions, audit_log, deletion_log, notifications, password_reset_tokens, api_tokens - Session-based auth (Redis + PostgreSQL audit trail) - Multi-tenant with ORM-level filtering + PostgreSQL RLS (set_config) - RBAC with roles/permissions + field-level permissions - CSRF protection via Origin header validation - Auth rate limiting (Redis counters with TTL) - CORS with explicit origins (no wildcard) - Health endpoint (no auth required) - Notification service + audit log middleware - 29 tests, 26 ACs, all passing - Coverage: 62% (infrastructure modules pending coverage in later tasks)
100 lines
3.8 KiB
Python
100 lines
3.8 KiB
Python
"""Notification tests — ACs 24-26: list, mark read, unread count."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import uuid
|
|
from datetime import datetime, timezone
|
|
|
|
import pytest
|
|
from httpx import AsyncClient
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from tests.conftest import ORIGIN_HEADER, seed_tenant_and_users, login_client
|
|
from app.core.notifications import create_notification
|
|
from app.models.notification import Notification
|
|
|
|
|
|
@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, update
|
|
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(timezone.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
|