paskia/tests/conftest.py

"""
Pytest configuration and fixtures for Paskia API tests.

FastAPI provides excellent testing support through httpx.ASGITransport,
which allows us to make async requests directly to the ASGI app without
running a server.

Since we can't emulate WebAuthn passkeys, we create sessions directly
in the database to test authenticated endpoints.
"""

from __future__ import annotations

import asyncio
import os
import secrets
import tempfile
from collections.abc import AsyncGenerator
from datetime import UTC, datetime, timedelta
from pathlib import Path
from uuid import UUID

import httpx
import pytest
import pytest_asyncio
from kanta import Kanta

import paskia.db.operations as ops_db
from paskia import realms
from paskia.authsession import reset_expires
from paskia.config import SESSION_LIFETIME
from paskia.db import (
    Credential,
    Org,
    Permission,
    Role,
    User,
    create_credential,
    create_reset_token,
    create_role,
    create_user,
)
from paskia.db.bootstrap import bootstrap
from paskia.db.operations import DB
from paskia.db.structs import Config, RealmConfig, Session
from paskia.fastapi.mainapp import app
from paskia.fastapi.session import AUTH_COOKIE_NAME
from paskia.util import avatar
from paskia.util.crypto import hash_secret

TEST_RP_ID = "localhost"
TEST_LISTEN = ["localhost:4401"]


@pytest.fixture(scope="session")
def event_loop():
    """Create an event loop for the test session."""
    loop = asyncio.get_event_loop_policy().new_event_loop()
    yield loop
    loop.close()


@pytest.fixture(autouse=True)
def _avatar_tmp_root(tmp_path, monkeypatch):
    """Redirect avatar storage to a per-test temporary directory."""
    root = tmp_path / "users"

    def users_root(create_root: bool = False) -> Path:
        if create_root:
            root.mkdir(parents=True, exist_ok=True)
        return root

    monkeypatch.setattr(avatar, "users_root_path", users_root)


@pytest_asyncio.fixture(scope="function")
async def test_db() -> AsyncGenerator[DB]:
    """Create a temporary JSONL database for testing using kanta.

    Uses a kanta bootstrap callback to properly initialize the database with:
    - auth:admin and auth:org:admin permissions
    - A default organization with Administration role
    - An admin user with the Administration role
    - The localhost realm configuration (with its OIDC provider)
    """
    with tempfile.NamedTemporaryFile(suffix=".jsonl", delete=True) as f:
        db = DB()
        kanta = Kanta(f.name, db)

        # Register bootstrap callback so kanta seeds the empty DB during open()
        @kanta.bootstrap(action="bootstrap")
        def bootstrap_test_db(data: DB) -> None:
            bootstrap(
                data,
                org_name="Test Organization",
                admin_name="Test Admin",
                config=Config(realms=[RealmConfig(rp_id=TEST_RP_ID)]),
            )

        await kanta.open()
        ops_db._db = db
        ops_db._db._store = kanta
        yield ops_db._db
        await kanta.close()
        ops_db._db = None


@pytest_asyncio.fixture(scope="function")
async def realm_registry(test_db: DB) -> realms.RealmRegistry:
    """Install the realm registry built from the test database config."""
    realms.configure(listen=TEST_LISTEN)
    return realms.init_registry(test_db.config)


@pytest_asyncio.fixture(scope="function")
async def admin_permission(test_db: DB) -> Permission:
    """Get the auth:admin permission created by bootstrap."""
    return next(p for p in test_db.permissions.values() if p.scope == "auth:admin")


@pytest_asyncio.fixture(scope="function")
async def org_admin_permission(test_db: DB) -> Permission:
    """Get the auth:org:admin permission created by bootstrap."""
    return next(p for p in test_db.permissions.values() if p.scope == "auth:org:admin")


@pytest_asyncio.fixture(scope="function")
async def test_org(test_db: DB) -> Org:
    """Get the test organization created by bootstrap."""
    # Bootstrap creates exactly one org
    return next(iter(test_db.orgs.values()))


@pytest_asyncio.fixture(scope="function")
async def test_role(test_db: DB) -> Role:
    """Get the Administration role created by bootstrap."""
    # Bootstrap creates exactly one role (Administration)
    return next(iter(test_db.roles.values()))


@pytest_asyncio.fixture(scope="function")
async def user_role(test_db: DB, test_org: Org) -> Role:
    """Create a test role without admin permission (regular user)."""
    role = Role.create(
        org=test_org.uuid,
        display_name="User Role",
    )
    create_role(role)
    return role


@pytest_asyncio.fixture(scope="function")
async def test_user(test_db: DB) -> User:
    """Get the admin user created by bootstrap."""
    # Bootstrap creates exactly one user (admin)
    return next(iter(test_db.users.values()))


@pytest_asyncio.fixture(scope="function")
async def regular_user(test_db: DB, user_role: Role) -> User:
    """Create a regular test user without admin permissions."""
    user = User.create(
        display_name="Regular User",
        role=user_role.uuid,
    )
    create_user(user)
    return user


