chitai/backend/tests/conftest.py

from pathlib import Path
from uuid import uuid4
import pytest

from advanced_alchemy.base import UUIDAuditBase
from litestar.testing import AsyncTestClient
from sqlalchemy import text
from sqlalchemy.ext.asyncio import (
    AsyncSession,
    AsyncEngine,
    async_sessionmaker,
    create_async_engine,
)
from sqlalchemy.engine import URL
from sqlalchemy.pool import NullPool

from chitai.database import models as m
from chitai.config import settings

from collections.abc import AsyncGenerator

from litestar import Litestar

from pytest_databases.docker.postgres import PostgresService
from sqlalchemy.orm import selectinload

pytest_plugins = [
    "tests.data_fixtures",
    "pytest_databases.docker",
    "pytest_databases.docker.postgres",
]

# Set the environment to use the testing database
# os.environ.update(
#     {
#         "POSTGRES_DB": "chitai_testing"
#     }
# )


@pytest.fixture(autouse=True)
def _patch_settings(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
    monkeypatch.setattr(settings, "book_cover_path", f"{tmp_path}/covers")


@pytest.fixture(name="engine")
async def fx_engine(
    postgres_service: PostgresService,
) -> AsyncGenerator[AsyncEngine, None]:
    """PostgreSQL instance for testing"""
    engine = create_async_engine(
        URL(
            drivername="postgresql+asyncpg",
            username=postgres_service.user,
            password=postgres_service.password,
            host=postgres_service.host,
            port=postgres_service.port,
            database=postgres_service.database,
            query={},  # type:ignore[arg-type]
        ),
        echo=False,
        poolclass=NullPool,
    )

    # Add pg_trgm extension
    async with engine.begin() as conn:
        await conn.execute(text("CREATE EXTENSION IF NOT EXISTS pg_trgm"))

    # Create all tables
    metadata = UUIDAuditBase.registry.metadata
    async with engine.begin() as conn:
        await conn.run_sync(metadata.create_all)

    yield engine

    # Clean up
    async with engine.begin() as conn:
        await conn.run_sync(metadata.drop_all)
    await engine.dispose()


@pytest.fixture(name="sessionmaker")
def fx_sessionmaker(engine: AsyncEngine) -> async_sessionmaker[AsyncSession]:
    """Create sessionmaker factory."""
    return async_sessionmaker(bind=engine, expire_on_commit=False)


@pytest.fixture
async def session(
    sessionmaker: async_sessionmaker[AsyncSession],
) -> AsyncGenerator[AsyncSession, None]:
    """Create database session for tests."""
    async with sessionmaker() as session:
        yield session
        await session.rollback()
        await session.close()


@pytest.fixture
async def app() -> Litestar:
    """Create Litestar app for testing."""

    from chitai.app import create_app

    return create_app()


@pytest.fixture
async def client(app: Litestar) -> AsyncGenerator[AsyncTestClient, None]:
    """Create test client."""
    async with AsyncTestClient(app=app) as client:
        yield client


@pytest.fixture
async def authenticated_client(
    client: AsyncTestClient, test_user: m.User
) -> AsyncTestClient:
    """Create authenticated test client."""

    # login and set auth headers
    login_response = await client.post(
        "access/login", data={"email": test_user.email, "password": "password123"}
    )

    assert login_response.status_code == 201

    result = login_response.json()
    token = result["access_token"]
    client.headers.update({"Authorization": f"Bearer {token}"})

    return client


@pytest.fixture
async def other_authenticated_client(
    app: Litestar, test_user2: m.User
) -> AsyncGenerator[AsyncTestClient, None]:
    """Create second authenticated test client as different user."""
    async with AsyncTestClient(app=app) as other_client:
        login_response = await other_client.post(
            "access/login", data={"email": test_user2.email, "password": "password234"}
        )
        assert login_response.status_code == 201
        result = login_response.json()
        token = result["access_token"]
        other_client.headers.update({"Authorization": f"Bearer {token}"})
        yield other_client


# Service fixtures
from chitai import services


@pytest.fixture
async def user_service(
    sessionmaker: async_sessionmaker[AsyncSession],
) -> AsyncGenerator[services.UserService, None]:
    """Create UserService instance."""
    async with sessionmaker() as session:
        async with services.UserService.new(session) as service:
            yield service


@pytest.fixture
async def library_service(
    sessionmaker: async_sessionmaker[AsyncSession],
) -> AsyncGenerator[services.LibraryService, None]:
    """Create LibraryService instance."""
    async with sessionmaker() as session:
        async with services.LibraryService.new(session) as service:
            yield service


@pytest.fixture
async def books_service(
    sessionmaker: async_sessionmaker[AsyncSession],
) -> AsyncGenerator[services.BookService, None]:
    """Create BookService instance."""
    async with sessionmaker() as session:
        async with services.BookService.new(
            session,
            load=[
                selectinload(m.Book.author_links).selectinload(m.BookAuthorLink.author),
                selectinload(m.Book.tag_links).selectinload(m.BookTagLink.tag),
                m.Book.publisher,
                m.Book.files,
                m.Book.identifiers,
                m.Book.series,
            ],
        ) as service:
            yield service


@pytest.fixture
async def bookshelf_service(
    sessionmaker: async_sessionmaker[AsyncSession],
) -> AsyncGenerator[services.ShelfService]:
    """Create ShelfService instance."""
    async with sessionmaker() as session:
        async with services.ShelfService.new(session) as service:
            yield service


# Data fixtures


@pytest.fixture
async def test_user(session: AsyncSession) -> AsyncGenerator[m.User, None]:
    """Create a test user."""

    unique_id = str(uuid4())[:8]
    user = m.User(
        email=f"user{unique_id}@example.com",
        password="password123",
    )

    session.add(user)
    await session.commit()
    await session.refresh(user)
    yield user


@pytest.fixture
async def test_user2(session: AsyncSession) -> AsyncGenerator[m.User, None]:
    """Create another test user."""

    unique_id = str(uuid4())[:8]
    user = m.User(
        email=f"user{unique_id}@example.com",
        password="password234",
    )

    session.add(user)
    await session.commit()
    await session.refresh(user)
    yield user


@pytest.fixture
async def test_library(
    session: AsyncSession, tmp_path: Path
) -> AsyncGenerator[m.Library, None]:
    """Create a test library."""

    library = m.Library(
        name="Testing Library",
        slug="testing-library",
        root_path=str(tmp_path),
        path_template="{author_name}/{title}.{ext}",
        read_only=False,
    )

    session.add(library)
    await session.commit()
    await session.refresh(library)
    yield library