Alo/animaltrack
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: add animal movement projection and service

Browse Source 
Implement AnimalMoved event handling:
- Update AnimalRegistryProjection for move events
- Update IntervalProjection to close/open location intervals
- Update EventAnimalsProjection to link move events to animals
- Add move_animals() to AnimalService with validations

Validations include:
- Destination location must exist and be active
- All animals must be from a single location
- Cannot move to the same location as current

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
Petru
2025-12-29 07:02:19 +00:00
parent 144d23235e
commit b89ea41d63
7 changed files with 1007 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files

287
tests/test_projection_animal_registry.py
View File

@@ -403,3 +403,290 @@ class TestAnimalRegistryProjectionRevert:
row = seeded_db.execute("SELECT animal_id FROM animal_registry").fetchone()
assert row[0] == animal_ids_2[0]
# =============================================================================
# AnimalMoved Tests
# =============================================================================
def make_move_event(
event_id: str,
animal_ids: list[str],
from_location_id: str,
to_location_id: str,
ts_utc: int = 1704067300000,
) -> Event:
"""Create a test AnimalMoved event."""
from animaltrack.events.types import ANIMAL_MOVED
return Event(
id=event_id,
type=ANIMAL_MOVED,
ts_utc=ts_utc,
actor="test_user",
entity_refs={
"from_location_id": from_location_id,
"to_location_id": to_location_id,
"animal_ids": animal_ids,
},
payload={
"to_location_id": to_location_id,
"resolved_ids": animal_ids,
"notes": None,
},
version=1,
)
class TestAnimalRegistryProjectionMoveEventTypes:
"""Tests for get_event_types method including move."""
def test_handles_animal_moved(self, seeded_db):
"""Projection handles AnimalMoved event type."""
from animaltrack.events.types import ANIMAL_MOVED
projection = AnimalRegistryProjection(seeded_db)
assert ANIMAL_MOVED in projection.get_event_types()
class TestAnimalRegistryProjectionApplyMove:
"""Tests for apply() on AnimalMoved."""
def test_updates_location_in_animal_registry(self, seeded_db):
"""Apply move updates location_id in animal_registry."""
# Get two location IDs
strip1 = seeded_db.execute("SELECT id FROM locations WHERE name = 'Strip 1'").fetchone()[0]
strip2 = seeded_db.execute("SELECT id FROM locations WHERE name = 'Strip 2'").fetchone()[0]
# First create a cohort at Strip 1
animal_ids = ["01ARZ3NDEKTSV4RRFFQ69G5A01"]
projection = AnimalRegistryProjection(seeded_db)
cohort_event = make_cohort_event(animal_ids, location_id=strip1)
projection.apply(cohort_event)
# Verify initial location
row = seeded_db.execute(
"SELECT location_id FROM animal_registry WHERE animal_id = ?",
(animal_ids[0],),
).fetchone()
assert row[0] == strip1
# Now move to Strip 2
move_event = make_move_event(
"01ARZ3NDEKTSV4RRFFQ69G5002",
animal_ids,
from_location_id=strip1,
to_location_id=strip2,
)
projection.apply(move_event)
# Verify new location
row = seeded_db.execute(
"SELECT location_id FROM animal_registry WHERE animal_id = ?",
(animal_ids[0],),
).fetchone()
assert row[0] == strip2
def test_updates_last_event_utc_in_registry(self, seeded_db):
"""Apply move updates last_event_utc in animal_registry."""
strip1 = seeded_db.execute("SELECT id FROM locations WHERE name = 'Strip 1'").fetchone()[0]
strip2 = seeded_db.execute("SELECT id FROM locations WHERE name = 'Strip 2'").fetchone()[0]
animal_ids = ["01ARZ3NDEKTSV4RRFFQ69G5A01"]
projection = AnimalRegistryProjection(seeded_db)
cohort_event = make_cohort_event(animal_ids, location_id=strip1, ts_utc=1704067200000)
projection.apply(cohort_event)
move_ts = 1704067300000
move_event = make_move_event(
"01ARZ3NDEKTSV4RRFFQ69G5002",
animal_ids,
from_location_id=strip1,
to_location_id=strip2,
ts_utc=move_ts,
)
projection.apply(move_event)
row = seeded_db.execute(
"SELECT last_event_utc FROM animal_registry WHERE animal_id = ?",
(animal_ids[0],),
).fetchone()
assert row[0] == move_ts
def test_updates_location_in_live_animals(self, seeded_db):
"""Apply move updates location_id in live_animals_by_location."""
strip1 = seeded_db.execute("SELECT id FROM locations WHERE name = 'Strip 1'").fetchone()[0]
strip2 = seeded_db.execute("SELECT id FROM locations WHERE name = 'Strip 2'").fetchone()[0]
animal_ids = ["01ARZ3NDEKTSV4RRFFQ69G5A01"]
projection = AnimalRegistryProjection(seeded_db)
cohort_event = make_cohort_event(animal_ids, location_id=strip1)
projection.apply(cohort_event)
move_event = make_move_event(
"01ARZ3NDEKTSV4RRFFQ69G5002",
animal_ids,
from_location_id=strip1,
to_location_id=strip2,
)
projection.apply(move_event)
row = seeded_db.execute(
"SELECT location_id FROM live_animals_by_location WHERE animal_id = ?",
(animal_ids[0],),
).fetchone()
assert row[0] == strip2
def test_updates_last_move_utc_in_live_animals(self, seeded_db):
"""Apply move updates last_move_utc in live_animals_by_location."""
strip1 = seeded_db.execute("SELECT id FROM locations WHERE name = 'Strip 1'").fetchone()[0]
strip2 = seeded_db.execute("SELECT id FROM locations WHERE name = 'Strip 2'").fetchone()[0]
animal_ids = ["01ARZ3NDEKTSV4RRFFQ69G5A01"]
projection = AnimalRegistryProjection(seeded_db)
cohort_event = make_cohort_event(animal_ids, location_id=strip1)
projection.apply(cohort_event)
# Verify last_move_utc is NULL initially
row = seeded_db.execute(
"SELECT last_move_utc FROM live_animals_by_location WHERE animal_id = ?",
(animal_ids[0],),
).fetchone()
assert row[0] is None
move_ts = 1704067300000
move_event = make_move_event(
"01ARZ3NDEKTSV4RRFFQ69G5002",
animal_ids,
from_location_id=strip1,
to_location_id=strip2,
ts_utc=move_ts,
)
projection.apply(move_event)
row = seeded_db.execute(
"SELECT last_move_utc FROM live_animals_by_location WHERE animal_id = ?",
(animal_ids[0],),
).fetchone()
assert row[0] == move_ts
def test_moves_multiple_animals(self, seeded_db):
"""Apply move updates all animals in resolved_ids."""
strip1 = seeded_db.execute("SELECT id FROM locations WHERE name = 'Strip 1'").fetchone()[0]
strip2 = seeded_db.execute("SELECT id FROM locations WHERE name = 'Strip 2'").fetchone()[0]
animal_ids = [
"01ARZ3NDEKTSV4RRFFQ69G5A01",
"01ARZ3NDEKTSV4RRFFQ69G5A02",
"01ARZ3NDEKTSV4RRFFQ69G5A03",
]
projection = AnimalRegistryProjection(seeded_db)
cohort_event = make_cohort_event(animal_ids, location_id=strip1)
projection.apply(cohort_event)
move_event = make_move_event(
"01ARZ3NDEKTSV4RRFFQ69G5002",
animal_ids,
from_location_id=strip1,
to_location_id=strip2,
)
projection.apply(move_event)
# All animals should now be at strip2
for animal_id in animal_ids:
row = seeded_db.execute(
"SELECT location_id FROM animal_registry WHERE animal_id = ?",
(animal_id,),
).fetchone()
assert row[0] == strip2
class TestAnimalRegistryProjectionRevertMove:
"""Tests for revert() on AnimalMoved."""
def test_revert_restores_original_location_in_registry(self, seeded_db):
"""Revert move restores location_id in animal_registry."""
strip1 = seeded_db.execute("SELECT id FROM locations WHERE name = 'Strip 1'").fetchone()[0]
strip2 = seeded_db.execute("SELECT id FROM locations WHERE name = 'Strip 2'").fetchone()[0]
animal_ids = ["01ARZ3NDEKTSV4RRFFQ69G5A01"]
projection = AnimalRegistryProjection(seeded_db)
cohort_event = make_cohort_event(animal_ids, location_id=strip1, ts_utc=1704067200000)
projection.apply(cohort_event)
move_event = make_move_event(
"01ARZ3NDEKTSV4RRFFQ69G5002",
animal_ids,
from_location_id=strip1,
to_location_id=strip2,
ts_utc=1704067300000,
)
projection.apply(move_event)
# Verify moved
row = seeded_db.execute(
"SELECT location_id FROM animal_registry WHERE animal_id = ?",
(animal_ids[0],),
).fetchone()
assert row[0] == strip2
# Revert the move
projection.revert(move_event)
# Location should be restored
row = seeded_db.execute(
"SELECT location_id FROM animal_registry WHERE animal_id = ?",
(animal_ids[0],),
).fetchone()
assert row[0] == strip1
def test_revert_restores_original_location_in_live_animals(self, seeded_db):
"""Revert move restores location_id in live_animals_by_location."""
strip1 = seeded_db.execute("SELECT id FROM locations WHERE name = 'Strip 1'").fetchone()[0]
strip2 = seeded_db.execute("SELECT id FROM locations WHERE name = 'Strip 2'").fetchone()[0]
animal_ids = ["01ARZ3NDEKTSV4RRFFQ69G5A01"]
projection = AnimalRegistryProjection(seeded_db)
cohort_event = make_cohort_event(animal_ids, location_id=strip1)
projection.apply(cohort_event)
move_event = make_move_event(
"01ARZ3NDEKTSV4RRFFQ69G5002",
animal_ids,
from_location_id=strip1,
to_location_id=strip2,
)
projection.apply(move_event)
projection.revert(move_event)
row = seeded_db.execute(
"SELECT location_id FROM live_animals_by_location WHERE animal_id = ?",
(animal_ids[0],),
).fetchone()
assert row[0] == strip1
def test_revert_clears_last_move_utc_if_first_move(self, seeded_db):
"""Revert first move clears last_move_utc to NULL."""
strip1 = seeded_db.execute("SELECT id FROM locations WHERE name = 'Strip 1'").fetchone()[0]
strip2 = seeded_db.execute("SELECT id FROM locations WHERE name = 'Strip 2'").fetchone()[0]
animal_ids = ["01ARZ3NDEKTSV4RRFFQ69G5A01"]
projection = AnimalRegistryProjection(seeded_db)
cohort_event = make_cohort_event(animal_ids, location_id=strip1)
projection.apply(cohort_event)
move_event = make_move_event(
"01ARZ3NDEKTSV4RRFFQ69G5002",
animal_ids,
from_location_id=strip1,
to_location_id=strip2,
)
projection.apply(move_event)
projection.revert(move_event)
row = seeded_db.execute(
"SELECT last_move_utc FROM live_animals_by_location WHERE animal_id = ?",
(animal_ids[0],),
).fetchone()
assert row[0] is None