patrick
2de0aac23a
ignore JWT related warning that was introduced when updating dependencies related to JWT secret size (test value of "secret" is too short).
278 lines
7.6 KiB
Python
278 lines
7.6 KiB
Python
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
|
|
|
|
|
|
@pytest.fixture
|
|
async def kosync_progress_service(
|
|
sessionmaker: async_sessionmaker[AsyncSession],
|
|
) -> AsyncGenerator[services.KosyncProgressService, None]:
|
|
"""Create KosyncProgressService instance."""
|
|
async with sessionmaker() as session:
|
|
async with services.KosyncProgressService.new(session) as service:
|
|
yield service
|
|
|
|
|
|
@pytest.fixture
|
|
async def kosync_device_service(
|
|
sessionmaker: async_sessionmaker[AsyncSession],
|
|
) -> AsyncGenerator[services.KosyncDeviceService, None]:
|
|
"""Create KosyncDeviceService instance."""
|
|
async with sessionmaker() as session:
|
|
async with services.KosyncDeviceService.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
|