Petru Paler
d53decdb66
Implements Step 4.3 from the plan: - Add selection/resolver.py with basic resolve_selection for validating animal IDs exist and are alive - Add ProductsProjection placeholder (stats tables added in Step 4.4) - Add ProductService with collect_product() function - Add PRODUCT_COLLECTED to EventAnimalsProjection for linking events to affected animals - Full test coverage for all new components 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
140 lines
4.7 KiB
Python
140 lines
4.7 KiB
Python
# ABOUTME: Tests for basic selection resolver.
|
|
# ABOUTME: Tests animal ID validation for product collection.
|
|
|
|
import time
|
|
|
|
import pytest
|
|
|
|
from animaltrack.events.payloads import AnimalCohortCreatedPayload
|
|
from animaltrack.events.store import EventStore
|
|
from animaltrack.projections import ProjectionRegistry
|
|
from animaltrack.projections.animal_registry import AnimalRegistryProjection
|
|
from animaltrack.projections.event_animals import EventAnimalsProjection
|
|
from animaltrack.projections.intervals import IntervalProjection
|
|
from animaltrack.selection import SelectionResolverError, resolve_selection
|
|
from animaltrack.services.animal import AnimalService
|
|
|
|
|
|
@pytest.fixture
|
|
def event_store(seeded_db):
|
|
"""Create an EventStore for testing."""
|
|
return EventStore(seeded_db)
|
|
|
|
|
|
@pytest.fixture
|
|
def projection_registry(seeded_db):
|
|
"""Create a ProjectionRegistry with animal projections registered."""
|
|
registry = ProjectionRegistry()
|
|
registry.register(AnimalRegistryProjection(seeded_db))
|
|
registry.register(EventAnimalsProjection(seeded_db))
|
|
registry.register(IntervalProjection(seeded_db))
|
|
return registry
|
|
|
|
|
|
@pytest.fixture
|
|
def animal_service(seeded_db, event_store, projection_registry):
|
|
"""Create an AnimalService for testing."""
|
|
return AnimalService(seeded_db, event_store, projection_registry)
|
|
|
|
|
|
@pytest.fixture
|
|
def valid_location_id(seeded_db):
|
|
"""Get a valid location ID from seeds."""
|
|
row = seeded_db.execute("SELECT id FROM locations WHERE name = 'Strip 1'").fetchone()
|
|
return row[0]
|
|
|
|
|
|
def make_cohort_payload(
|
|
location_id: str,
|
|
count: int = 3,
|
|
species: str = "duck",
|
|
) -> AnimalCohortCreatedPayload:
|
|
"""Create a cohort payload for testing."""
|
|
return AnimalCohortCreatedPayload(
|
|
species=species,
|
|
count=count,
|
|
life_stage="adult",
|
|
sex="unknown",
|
|
location_id=location_id,
|
|
origin="purchased",
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def animal_ids(seeded_db, animal_service, valid_location_id):
|
|
"""Create a cohort and return the animal IDs."""
|
|
payload = make_cohort_payload(valid_location_id, count=5)
|
|
ts_utc = int(time.time() * 1000)
|
|
event = animal_service.create_cohort(payload, ts_utc, "test_user")
|
|
return event.entity_refs["animal_ids"]
|
|
|
|
|
|
class TestResolveSelectionValid:
|
|
"""Tests for resolve_selection with valid inputs."""
|
|
|
|
def test_returns_validated_ids_when_all_exist(self, seeded_db, animal_ids):
|
|
"""resolve_selection returns the IDs when all are valid and alive."""
|
|
result = resolve_selection(seeded_db, animal_ids)
|
|
|
|
assert result == animal_ids
|
|
|
|
def test_handles_single_animal(self, seeded_db, animal_ids):
|
|
"""resolve_selection works with a single animal."""
|
|
single_id = [animal_ids[0]]
|
|
|
|
result = resolve_selection(seeded_db, single_id)
|
|
|
|
assert result == single_id
|
|
|
|
def test_handles_subset_of_animals(self, seeded_db, animal_ids):
|
|
"""resolve_selection works with a subset of animals."""
|
|
subset = animal_ids[:2]
|
|
|
|
result = resolve_selection(seeded_db, subset)
|
|
|
|
assert result == subset
|
|
|
|
|
|
class TestResolveSelectionErrors:
|
|
"""Tests for resolve_selection error cases."""
|
|
|
|
def test_raises_for_nonexistent_animal(self, seeded_db, animal_ids):
|
|
"""Raises SelectionResolverError for animal not found."""
|
|
fake_id = "01ARZ3NDEKTSV4RRFFQ69G5FAV"
|
|
ids_with_fake = animal_ids[:1] + [fake_id]
|
|
|
|
with pytest.raises(SelectionResolverError, match="not found"):
|
|
resolve_selection(seeded_db, ids_with_fake)
|
|
|
|
def test_raises_for_dead_animal(self, seeded_db, animal_ids):
|
|
"""Raises SelectionResolverError for animal with status != 'alive'."""
|
|
# Mark the first animal as dead
|
|
dead_id = animal_ids[0]
|
|
seeded_db.execute(
|
|
"UPDATE animal_registry SET status = 'dead' WHERE animal_id = ?",
|
|
(dead_id,),
|
|
)
|
|
|
|
with pytest.raises(SelectionResolverError, match="not alive"):
|
|
resolve_selection(seeded_db, [dead_id])
|
|
|
|
def test_raises_for_mixed_valid_invalid(self, seeded_db, animal_ids):
|
|
"""Raises SelectionResolverError when mix of valid and invalid animals."""
|
|
# Mark one as dead
|
|
dead_id = animal_ids[0]
|
|
seeded_db.execute(
|
|
"UPDATE animal_registry SET status = 'dead' WHERE animal_id = ?",
|
|
(dead_id,),
|
|
)
|
|
|
|
# Mix: one dead, one alive
|
|
mixed_ids = [dead_id, animal_ids[1]]
|
|
|
|
with pytest.raises(SelectionResolverError):
|
|
resolve_selection(seeded_db, mixed_ids)
|
|
|
|
def test_raises_for_empty_list(self, seeded_db):
|
|
"""Raises SelectionResolverError for empty resolved_ids list."""
|
|
with pytest.raises(SelectionResolverError, match="empty"):
|
|
resolve_selection(seeded_db, [])
|