bambuddy/backend/tests/integration/test_announcements_api.py

"""Who sees announcements, and per-user read state.

Administrators see them; other signed-in users only when
``announcements_all_users`` is on. With authentication off, the person running
Bambuddy sees them. Anyone else gets an empty list, which hides the sidebar entry
and the banner the same way "nothing published" does.
"""

import json
import secrets

import pytest
from httpx import AsyncClient

from backend.app.models.announcement import Announcement
from backend.app.models.settings import Settings

URL = "/api/v1/announcements"
_FIXTURE_PW = "Aa1!" + secrets.token_urlsafe(12)  # pragma: allowlist secret


async def _seed(db, *public_ids: str, level: str = "important") -> None:
    for pid in public_ids:
        db.add(
            Announcement(
                public_id=pid,
                level=level,
                texts=json.dumps({"en": {"title": f"T {pid}", "body": "B"}, "de": {"title": "D", "body": "B"}}),
            )
        )
    await db.commit()


async def _setting(db, key: str, value: str) -> None:
    db.add(Settings(key=key, value=value))
    await db.commit()


async def _admin(client: AsyncClient, username: str) -> dict:
    await client.post(
        "/api/v1/auth/setup",
        json={"auth_enabled": True, "admin_username": username, "admin_password": _FIXTURE_PW},
    )
    login = await client.post("/api/v1/auth/login", json={"username": username, "password": _FIXTURE_PW})
    assert login.status_code == 200, login.text
    return {"Authorization": f"Bearer {login.json()['access_token']}"}


async def _user(client: AsyncClient, admin: dict, username: str) -> dict:
    grp = await client.post(
        "/api/v1/groups/", headers=admin, json={"name": f"ann_{username}", "permissions": ["printers:read"]}
    )
    assert grp.status_code == 201, grp.text
    created = await client.post(
        "/api/v1/users/",
        headers=admin,
        json={"username": username, "password": _FIXTURE_PW, "role": "user", "group_ids": [grp.json()["id"]]},
    )
    assert created.status_code == 201, created.text
    login = await client.post("/api/v1/auth/login", json={"username": username, "password": _FIXTURE_PW})
    return {"Authorization": f"Bearer {login.json()['access_token']}"}


@pytest.mark.integration
class TestAuthOff:
    @pytest.mark.asyncio
    async def test_listed_with_every_language_and_unread(self, async_client: AsyncClient, db_session):
        await _seed(db_session, "a1")
        r = await async_client.get(URL)
        assert r.status_code == 200
        [item] = r.json()
        assert item["id"] == "a1" and item["level"] == "important" and item["read"] is False
        assert set(item["texts"]) == {"en", "de"}

    @pytest.mark.asyncio
    async def test_read_sticks(self, async_client: AsyncClient, db_session):
        await _seed(db_session, "a1")
        assert (await async_client.post(f"{URL}/a1/read")).status_code == 204
        assert (await async_client.get(URL)).json()[0]["read"] is True

    @pytest.mark.asyncio
    async def test_unknown_id_is_404(self, async_client: AsyncClient):
        assert (await async_client.post(f"{URL}/nope/read")).status_code == 404

    @pytest.mark.asyncio
    async def test_switched_off_shows_nothing(self, async_client: AsyncClient, db_session):
        await _seed(db_session, "a1")
        await _setting(db_session, "announcements_enabled", "false")
        assert (await async_client.get(URL)).json() == []
        assert (await async_client.post(f"{URL}/a1/read")).status_code == 404


@pytest.mark.integration
class TestAuthOn:
    @pytest.mark.asyncio
    async def test_signed_out_is_401(self, async_client: AsyncClient, db_session):
        await _admin(async_client, "annadmin0")
        assert (await async_client.get(URL)).status_code == 401

    @pytest.mark.asyncio
    async def test_admin_sees_them_a_user_does_not(self, async_client: AsyncClient, db_session):
        await _seed(db_session, "a1")
        admin = await _admin(async_client, "annadmin1")
        user = await _user(async_client, admin, "annuser1")
        assert [i["id"] for i in (await async_client.get(URL, headers=admin)).json()] == ["a1"]
        assert (await async_client.get(URL, headers=user)).json() == []
        assert (await async_client.post(f"{URL}/a1/read", headers=user)).status_code == 404

    @pytest.mark.asyncio
    async def test_all_users_setting_shows_them_with_separate_read_state(self, async_client: AsyncClient, db_session):
        await _seed(db_session, "a1")
        admin = await _admin(async_client, "annadmin2")
        user = await _user(async_client, admin, "annuser2")
        await _setting(db_session, "announcements_all_users", "true")

        assert (await async_client.post(f"{URL}/a1/read", headers=user)).status_code == 204
        assert (await async_client.get(URL, headers=user)).json()[0]["read"] is True
        assert (await async_client.get(URL, headers=admin)).json()[0]["read"] is False


@pytest.mark.integration
class TestSettings:
    @pytest.mark.asyncio
    async def test_both_switches_round_trip(self, async_client: AsyncClient):
        defaults = (await async_client.get("/api/v1/settings/")).json()
        assert defaults["announcements_enabled"] is True
        assert defaults["announcements_all_users"] is False
        r = await async_client.put(
            "/api/v1/settings/", json={"announcements_enabled": False, "announcements_all_users": True}
        )
        assert r.status_code == 200, r.text
        after = (await async_client.get("/api/v1/settings/")).json()
        assert after["announcements_enabled"] is False
        assert after["announcements_all_users"] is True