bambuddy/backend/tests/integration/test_notification_scope.py

"""Notification providers and logs keep to the caller's printers."""

from __future__ import annotations

from datetime import datetime, timedelta

import pytest
from httpx import AsyncClient
from sqlalchemy import select

from backend.app.models.notification import NotificationLog
from backend.tests.integration.test_printer_scope_1727 import _admin_token, _auth, _group, _user

pytestmark = [pytest.mark.asyncio, pytest.mark.integration]

PERMS = [
    "printers:read",
    "notifications:read",
    "notifications:create",
    "notifications:update",
    "notifications:delete",
]


def _provider(name: str, printer_id: int | None) -> dict:
    return {
        "name": name,
        "provider_type": "callmebot",
        "enabled": True,
        "config": {"phone_number": "+1234567890", "api_key": "test-api-key"},
        "printer_id": printer_id,
    }


async def _setup(client: AsyncClient, printer_factory):
    a, b = await printer_factory(name="A"), await printer_factory(name="B")
    admin = await _admin_token(client)
    perms = await _group(client, admin, "perms", permissions=PERMS)
    team = await _group(client, admin, "team", printer_ids=[a.id])
    member = _auth((await _user(client, admin, "member", [perms, team]))[0])
    ids = {}
    for name, printer in (("ours", a.id), ("theirs", b.id), ("everyone", None)):
        created = await client.post("/api/v1/notifications/", headers=_auth(admin), json=_provider(name, printer))
        assert created.status_code == 200, created.text
        ids[name] = created.json()["id"]
    return a, b, member, ids


async def test_a_member_sees_and_manages_only_providers_of_their_printers(async_client: AsyncClient, printer_factory):
    a, b, member, ids = await _setup(async_client, printer_factory)

    listed = await async_client.get("/api/v1/notifications/", headers=member)

    assert [p["name"] for p in listed.json()] == ["ours"]
    for name in ("theirs", "everyone"):
        pid = ids[name]
        assert (await async_client.get(f"/api/v1/notifications/{pid}", headers=member)).status_code == 404
        patched = await async_client.patch(f"/api/v1/notifications/{pid}", headers=member, json={"name": "mine now"})
        assert patched.status_code == 404
        assert (await async_client.post(f"/api/v1/notifications/{pid}/test", headers=member)).status_code == 404
        assert (await async_client.delete(f"/api/v1/notifications/{pid}", headers=member)).status_code == 404
    assert (await async_client.get(f"/api/v1/notifications/{ids['ours']}", headers=member)).status_code == 200


async def test_a_member_binds_providers_to_their_printers_only(async_client: AsyncClient, printer_factory):
    a, b, member, ids = await _setup(async_client, printer_factory)

    to_other = await async_client.post("/api/v1/notifications/", headers=member, json=_provider("x", b.id))
    to_all = await async_client.post("/api/v1/notifications/", headers=member, json=_provider("y", None))
    to_own = await async_client.post("/api/v1/notifications/", headers=member, json=_provider("z", a.id))
    retarget_other = await async_client.patch(
        f"/api/v1/notifications/{ids['ours']}", headers=member, json={"printer_id": b.id}
    )
    retarget_all = await async_client.patch(
        f"/api/v1/notifications/{ids['ours']}", headers=member, json={"printer_id": None}
    )

    assert (to_other.status_code, to_all.status_code, to_own.status_code) == (404, 400, 200)
    assert (retarget_other.status_code, retarget_all.status_code) == (404, 400)


async def test_logs_and_their_counts_keep_to_the_team_printers(async_client: AsyncClient, db_session, printer_factory):
    a, b, member, ids = await _setup(async_client, printer_factory)
    a_id, b_id = a.id, b.id
    old = datetime.utcnow() - timedelta(days=40)
    for printer, when in ((a_id, None), (b_id, None), (None, None), (a_id, old), (b_id, old), (None, old)):
        row = NotificationLog(
            provider_id=ids["ours"], event_type="print_start", title="t", message="m", printer_id=printer
        )
        if when is not None:
            row.created_at = when
        db_session.add(row)
    await db_session.commit()

    listed = await async_client.get("/api/v1/notifications/logs", headers=member)
    stats = await async_client.get("/api/v1/notifications/logs/stats", headers=member)
    cleared = await async_client.delete("/api/v1/notifications/logs", headers=member, params={"older_than_days": 30})

    assert sorted((row["printer_id"] or 0) for row in listed.json()) == [0, a_id]
    assert stats.json()["total"] == 2
    assert cleared.json()["deleted"] == 1
    db_session.expire_all()
    left = sorted((pid or 0) for (pid,) in (await db_session.execute(select(NotificationLog.printer_id))).all())
    assert left == sorted([a_id, b_id, 0, b_id, 0])


