animaltrack/tests/conftest.py

# ABOUTME: Pytest configuration and fixtures for AnimalTrack tests.
# ABOUTME: Provides database fixtures using template pattern for fast test isolation.

import os
import shutil
import sqlite3
import tempfile
from pathlib import Path

import pytest
from filelock import FileLock

from animaltrack.db import get_db
from animaltrack.migrations import run_migrations


def _get_template_dir() -> Path:
    """Get shared template directory for all workers.

    Uses system temp directory with a stable name so all pytest-xdist
    workers find the same templates.
    """
    return Path(tempfile.gettempdir()) / "animaltrack_test_templates"


def _checkpoint_wal(db_path: str) -> None:
    """Checkpoint WAL file into main database for clean copy.

    SQLite's WAL mode keeps changes in a separate -wal file. Before copying,
    we checkpoint to merge all changes into the main .db file, then switch
    to DELETE mode temporarily for a clean single-file state.
    """
    conn = sqlite3.connect(db_path)
    try:
        conn.execute("PRAGMA wal_checkpoint(TRUNCATE)")
        conn.execute("PRAGMA journal_mode=DELETE")
    finally:
        conn.close()


def _copy_db(src_path: Path, dest_path: Path) -> None:
    """Copy SQLite database file.

    Assumes source has been checkpointed and is in DELETE journal mode.
    """
    shutil.copy2(src_path, dest_path)


@pytest.fixture(scope="session")
def template_migrated_db(tmp_path_factory):
    """Session-scoped template database with migrations applied.

    Creates the template once across all xdist workers using file locking.
    Returns the path to the template database (not a connection).
    """
    template_dir = _get_template_dir()
    template_dir.mkdir(exist_ok=True)

    template_path = template_dir / "migrated_template.db"
    lock_path = template_dir / "migrated_template.lock"
    marker_path = template_dir / "migrated_template.ready"

    with FileLock(str(lock_path)):
        if not marker_path.exists():
            # First worker creates the template
            run_migrations(str(template_path), "migrations", verbose=False)
            _checkpoint_wal(str(template_path))
            marker_path.touch()

    return template_path


@pytest.fixture(scope="session")
def template_seeded_db(template_migrated_db, tmp_path_factory):
    """Session-scoped template database with migrations and seeds.

    Creates the seeded template once across all xdist workers.
    Returns the path to the template database (not a connection).
    """
    from animaltrack.seeds import run_seeds

    template_dir = _get_template_dir()
    template_path = template_dir / "seeded_template.db"
    lock_path = template_dir / "seeded_template.lock"
    marker_path = template_dir / "seeded_template.ready"

    with FileLock(str(lock_path)):
        if not marker_path.exists():
            # Copy from migrated template, then seed
            _copy_db(template_migrated_db, template_path)
            db = get_db(str(template_path))
            run_seeds(db)
            # Close connection and checkpoint
            del db
            _checkpoint_wal(str(template_path))
            marker_path.touch()

    return template_path


@pytest.fixture
def migrated_db(template_migrated_db, tmp_path):
    """Function-scoped database with migrations (isolated copy).

    Creates a fast copy of the template for test isolation.
    """
    db_path = tmp_path / "test.db"
    _copy_db(template_migrated_db, db_path)
    return get_db(str(db_path))


@pytest.fixture
def seeded_db(template_seeded_db, tmp_path):
    """Function-scoped database with migrations and seeds (isolated copy).

    Creates a fast copy of the seeded template for test isolation.
    """
    db_path = tmp_path / "test.db"
    _copy_db(template_seeded_db, db_path)
    return get_db(str(db_path))


@pytest.fixture
def temp_migrations_dir(tmp_path):
    """Create a temporary migrations directory for testing."""
    migrations_path = tmp_path / "migrations"
    migrations_path.mkdir()
    return migrations_path


@pytest.fixture
def fresh_db_path(tmp_path):
    """Provide a path for a non-existent database file.

    Unlike temp_db_path, this does not create the file - it just provides
    a path. This is needed for fastmigrate which expects to create the db.
    """
    db_path = tmp_path / "test.db"
    yield db_path
    # Cleanup
    if db_path.exists():
        db_path.unlink()
    # Also clean up WAL files if they exist
    for suffix in ["-shm", "-wal", "-journal"]:
        wal_path = db_path.with_suffix(db_path.suffix + suffix)
        if wal_path.exists():
            wal_path.unlink()


@pytest.fixture
def temp_db_path():
    """Create a temporary database file path for testing."""
    with tempfile.NamedTemporaryFile(suffix=".db", delete=False) as f:
        db_path = f.name
    yield db_path
    # Cleanup
    if os.path.exists(db_path):
        os.unlink(db_path)
    # Also clean up WAL files if they exist
    for suffix in ["-shm", "-wal", "-journal"]:
        wal_path = db_path + suffix
        if os.path.exists(wal_path):
            os.unlink(wal_path)


def pytest_sessionfinish(session, exitstatus):
    """Clean up template databases after test session.

    Only the master worker (or single-process mode) should clean up.
    """
    worker_id = os.environ.get("PYTEST_XDIST_WORKER", "master")
    if worker_id == "master":
        template_dir = _get_template_dir()
        if template_dir.exists():
            shutil.rmtree(template_dir, ignore_errors=True)