YPMS/tests/conftest.py

"""Shared pytest fixtures.

Uses SQLite for speed (see ARCHITECTURE.md §4.8). Tests exercise the framework
against an in-memory Postgres-shaped SQLAlchemy Base — no Alembic migrations
in the test path; models create their tables from metadata.

Environment defaults are set at *module import* time (not via fixture) so
they are in place before pytest collects any test module that transitively
imports `ypms.main` (which instantiates the FastAPI app at import time).
"""

from __future__ import annotations

import os
from collections.abc import Callable, Iterator
from typing import Any

os.environ.setdefault("DATABASE_URL", "sqlite:///:memory:")
# Test-only value, never deployed. 32+ bytes to satisfy the HS256 floor.
os.environ.setdefault("JWT_SECRET", "test-jwt-secret-at-least-32-bytes-long")  # gitleaks:allow
os.environ.setdefault(
    "COLUMN_ENCRYPTION_KEY", "test-col-enc-key-32bytes-______________"
)
os.environ.setdefault("ENV", "test")

import pytest  # noqa: E402
from fastapi.testclient import TestClient  # noqa: E402
from sqlalchemy import create_engine  # noqa: E402
from sqlalchemy.orm import Session, sessionmaker  # noqa: E402
from sqlalchemy.pool import StaticPool  # noqa: E402


@pytest.fixture
def db() -> Iterator[Session]:
    """Fresh in-memory SQLite DB per test with all model tables created.

    Import happens inside the fixture so the model modules register themselves
    on Base.metadata before create_all runs.
    """
    import ypms.models  # noqa: F401
    from ypms.db import Base

    # StaticPool + check_same_thread=False share the same in-memory DB across
    # threads. TestClient runs handlers in a worker thread, so without this
    # the fixture-scoped session and the request-scoped session see different
    # empty databases (SADeprecationWarning: SQLite objects created in a
    # thread can only be used in that same thread).
    engine = create_engine(
        "sqlite:///:memory:",
        connect_args={"check_same_thread": False},
        poolclass=StaticPool,
        future=True,
    )
    Base.metadata.create_all(engine)
    SessionLocal = sessionmaker(bind=engine, autoflush=False, autocommit=False, future=True)
    session = SessionLocal()
    try:
        yield session
    finally:
        session.close()
        engine.dispose()


@pytest.fixture
def client() -> Iterator[TestClient]:
    from ypms.main import create_app

    app = create_app()
    with TestClient(app) as c:
        yield c


# ---------------- seeded RBAC + authenticated API client ----------------

Headers = dict[str, str]


@pytest.fixture
def roles(db: Session) -> dict[str, Any]:
    """All seed roles with the real permissions matrix loaded."""
    from ypms.seed.permissions import seed_permissions
    from ypms.seed.run import seed_roles

    r = seed_roles(db)
    seed_permissions(db, r)
    db.flush()
    return r


_HASH_CACHE: list[str] = []


def _unused_password_hash(hasher: Callable[[str], str]) -> str:
    # PBKDF2 is deliberately slow; these users authenticate by minted token,
    # so one hash shared across the session is enough.
    if not _HASH_CACHE:
        _HASH_CACHE.append(hasher("unused-in-these-tests"))
    return _HASH_CACHE[0]


@pytest.fixture
def make_user(db: Session, roles: dict[str, Any]) -> Callable[..., tuple[Any, Headers]]:
    """Factory: make_user("name", "ROLE_A", ...) -> (User, auth headers)."""
    from ypms.auth.security import create_access_token, hash_password
    from ypms.users.models import ASSURANCE_STANDARD, User, UserRole

    def _make(username: str, *role_codes: str) -> tuple[User, Headers]:
        user = User(
            username=username,
            password_hash=_unused_password_hash(hash_password),
            is_active=True,
            assurance_level=ASSURANCE_STANDARD,
        )
        db.add(user)
        db.flush()
        for code in role_codes:
            db.add(UserRole(user_id=user.id, role_id=roles[code].id))
        db.flush()
        db.refresh(user)
        token = create_access_token(
            sub=username, uid=user.id, roles=list(role_codes),
            assurance_level=ASSURANCE_STANDARD, mfa_verified=False,
        )
        return user, {"Authorization": f"Bearer {token}"}

    return _make


@pytest.fixture(scope="session")
def _shared_app() -> Any:
    """One app per session: building ~240 routes costs ~0.1s per create_app()."""
    from ypms.main import create_app

    return create_app()


@pytest.fixture
def api(db: Session, _shared_app: Any) -> Iterator[TestClient]:
    """TestClient on the shared app, wired to the per-test SQLite session."""
    from ypms.db import get_db

    def _override() -> Iterator[Session]:
        yield db

    _shared_app.dependency_overrides[get_db] = _override
    with TestClient(_shared_app) as c:
        yield c
    _shared_app.dependency_overrides.clear()