@pytest_asyncio.fixture(scope="function")
async def test_credential(test_db: DB, test_user: User) -> Credential:
    """Create a test credential for the admin user."""
    credential = Credential.create(
        credential_id=os.urandom(32),
        user=test_user.uuid,
        aaguid=UUID("00000000-0000-0000-0000-000000000000"),
        public_key=os.urandom(64),
        sign_count=0,
        rp_id=TEST_RP_ID,
    )
    create_credential(credential)
    return credential


@pytest_asyncio.fixture(scope="function")
async def regular_credential(test_db: DB, regular_user: User) -> Credential:
    """Create a test credential for the regular user."""
    credential = Credential.create(
        credential_id=os.urandom(32),
        user=regular_user.uuid,
        aaguid=UUID("00000000-0000-0000-0000-000000000000"),
        public_key=os.urandom(64),
        sign_count=0,
        rp_id=TEST_RP_ID,
    )
    create_credential(credential)
    return credential


@pytest_asyncio.fixture(scope="function")
async def session_token(
    test_db: DB, test_user: User, test_credential: Credential
) -> str:
    """Create a session for the admin user and return the token."""
    _db_key, secret = create_test_session(
        user_uuid=test_user.uuid,
        credential_uuid=test_credential.uuid,
    )
    return secret


@pytest_asyncio.fixture(scope="function")
async def regular_session_token(
    test_db: DB, regular_user: User, regular_credential: Credential
) -> str:
    """Create a session for a regular user and return the token."""
    _db_key, secret = create_test_session(
        user_uuid=regular_user.uuid,
        credential_uuid=regular_credential.uuid,
    )
    return secret


@pytest_asyncio.fixture(scope="function")
async def reset_token(test_db: DB, test_user: User, test_credential: Credential) -> str:
    """Create a reset token for the test user."""
    return create_reset_token(
        user_uuid=test_user.uuid,
        expiry=reset_expires(),
        token_type="reset",
    )


@pytest_asyncio.fixture(scope="function")
async def client(
    test_db: DB, realm_registry: realms.RealmRegistry
) -> AsyncGenerator[httpx.AsyncClient]:
    """Create an async test client for the FastAPI app."""
    transport = httpx.ASGITransport(app=app)
    async with httpx.AsyncClient(
        transport=transport,
        base_url="http://localhost:4401",
    ) as client:
        yield client


def auth_headers(token: str) -> dict[str, str]:
    """Return headers with auth cookie set."""
    return {"Cookie": f"{AUTH_COOKIE_NAME}={token}"}


def auth_cookie(token: str) -> httpx.Cookies:
    """Return cookies dict with auth cookie."""
    cookies = httpx.Cookies()
    cookies.set(AUTH_COOKIE_NAME, token, domain="localhost")
    return cookies


def create_test_session(
    user_uuid: UUID,
    credential_uuid: UUID,
    host: str = "localhost",
    ip: str = "127.0.0.1",
    user_agent: str = "pytest",
    duration: timedelta | None = None,
    rp_id: str = TEST_RP_ID,
) -> tuple[str, str]:
    """Create a test session. Returns (key, token) tuple.

    - key: str used for session lookup (base64url encoded)
    - token: stored in cookie/sent to client
    """
    if duration is None:
        duration = SESSION_LIFETIME
    if user_uuid not in ops_db._db.users:
        raise ValueError(f"User {user_uuid} not found")
    if credential_uuid not in ops_db._db.credentials:
        raise ValueError(f"Credential {credential_uuid} not found")
    now = datetime.now(UTC)

    # Generate token and derive key
    token = secrets.token_urlsafe(12)
    key = hash_secret("cookie", token)

    session = Session.create(
        user=user_uuid,
        credential=credential_uuid,
        key=key,
        host=host,
        ip=ip,
        user_agent=user_agent,
        validated=now,
        rp_id=rp_id,
    )
    if session.key in ops_db._db.sessions:
        raise ValueError("Session already exists")
    store = ops_db._db._store
    if store is None:
        raise RuntimeError("Test DB store is not initialized")
    with store.transaction("create_test_session"):
        session.store(now)
    return session.key, token


def create_test_image_bytes(
    *,
    image_format: str = "WEBP",
) -> bytes:
    """Return deterministic test upload bytes without image-library dependencies."""
    fixtures = {
        "WEBP": (
            b"RIFF\x1a\x00\x00\x00WEBPVP8 "
            b"\x0e\x00\x00\x000\x01\x00\x9d\x01*\x01\x00\x01\x00\x01\x00"
        ),
        "PNG": (
            b"\x89PNG\r\n\x1a\n"
            b"\x00\x00\x00\rIHDR"
            b"\x00\x00\x00\x01\x00\x00\x00\x01\x08\x02\x00\x00\x00"
            b"\x90wS\xde"
        ),
    }

    try:
        return fixtures[image_format.upper()]
    except KeyError as exc:
        raise ValueError(f"Unsupported test image format: {image_format}") from exc