Speed up tests.
This commit is contained in:
@@ -1,21 +1,123 @@
|
||||
# ABOUTME: Pytest configuration and fixtures for AnimalTrack tests.
|
||||
# ABOUTME: Provides database fixtures, test clients, and common utilities.
|
||||
# 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(tmp_path):
|
||||
"""Create a database with migrations applied."""
|
||||
db_path = str(tmp_path / "test.db")
|
||||
run_migrations(db_path, "migrations", verbose=False)
|
||||
return get_db(db_path)
|
||||
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
|
||||
@@ -26,15 +128,6 @@ def temp_migrations_dir(tmp_path):
|
||||
return migrations_path
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def seeded_db(migrated_db):
|
||||
"""Database with migrations and seed data applied."""
|
||||
from animaltrack.seeds import run_seeds
|
||||
|
||||
run_seeds(migrated_db)
|
||||
return migrated_db
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def fresh_db_path(tmp_path):
|
||||
"""Provide a path for a non-existent database file.
|
||||
@@ -68,3 +161,15 @@ def temp_db_path():
|
||||
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)
|
||||
|
||||
Reference in New Issue
Block a user