webmail-client/backend/tests/identity/test_registration_policy.py

"""Registration policy: first user bootstraps as administrator, then signup closes.

The bootstrap cases run against the rollback-per-test ``db_session`` fixture, so
they can empty the users table to simulate a fresh install without touching the
dev database.
"""

from __future__ import annotations

import uuid
from collections.abc import AsyncIterator

import httpx
import pytest
import pytest_asyncio
from httpx import ASGITransport
from sqlalchemy import delete, select, update
from sqlalchemy.ext.asyncio import AsyncSession, 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.service import (
    RegistrationError,
    app_settings,
    register_user,
    registration_status,
)
from app.identity.sessions import create_session
from app.main import create_app
from app.models import AppSettings, User

CSRF = "csrf-token"
PASSWORD = "hunter2pass"


def _localpart() -> str:
    return f"reg{uuid.uuid4().hex[:8]}"


# --- service level: the bootstrap decision ----------------------------------


async def test_first_user_becomes_admin_and_closes_registration(db_session: AsyncSession) -> None:
    """On an empty instance the first signup is allowed, is admin, and shuts the door."""
    await db_session.execute(delete(User))
    settings = await app_settings(db_session)
    settings.registration_open = False  # even closed, an empty instance must bootstrap
    await db_session.flush()

    status = await registration_status(db_session)
    assert (status.open, status.bootstrap) == (True, True)

    first = await register_user(
        db_session, localpart=_localpart(), password=PASSWORD, recovery_email=None
    )
    assert first.is_admin is True

    settings = await app_settings(db_session)
    assert settings.registration_open is False
    status = await registration_status(db_session)
    assert (status.open, status.bootstrap) == (False, False)

    with pytest.raises(RegistrationError) as exc:
        await register_user(
            db_session, localpart=_localpart(), password=PASSWORD, recovery_email=None
        )
    assert exc.value.code == "registration_closed"


async def test_reopened_registration_admits_plain_users(db_session: AsyncSession) -> None:
    """A second account created while open is an ordinary user, not an admin."""
    await db_session.execute(delete(User))
    await register_user(db_session, localpart=_localpart(), password=PASSWORD, recovery_email=None)

    settings = await app_settings(db_session)
    settings.registration_open = True
    await db_session.flush()

    second = await register_user(
        db_session, localpart=_localpart(), password=PASSWORD, recovery_email=None
    )
    assert second.is_admin is False
    # Reopening is a deliberate admin act; it stays open until they close it.
    assert (await app_settings(db_session)).registration_open is True


async def test_missing_settings_row_defaults_to_closed(db_session: AsyncSession) -> None:
    """A database predating the settings row must not fall open."""
    await db_session.execute(delete(AppSettings))
    assert (await app_settings(db_session)).registration_open is False


# --- API level ---------------------------------------------------------------


@pytest_asyncio.fixture
async def api() -> AsyncIterator[tuple[httpx.AsyncClient, AsyncSession, uuid.UUID]]:
    """A logged-in non-admin user plus a restored-on-exit app_settings row."""
    engine = create_async_engine(get_settings().database_url, poolclass=NullPool)
    sm = async_sessionmaker(engine, expire_on_commit=False)
    session = sm()
    before = await app_settings(session)
    was_open, was_quota = before.registration_open, before.default_quota_bytes
    user = User(handle=f"{_localpart()}@test.local", password_hash=hash_password(PASSWORD))
    session.add(user)
    await session.commit()
    token = await create_session(user.id, "pytest")
    client = httpx.AsyncClient(
        transport=ASGITransport(app=create_app()),
        base_url="http://testserver",
        cookies={"session": token, "csrf": CSRF},
    )
    try:
        yield client, session, user.id
    finally:
        await client.aclose()
        await session.execute(delete(User).where(User.id == user.id))
        await session.execute(
            update(AppSettings).values(registration_open=was_open, default_quota_bytes=was_quota)
        )
        await session.commit()
        await session.close()
        await engine.dispose()


async def _set_open(session: AsyncSession, value: bool) -> None:
    await session.execute(update(AppSettings).values(registration_open=value))
    await session.commit()


