a94609c07b
- conftest: default SHONAR_TEST_DATABASE_URL is a temp SQLite file, matching the bundled-lite engine; set the env var to a PG URL to exercise that path - models.py: add sqlite_where to the two partial unique indexes — without it SQLite built a FULL unique index on recording_id (WHERE not carried over), wrongly blocking a second export asset per recording - test_m9: pass UUID objects (not str) to direct ORM inserts/gets; SQLite's GUID bind processor rejects strings (asyncpg tolerated them) Verified: pytest -q = 78 passed, 1 skipped; ruff check clean
118 lines
4.3 KiB
Python
118 lines
4.3 KiB
Python
"""Pytest fixtures.
|
|
|
|
Default: SQLite (aiosqlite) in a temp dir — matches the desktop
|
|
bundled-lite engine and needs no services running. Set
|
|
``SHONAR_TEST_DATABASE_URL`` to a PostgreSQL URL (e.g.
|
|
``postgresql+asyncpg://shonar:shonar@localhost:5432/shonar_test``) to
|
|
exercise the Postgres path, including the tsvector FTS columns.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import tempfile
|
|
from collections.abc import AsyncIterator
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
import pytest_asyncio
|
|
|
|
# Configure env BEFORE importing the app so Settings picks it up.
|
|
_db_dir = tempfile.mkdtemp(prefix="shonar-test-db-")
|
|
TEST_DB = os.environ.get(
|
|
"SHONAR_TEST_DATABASE_URL",
|
|
# Bundled-lite default: throwaway SQLite file; no services required.
|
|
f"sqlite+aiosqlite:///{os.path.join(_db_dir, 'test.db')}",
|
|
)
|
|
os.environ["SHONAR_DATABASE_URL"] = TEST_DB
|
|
os.environ["SHONAR_SECRET_KEY"] = "test-secret-key-0123456789abcdef0123456789abcdef"
|
|
os.environ["SHONAR_STORAGE_BACKEND"] = "local"
|
|
# Effectively disable the auth rate limit under test (dedicated tests cover
|
|
# the limiter behaviour itself).
|
|
os.environ["SHONAR_RATE_LIMIT_AUTH"] = "10000/minute"
|
|
|
|
_tmp_storage = tempfile.mkdtemp(prefix="shonar-test-storage-")
|
|
os.environ["SHONAR_STORAGE_PATH"] = _tmp_storage
|
|
|
|
|
|
@pytest_asyncio.fixture(scope="session", loop_scope="session")
|
|
async def _setup_db() -> AsyncIterator[None]:
|
|
from shonar.db import models # noqa: F401
|
|
from shonar.db.base import Base
|
|
from shonar.db.session import dispose_engine, get_engine
|
|
|
|
engine = get_engine()
|
|
async with engine.begin() as conn:
|
|
await conn.run_sync(Base.metadata.drop_all)
|
|
await conn.run_sync(Base.metadata.create_all)
|
|
# The tsvector generated columns live only in migration
|
|
# fts0000000001 (not in the ORM models), so create_all misses them.
|
|
# Apply the same DDL the migration applies (M9 search needs them).
|
|
if engine.dialect.name == "postgresql":
|
|
# Reuse the real migration's DDL (not a copy) via a sync
|
|
# MigrationContext — op.execute() is synchronous there.
|
|
import importlib.util
|
|
from pathlib import Path
|
|
|
|
def _apply(sync_conn):
|
|
from alembic.migration import MigrationContext
|
|
from alembic.operations import Operations
|
|
|
|
spec = importlib.util.spec_from_file_location(
|
|
"fts_migration",
|
|
Path(__file__).resolve().parents[1]
|
|
/ "migrations/versions/fts0000000001_fts_columns.py",
|
|
)
|
|
assert spec is not None and spec.loader is not None
|
|
mod = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(mod)
|
|
ctx = MigrationContext.configure(sync_conn)
|
|
with Operations.context(ctx):
|
|
mod.upgrade()
|
|
|
|
await conn.run_sync(_apply)
|
|
yield
|
|
await dispose_engine()
|
|
|
|
|
|
@pytest_asyncio.fixture(loop_scope="session", autouse=True)
|
|
async def clean_db(_setup_db: None) -> AsyncIterator[None]:
|
|
"""Truncate between tests for isolation."""
|
|
yield
|
|
from sqlalchemy import delete, text
|
|
|
|
from shonar.db.base import Base
|
|
from shonar.db.session import _session_factory, get_engine # type: ignore[attr-defined]
|
|
|
|
assert _session_factory is not None
|
|
async with _session_factory() as s:
|
|
if get_engine().dialect.name == "postgresql":
|
|
await s.execute(
|
|
text(
|
|
"TRUNCATE users, devices, refresh_tokens, recordings, assets, "
|
|
"upload_sessions, upload_chunks, transcripts, summaries, tags, "
|
|
"recording_tags, processing_jobs, export_jobs, app_settings "
|
|
"RESTART IDENTITY CASCADE"
|
|
)
|
|
)
|
|
else:
|
|
# SQLite: delete child-first (reverse dependency order).
|
|
for table in reversed(Base.metadata.sorted_tables):
|
|
await s.execute(delete(table))
|
|
await s.commit()
|
|
|
|
|
|
@pytest_asyncio.fixture(loop_scope="session")
|
|
async def client(_setup_db) -> AsyncIterator:
|
|
from httpx import ASGITransport, AsyncClient
|
|
|
|
from shonar.main import app
|
|
|
|
transport = ASGITransport(app=app)
|
|
async with AsyncClient(transport=transport, base_url="http://test") as c:
|
|
yield c
|
|
|
|
|
|
@pytest.fixture()
|
|
def storage_root() -> Path:
|
|
return Path(_tmp_storage)
|