bambuddy/backend/tests/integration/test_notification_app_message.py

"""Messages from other applications through the notification channels.

The contract these tests pin:

  ``POST /notifications/app-message`` hands an app's message to every enabled
  channel with ``on_app_message`` on, and to no other. An API key needs the
  opt-in ``can_send_notifications`` scope, and its owner the
  ``notifications:update`` permission (a key can't do what its owner may not).
  Users need ``notifications:update``. Text is plain, the link http(s) only,
  and each caller is rate-limited.
"""

from unittest.mock import AsyncMock, patch

import pytest
from httpx import AsyncClient
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from backend.app.api.routes import notifications as notification_routes
from backend.app.core.auth import generate_api_key
from backend.app.models.api_key import APIKey
from backend.app.models.notification import NotificationProvider
from backend.app.models.user import User

URL = "/api/v1/notifications/app-message"


@pytest.fixture(autouse=True)
def _fresh_rate_limit():
    notification_routes._app_message_times.clear()
    yield
    notification_routes._app_message_times.clear()


async def _admin_token(client: AsyncClient) -> str:
    await client.post(
        "/api/v1/auth/setup",
        json={
            "auth_enabled": True,
            "admin_username": "notifyadmin",
            "admin_password": "AdminPass1!",  # pragma: allowlist secret
        },
    )
    login = await client.post(
        "/api/v1/auth/login",
        json={"username": "notifyadmin", "password": "AdminPass1!"},  # pragma: allowlist secret
    )
    return login.json()["access_token"]


async def _user_id(db: AsyncSession, username: str) -> int:
    return (await db.execute(select(User).where(User.username == username))).scalar_one().id


async def _key(db: AsyncSession, *, owner_id: int | None, allowed: bool, name: str = "Bambuddy Orders") -> str:
    full_key, key_hash, key_prefix = generate_api_key()
    db.add(
        APIKey(name=name, key_hash=key_hash, key_prefix=key_prefix, user_id=owner_id, can_send_notifications=allowed)
    )
    await db.commit()
    return full_key


async def _channels(db: AsyncSession) -> None:
    for name, on, enabled in (
        ("Telegram", True, True),
        ("ntfy", True, True),
        ("Email", False, True),
        ("Old", True, False),
    ):
        db.add(NotificationProvider(name=name, provider_type="ntfy", enabled=enabled, config="{}", on_app_message=on))
    await db.commit()


def _sent():
    return patch(
        "backend.app.services.notification_service.notification_service._send_to_provider",
        new=AsyncMock(return_value=(True, "")),
    )


