leocrm/tests/test_notifications.py

"""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


# ── Cross-tenant isolation + RBAC tests ──


@pytest.mark.asyncio
class TestNotificationCrossTenant:
    """Notifications must not leak across tenants."""

    async def test_cross_tenant_isolation(self, client: AsyncClient, db_session):
        """Notification for tenant A user is not visible to tenant B admin."""
        from httpx import ASGITransport
        from httpx import AsyncClient as AC

        import app.main

        seed = await seed_tenant_and_users(db_session)
        await create_notification(
            db_session,
            seed["tenant_a"].id,
            seed["admin_a"].id,
            "info",
            "Tenant A Notif",
            "Body",
        )
        await db_session.commit()

        await login_client(client, "admin@tenanta.com")
        resp = await client.get("/api/v1/notifications")
        assert resp.status_code == 200
        assert any(i["title"] == "Tenant A Notif" for i in resp.json()["items"])

        # Tenant B admin must not see tenant A's notification
        app_instance = app.main.app
        async with AC(transport=ASGITransport(app=app_instance), base_url="http://test") as client_b:
            await login_client(client_b, "admin@tenantb.com")
            resp_b = await client_b.get("/api/v1/notifications")
            assert resp_b.status_code == 200
            assert all(i["title"] != "Tenant A Notif" for i in resp_b.json()["items"])

    async def test_rbac_no_permission(self, client: AsyncClient, db_session):
        """User without notifications:read permission gets 403 on list."""
        from tests.conftest import create_no_perm_user

        seed = await seed_tenant_and_users(db_session)
        await create_no_perm_user(db_session, seed)
        await login_client(client, "noperm@tenanta.com")
        resp = await client.get("/api/v1/notifications")
        assert resp.status_code == 403