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

"""End-to-end test of the hello reference plugin (SPEC §11, §12 M3 acceptance).

Exercises the whole vertical against the real HTTP surface: entry-point discovery
-> router mounted at /api/plugins/hello -> SDK auth (CurrentUserId) -> RLS-scoped
session -> the plg_hello_note table. Also proves per-user isolation: one user never
sees another's notes. Requires the plugin installed + `install hello` 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


async def _mk_user(sm: async_sessionmaker) -> uuid.UUID:  # type: ignore[type-arg]
    async with sm() as s:
        user = User(
            handle=f"hp{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:
            await s.execute(
                delete(User).where(User.id.in_([c.user_a, c.user_b]))
            )
            await s.execute(
                text("DELETE FROM plg_hello_note WHERE user_id IN (:a, :b)"),
                {"a": str(c.user_a), "b": str(c.user_b)},
            )
            await s.commit()
        await engine.dispose()


async def test_notes_crud_and_rls_isolation(ctx: _Ctx) -> None:
    # A adds a note and reads it back.
    r = await ctx.client_a.post("/api/plugins/hello/notes", json={"body": "a-note"})
    assert r.status_code == 200
    a_notes = (await ctx.client_a.get("/api/plugins/hello/notes")).json()
    assert [n["body"] for n in a_notes] == ["a-note"]

    # B sees none of A's notes (RLS), then adds and sees only its own.
    assert (await ctx.client_b.get("/api/plugins/hello/notes")).json() == []
    await ctx.client_b.post("/api/plugins/hello/notes", json={"body": "b-note"})
    b_notes = (await ctx.client_b.get("/api/plugins/hello/notes")).json()
    assert [n["body"] for n in b_notes] == ["b-note"]

    # A still sees only its own note.
    a_again = (await ctx.client_a.get("/api/plugins/hello/notes")).json()
    assert [n["body"] for n in a_again] == ["a-note"]


async def test_notes_requires_auth(ctx: _Ctx) -> None:
    transport = ASGITransport(app=create_app())
    async with httpx.AsyncClient(transport=transport, base_url="http://testserver") as anon:
        assert (await anon.get("/api/plugins/hello/notes")).status_code == 401