class TestAppMessage:
    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_goes_to_the_channels_that_accept_app_messages(self, async_client: AsyncClient, db_session):
        await _admin_token(async_client)
        await _channels(db_session)
        key = await _key(db_session, owner_id=await _user_id(db_session, "notifyadmin"), allowed=True)

        with _sent() as send:
            resp = await async_client.post(
                URL,
                headers={"X-API-Key": key},
                json={
                    "title": "3 orders need you",
                    "message": "#1004, #1009, #1010",
                    "url": "http://orders.lan:8090/todo",
                },
            )

        assert resp.status_code == 200, resp.text
        assert resp.json() == {"channels": 2}
        names = sorted(call.args[0].name for call in send.await_args_list)
        assert names == ["Telegram", "ntfy"]
        _, title, body = send.await_args_list[0].args[:3]
        assert title == "3 orders need you"
        assert body == "#1004, #1009, #1010\nhttp://orders.lan:8090/todo"
        assert send.await_args_list[0].kwargs["event_type"] == "app:Bambuddy Orders"

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_lists_the_channels_by_name(self, async_client: AsyncClient, db_session):
        await _admin_token(async_client)
        await _channels(db_session)
        key = await _key(db_session, owner_id=await _user_id(db_session, "notifyadmin"), allowed=True)
        resp = await async_client.get(f"{URL}/channels", headers={"X-API-Key": key})
        assert resp.status_code == 200
        assert resp.json() == [{"name": "Telegram", "provider_type": "ntfy"}, {"name": "ntfy", "provider_type": "ntfy"}]

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_a_key_without_the_scope_is_refused(self, async_client: AsyncClient, db_session):
        await _admin_token(async_client)
        key = await _key(db_session, owner_id=await _user_id(db_session, "notifyadmin"), allowed=False)
        resp = await async_client.post(URL, headers={"X-API-Key": key}, json={"title": "t", "message": "m"})
        assert resp.status_code == 403
        assert "send_notifications" in resp.json()["detail"]

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_a_key_cant_do_what_its_owner_may_not(self, async_client: AsyncClient, db_session):
        await _admin_token(async_client)
        db_session.add(User(username="nobody", password_hash="x", role="user", is_active=True))
        await db_session.commit()
        key = await _key(db_session, owner_id=await _user_id(db_session, "nobody"), allowed=True)
        resp = await async_client.post(URL, headers={"X-API-Key": key}, json={"title": "t", "message": "m"})
        assert resp.status_code == 403
        assert "notifications:update" in resp.json()["detail"]

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_an_admin_user_may_send(self, async_client: AsyncClient, db_session):
        token = await _admin_token(async_client)
        with _sent():
            resp = await async_client.post(
                URL, headers={"Authorization": f"Bearer {token}"}, json={"title": "Test", "message": "Hello"}
            )
        assert resp.status_code == 200
        assert resp.json() == {"channels": 0}

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_unauthenticated_is_refused(self, async_client: AsyncClient):
        await _admin_token(async_client)
        resp = await async_client.post(URL, json={"title": "t", "message": "m"})
        assert resp.status_code == 401

    @pytest.mark.asyncio
    @pytest.mark.integration
    @pytest.mark.parametrize(
        "payload",
        [
            {"title": "", "message": "m"},
            {"title": "t", "message": "   "},
            {"title": "t" * 121, "message": "m"},
            {"title": "t", "message": "m", "url": "javascript:alert(1)"},
            {"title": "t", "message": "m", "url": "http://a b"},
        ],
    )
    async def test_refuses_what_isnt_plain_text_or_a_web_link(self, async_client: AsyncClient, db_session, payload):
        await _admin_token(async_client)
        key = await _key(db_session, owner_id=await _user_id(db_session, "notifyadmin"), allowed=True)
        resp = await async_client.post(URL, headers={"X-API-Key": key}, json=payload)
        assert resp.status_code == 422

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_control_characters_are_dropped(self, async_client: AsyncClient, db_session):
        await _admin_token(async_client)
        await _channels(db_session)
        key = await _key(db_session, owner_id=await _user_id(db_session, "notifyadmin"), allowed=True)
        with _sent() as send:
            await async_client.post(URL, headers={"X-API-Key": key}, json={"title": "Hi\x07", "message": "a\nb\x1b"})
        assert send.await_args_list[0].args[1:3] == ("Hi", "a\nb")

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_each_caller_is_rate_limited(self, async_client: AsyncClient, db_session):
        await _admin_token(async_client)
        owner = await _user_id(db_session, "notifyadmin")
        key = await _key(db_session, owner_id=owner, allowed=True)
        other = await _key(db_session, owner_id=owner, allowed=True, name="Other app")
        with _sent():
            codes = [
                (
                    await async_client.post(URL, headers={"X-API-Key": key}, json={"title": "t", "message": "m"})
                ).status_code
                for _ in range(notification_routes.APP_MESSAGE_LIMIT + 1)
            ]
            other_code = (
                await async_client.post(URL, headers={"X-API-Key": other}, json={"title": "t", "message": "m"})
            ).status_code
        assert codes[:-1] == [200] * notification_routes.APP_MESSAGE_LIMIT
        assert codes[-1] == 429
        assert other_code == 200


class TestProviderAndKeyFlags:
    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_the_channel_switch_round_trips(self, async_client: AsyncClient):
        resp = await async_client.post(
            "/api/v1/notifications/",
            json={"name": "Phone", "provider_type": "ntfy", "config": {"server": "https://ntfy.sh", "topic": "x"}},
        )
        assert resp.status_code == 200, resp.text
        provider = resp.json()
        assert provider["on_app_message"] is False
        resp = await async_client.patch(f"/api/v1/notifications/{provider['id']}", json={"on_app_message": True})
        assert resp.json()["on_app_message"] is True

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_the_key_scope_round_trips_and_defaults_off(self, async_client: AsyncClient):
        token = await _admin_token(async_client)
        headers = {"Authorization": f"Bearer {token}"}
        resp = await async_client.post("/api/v1/api-keys/", headers=headers, json={"name": "orders"})
        assert resp.json()["can_send_notifications"] is False
        resp = await async_client.patch(
            f"/api/v1/api-keys/{resp.json()['id']}", headers=headers, json={"can_send_notifications": True}
        )
        assert resp.json()["can_send_notifications"] is True