Implements Step 3.3: Animal Cohort Creation - Add AnimalRegistryProjection for animal_registry and live_animals_by_location - Add EventAnimalsProjection for event_animals link table - Add IntervalProjection for location and attribute intervals - Add AnimalService with create_cohort() for coordinating event + projections - Add seeded_db fixture to conftest.py - Update projections/__init__.py with new exports All operations atomic within single transaction. Includes validation for location (exists, active) and species (exists, active). 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
71 lines
1.9 KiB
Python
71 lines
1.9 KiB
Python
# ABOUTME: Pytest configuration and fixtures for AnimalTrack tests.
|
|
# ABOUTME: Provides database fixtures, test clients, and common utilities.
|
|
|
|
import os
|
|
import tempfile
|
|
|
|
import pytest
|
|
|
|
from animaltrack.db import get_db
|
|
from animaltrack.migrations import run_migrations
|
|
|
|
|
|
@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)
|
|
|
|
|
|
@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 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.
|
|
|
|
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)
|