async def test_testing_a_stored_provider_answers_a_hidden_one_like_a_missing_one(
    async_client: AsyncClient, printer_factory
):
    a, b, member, ids = await _setup(async_client, printer_factory)
    body = {"provider_type": "callmebot", "config": {"phone_number": "+1234567890", "api_key": "********"}}

    hidden = await async_client.post(
        "/api/v1/notifications/test-config", headers=member, json={**body, "provider_id": ids["theirs"]}
    )
    missing = await async_client.post(
        "/api/v1/notifications/test-config", headers=member, json={**body, "provider_id": 999_999}
    )

    assert (
        (hidden.status_code, hidden.json())
        == (missing.status_code, missing.json())
        == (
            404,
            {"detail": "Notification provider not found"},
        )
    )


async def test_app_messages_reach_only_the_team_printers_channels(async_client: AsyncClient, printer_factory):
    from unittest.mock import AsyncMock, patch

    a, b = await printer_factory(name="A"), await printer_factory(name="B")
    admin = await _admin_token(async_client)
    perms = await _group(async_client, admin, "perms", permissions=PERMS)
    team = await _group(async_client, admin, "team", printer_ids=[a.id])
    member = _auth((await _user(async_client, admin, "member", [perms, team]))[0])
    for name, printer in (("ours", a.id), ("theirs", b.id), ("everyone", None)):
        created = await async_client.post(
            "/api/v1/notifications/", headers=_auth(admin), json={**_provider(name, printer), "on_app_message": True}
        )
        assert created.status_code == 200, created.text

    channels = await async_client.get("/api/v1/notifications/app-message/channels", headers=member)
    with patch(
        "backend.app.services.notification_service.NotificationService._send_to_providers", new=AsyncMock()
    ) as send:
        sent = await async_client.post(
            "/api/v1/notifications/app-message", headers=member, json={"title": "t", "message": "m"}
        )

    assert [c["name"] for c in channels.json()] == ["ours"]
    assert sent.json()["channels"] == 1
    assert [p.name for p in send.await_args.args[0]] == ["ours"]


async def test_the_log_does_not_name_providers_the_caller_cannot_see(
    async_client: AsyncClient, db_session, printer_factory
):
    a, b, member, ids = await _setup(async_client, printer_factory)
    a_id = a.id
    for provider in ("ours", "everyone"):
        db_session.add(
            NotificationLog(
                provider_id=ids[provider],
                event_type="print_start",
                title="t",
                message="m",
                printer_id=a_id,
                success=False,
                error_message=f"failed for {provider}",
            )
        )
    await db_session.commit()

    listed = await async_client.get("/api/v1/notifications/logs", headers=member)
    stats = await async_client.get("/api/v1/notifications/logs/stats", headers=member)

    rows = sorted(listed.json(), key=lambda row: str(row["provider_name"]))
    assert [(row["provider_name"], row["error_message"]) for row in rows] == [
        (None, None),
        ("ours", "failed for ours"),
    ]
    assert stats.json()["by_provider"] == {"ours": 1, "Other": 1}


async def test_log_entries_without_a_printer_stay_with_their_provider(
    async_client: AsyncClient, db_session, printer_factory
):
    a, b, member, ids = await _setup(async_client, printer_factory)
    for provider in ("ours", "everyone"):
        db_session.add(
            NotificationLog(
                provider_id=ids[provider], event_type="app:key", title=provider, message="m", printer_id=None
            )
        )
    await db_session.commit()

    listed = await async_client.get("/api/v1/notifications/logs", headers=member)
    stats = await async_client.get("/api/v1/notifications/logs/stats", headers=member)

    assert [row["title"] for row in listed.json()] == ["ours"]
    assert stats.json()["total"] == 1


async def test_a_log_entry_whose_provider_is_gone_shows_no_error(
    async_client: AsyncClient, db_session, printer_factory
):
    if db_session.bind.dialect.name != "sqlite":
        pytest.skip("only SQLite can keep such an entry")
    a, b, member, ids = await _setup(async_client, printer_factory)
    db_session.add(
        NotificationLog(
            provider_id=999_999,
            event_type="print_start",
            title="t",
            message="m",
            printer_id=a.id,
            success=False,
            error_message="failed",
        )
    )
    await db_session.commit()

    listed = await async_client.get("/api/v1/notifications/logs", headers=member)

    assert [(row["provider_name"], row["error_message"]) for row in listed.json()] == [(None, None)]