webmail-client/backend/tests/plugins/test_antispam_api.py

"""Antispam HTTP surface + RLS isolation (SPEC §C.6, acceptance 1 and 4).

Exercises the real vertical: entry-point discovery -> router at
/api/plugins/antispam -> SDK auth -> RLS-scoped session -> the plg_antispam_*
tables. Requires the plugin installed and `install antispam` run.
"""

from __future__ import annotations

import uuid
from collections.abc import AsyncIterator

import httpx
import pytest_asyncio
from httpx import ASGITransport
from sqlalchemy import delete, text
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
from sqlalchemy.pool import NullPool

from app.config import get_settings
from app.identity.passwords import hash_password
from app.identity.sessions import create_session
from app.main import create_app
from app.models import User

BASE = "/api/plugins/antispam"


async def _mk_user(sm: async_sessionmaker) -> uuid.UUID:  # type: ignore[type-arg]
    async with sm() as s:
        user = User(
            handle=f"as{uuid.uuid4().hex[:8]}@test.local", password_hash=hash_password("pw")
        )
        s.add(user)
        await s.commit()
        return user.id


class _Ctx:
    user_a: uuid.UUID
    user_b: uuid.UUID
    client_a: httpx.AsyncClient
    client_b: httpx.AsyncClient


@pytest_asyncio.fixture
async def ctx() -> AsyncIterator[_Ctx]:
    engine = create_async_engine(get_settings().database_url, poolclass=NullPool)
    sm = async_sessionmaker(engine, expire_on_commit=False)
    c = _Ctx()
    c.user_a = await _mk_user(sm)
    c.user_b = await _mk_user(sm)
    transport = ASGITransport(app=create_app())

    async def client(uid: uuid.UUID) -> httpx.AsyncClient:
        token = await create_session(uid, "pytest")
        return httpx.AsyncClient(
            transport=transport,
            base_url="http://testserver",
            cookies={"session": token, "csrf": "csrf"},
            headers={"X-CSRF": "csrf"},
        )

    c.client_a = await client(c.user_a)
    c.client_b = await client(c.user_b)
    try:
        yield c
    finally:
        await c.client_a.aclose()
        await c.client_b.aclose()
        async with sm() as s:
            for table in ("plg_antispam_rules", "plg_antispam_settings", "plg_antispam_verdicts"):
                await s.execute(
                    text(f"DELETE FROM {table} WHERE user_id IN (:a, :b)"),
                    {"a": str(c.user_a), "b": str(c.user_b)},
                )
            await s.execute(delete(User).where(User.id.in_([c.user_a, c.user_b])))
            await s.commit()
        await engine.dispose()


async def test_settings_defaults_then_merge(ctx: _Ctx) -> None:
    """A user who has never opened the page still gets the §C.3 defaults."""
    s = (await ctx.client_a.get(f"{BASE}/settings")).json()
    assert (s["move_threshold"], s["banner_threshold"]) == (12.0, 6.0)
    assert s["network_checks"] is False  # privacy default (§C.2)
    assert s["enabled_accounts"] == []  # empty = every account

    # PATCH merges: the field we leave out keeps its value.
    s = (await ctx.client_a.patch(f"{BASE}/settings", json={"move_threshold": 15.0})).json()
    assert s["move_threshold"] == 15.0
    assert s["banner_threshold"] == 6.0


async def test_banner_threshold_cannot_exceed_move(ctx: _Ctx) -> None:
    """A banner above the move score would never be seen — it is capped, not refused."""
    s = (
        await ctx.client_a.patch(
            f"{BASE}/settings", json={"move_threshold": 10.0, "banner_threshold": 20.0}
        )
    ).json()
    assert s["banner_threshold"] == 10.0


async def test_rules_crud_and_validation(ctx: _Ctx) -> None:
    created = await ctx.client_a.post(
        f"{BASE}/rules", json={"kind": "deny", "pattern": "@Bad.TEST"}
    )
    assert created.status_code == 200
    assert created.json()["pattern"] == "@bad.test"  # normalised

    listed = (await ctx.client_a.get(f"{BASE}/rules")).json()
    assert [r["pattern"] for r in listed] == ["@bad.test"]

    # Re-adding the same pattern is idempotent, not a duplicate.
    await ctx.client_a.post(f"{BASE}/rules", json={"kind": "deny", "pattern": "@bad.test"})
    assert len((await ctx.client_a.get(f"{BASE}/rules")).json()) == 1

    assert (
        await ctx.client_a.post(f"{BASE}/rules", json={"kind": "deny", "pattern": "nonsense"})
    ).status_code == 400
    assert (
        await ctx.client_a.post(f"{BASE}/rules", json={"kind": "maybe", "pattern": "a@b.test"})
    ).status_code == 422

    await ctx.client_a.delete(f"{BASE}/rules/{listed[0]['id']}")
    assert (await ctx.client_a.get(f"{BASE}/rules")).json() == []


async def test_rls_isolation_between_users(ctx: _Ctx) -> None:
    """Acceptance 4's per-user isolation, at the storage layer."""
    await ctx.client_a.post(f"{BASE}/rules", json={"kind": "allow", "pattern": "a@only.test"})
    await ctx.client_a.patch(f"{BASE}/settings", json={"move_threshold": 9.0})

    assert (await ctx.client_b.get(f"{BASE}/rules")).json() == []
    assert (await ctx.client_b.get(f"{BASE}/settings")).json()["move_threshold"] == 12.0

    # B cannot delete A's rule even knowing its id.
    a_rule = (await ctx.client_a.get(f"{BASE}/rules")).json()[0]
    await ctx.client_b.delete(f"{BASE}/rules/{a_rule['id']}")
    assert len((await ctx.client_a.get(f"{BASE}/rules")).json()) == 1


async def test_verdict_absent_for_unclassified_message(ctx: _Ctx) -> None:
    assert (await ctx.client_a.get(f"{BASE}/verdict/{uuid.uuid4()}")).json() is None


async def test_stats_work_without_an_engine(ctx: _Ctx) -> None:
    """The stats page must render on a box where Rspamd is not running."""
    body = (await ctx.client_a.get(f"{BASE}/stats")).json()
    assert body["counts"] == {}
    assert body["top_symbols"] == []
    assert body["bayes"] is None  # unreachable engine, not an error


async def test_ui_bundles_are_served_and_gated(ctx: _Ctx) -> None:
    for name in ("settings.js", "banner.js", "actions.js"):
        r = await ctx.client_a.get(f"{BASE}/ui/{name}")
        assert r.status_code == 200
        assert "javascript" in r.headers["content-type"]
    assert (await ctx.client_a.get(f"{BASE}/ui/../hook.py")).status_code == 404