async def test_register_is_refused_while_closed(
    api: tuple[httpx.AsyncClient, AsyncSession, uuid.UUID],
) -> None:
    client, session, _ = api
    await _set_open(session, False)

    status = await client.get("/api/auth/registration")
    assert status.status_code == 200
    assert status.json() == {
        "open": False,
        "bootstrap": False,
        "domain": get_settings().app_domain,
    }

    refused = await client.post(
        "/api/auth/register", json={"localpart": _localpart(), "password": PASSWORD}
    )
    assert refused.status_code == 403
    assert refused.json()["error"]["code"] == "registration_closed"


async def test_admin_settings_endpoint_requires_admin(
    api: tuple[httpx.AsyncClient, AsyncSession, uuid.UUID],
) -> None:
    client, session, user_id = api
    denied = await client.get("/api/admin/settings")
    assert denied.status_code == 403
    assert denied.json()["error"]["code"] == "not_admin"

    await session.execute(update(User).where(User.id == user_id).values(is_admin=True))
    await session.commit()

    assert (await client.get("/api/admin/settings")).status_code == 200
    opened = await client.put(
        "/api/admin/settings", json={"registration_open": True}, headers={"X-CSRF": CSRF}
    )
    assert opened.status_code == 200
    assert opened.json()["registration_open"] is True
    assert (await client.get("/api/auth/registration")).json()["open"] is True


async def test_me_exposes_admin_flag(
    api: tuple[httpx.AsyncClient, AsyncSession, uuid.UUID],
) -> None:
    client, session, user_id = api
    assert (await client.get("/api/auth/me")).json()["is_admin"] is False

    await session.execute(update(User).where(User.id == user_id).values(is_admin=True))
    await session.commit()
    assert (await client.get("/api/auth/me")).json()["is_admin"] is True


async def test_settings_row_is_a_singleton(db_session: AsyncSession) -> None:
    """The CHECK constraint is what stops a second row drifting into existence."""
    rows = (await db_session.execute(select(AppSettings.id))).scalars().all()
    assert rows == [1]


# --- default quota -----------------------------------------------------------

_GIB = 1024**3


async def test_default_quota_applies_only_to_later_users(db_session: AsyncSession) -> None:
    """Raising the default must not resize a mailbox that already exists."""
    await db_session.execute(delete(User))
    settings = await app_settings(db_session)
    settings.default_quota_bytes = 5 * _GIB
    await db_session.flush()

    first = await register_user(
        db_session, localpart=_localpart(), password=PASSWORD, recovery_email=None
    )
    assert first.quota_bytes == 5 * _GIB

    settings = await app_settings(db_session)
    settings.default_quota_bytes = 20 * _GIB
    settings.registration_open = True
    await db_session.flush()

    second = await register_user(
        db_session, localpart=_localpart(), password=PASSWORD, recovery_email=None
    )
    assert second.quota_bytes == 20 * _GIB
    # The whole point: the earlier account is untouched.
    await db_session.refresh(first)
    assert first.quota_bytes == 5 * _GIB


async def test_admin_can_set_default_quota(
    api: tuple[httpx.AsyncClient, AsyncSession, uuid.UUID],
) -> None:
    client, session, user_id = api
    await session.execute(update(User).where(User.id == user_id).values(is_admin=True))
    await session.commit()
    # Establish the value the merge must preserve rather than assuming the
    # ambient one — the dev database is shared and its state moves around.
    await _set_open(session, True)

    saved = await client.put(
        "/api/admin/settings",
        json={"default_quota_bytes": 10 * _GIB},
        headers={"X-CSRF": CSRF},
    )
    assert saved.status_code == 200
    assert saved.json()["default_quota_bytes"] == 10 * _GIB
    # A merge, not a replace: the field we left out kept its value.
    assert saved.json()["registration_open"] is True

    rejected = await client.put(
        "/api/admin/settings", json={"default_quota_bytes": 0}, headers={"X-CSRF": CSRF}
    )
    assert rejected.status_code == 422