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

feat: implement animal lifecycle events (Step 6.3)

Browse Source 
Add 5 animal lifecycle event handlers with TDD:
- HatchRecorded: Creates hatchling animals at brood/event location
- AnimalOutcome: Records death/harvest/sold with yields, status updates
- AnimalPromoted: Sets identified flag, nickname, optionally updates sex/repro_status
- AnimalMerged: Merges animal records, creates aliases, removes merged from live roster
- AnimalStatusCorrected: Admin-only status correction with required reason

All events include:
- Projection handlers in animal_registry.py and intervals.py
- Event-animal linking in event_animals.py
- Service methods with validation in animal.py
- 51 unit tests covering event creation, projections, and validation
- E2E test #7 (harvest with yields) per spec §21.7

🤖 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 19:20:33 +00:00
parent da95e85452
commit 1153f6c5b6
7 changed files with 2728 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files

325
tests/test_e2e_harvest.py Normal file
View File

@@ -0,0 +1,325 @@
# ABOUTME: E2E test #7 from spec section 21.7: Harvest with yields.
# ABOUTME: Tests AnimalOutcome=harvest with yields, status updates, and live count changes.
import time
import pytest
from animaltrack.events.payloads import (
AnimalCohortCreatedPayload,
AnimalOutcomePayload,
)
from animaltrack.events.store import EventStore
@pytest.fixture
def now_utc():
"""Current time in milliseconds since epoch."""
return int(time.time() * 1000)
@pytest.fixture
def full_projection_registry(seeded_db):
"""Create a ProjectionRegistry with all projections."""
from animaltrack.projections import ProjectionRegistry
from animaltrack.projections.animal_registry import AnimalRegistryProjection
from animaltrack.projections.event_animals import EventAnimalsProjection
from animaltrack.projections.feed import FeedInventoryProjection
from animaltrack.projections.intervals import IntervalProjection
from animaltrack.projections.products import ProductsProjection
registry = ProjectionRegistry()
registry.register(AnimalRegistryProjection(seeded_db))
registry.register(IntervalProjection(seeded_db))
registry.register(EventAnimalsProjection(seeded_db))
registry.register(ProductsProjection(seeded_db))
registry.register(FeedInventoryProjection(seeded_db))
return registry
@pytest.fixture
def services(seeded_db, full_projection_registry):
"""Create all services needed for E2E test."""
from animaltrack.services.animal import AnimalService
event_store = EventStore(seeded_db)
return {
"db": seeded_db,
"event_store": event_store,
"registry": full_projection_registry,
"animal_service": AnimalService(seeded_db, event_store, full_projection_registry),
}
@pytest.fixture
def strip1_id(seeded_db):
"""Get Strip 1 location ID from seeds."""
return seeded_db.execute("SELECT id FROM locations WHERE name = 'Strip 1'").fetchone()[0]
@pytest.fixture
def strip2_id(seeded_db):
"""Get Strip 2 location ID from seeds."""
return seeded_db.execute("SELECT id FROM locations WHERE name = 'Strip 2'").fetchone()[0]
@pytest.fixture
def harvest_scenario(seeded_db, services, now_utc, strip1_id, strip2_id):
"""Set up harvest test scenario.
Creates:
- 5 adult female ducks at Strip 1
- 5 adult female ducks at Strip 2 (2 will be harvested)
Returns dict with animal IDs and references.
"""
one_day_ms = 24 * 60 * 60 * 1000
animal_creation_ts = now_utc - one_day_ms
# Create 5 adult female ducks at Strip 1
cohort1_payload = AnimalCohortCreatedPayload(
species="duck",
count=5,
life_stage="adult",
sex="female",
location_id=strip1_id,
origin="purchased",
)
cohort1_event = services["animal_service"].create_cohort(
cohort1_payload, animal_creation_ts, "test_user"
)
strip1_animal_ids = cohort1_event.entity_refs["animal_ids"]
# Create 5 adult female ducks at Strip 2
cohort2_payload = AnimalCohortCreatedPayload(
species="duck",
count=5,
life_stage="adult",
sex="female",
location_id=strip2_id,
origin="purchased",
)
cohort2_event = services["animal_service"].create_cohort(
cohort2_payload, animal_creation_ts, "test_user"
)
strip2_animal_ids = cohort2_event.entity_refs["animal_ids"]
return {
"strip1_id": strip1_id,
"strip2_id": strip2_id,
"strip1_animal_ids": strip1_animal_ids,
"strip2_animal_ids": strip2_animal_ids,
"animal_creation_ts": animal_creation_ts,
}
class TestE2E7HarvestWithYields:
"""E2E test #7: Harvest with yields from spec section 21.7.
At Strip 2 select 2 adult females -> AnimalOutcome=harvest with yields:
- meat.part.breast.duck qty=2 weight_kg=1.4
- fat.rendered.duck qty=1 weight_kg=0.3
Expect:
- Both animals status=harvested
- Strip 2 live female count -2
- Yields present in history/export
- EggStats unchanged
"""
def test_harvest_updates_status_to_harvested(
self, seeded_db, services, now_utc, harvest_scenario
):
"""Both harvested animals should have status=harvested."""
strip2_animal_ids = harvest_scenario["strip2_animal_ids"]
animals_to_harvest = strip2_animal_ids[:2]
yield_items = [
{
"product_code": "meat.part.breast.duck",
"unit": "kg",
"quantity": 2,
"weight_kg": 1.4,
},
{"product_code": "fat.rendered.duck", "unit": "kg", "quantity": 1, "weight_kg": 0.3},
]
outcome_payload = AnimalOutcomePayload(
outcome="harvest",
resolved_ids=animals_to_harvest,
yield_items=yield_items,
)
services["animal_service"].record_outcome(outcome_payload, now_utc, "test_user")
# Verify both animals have status=harvested
for animal_id in animals_to_harvest:
row = seeded_db.execute(
"SELECT status FROM animal_registry WHERE animal_id = ?",
(animal_id,),
).fetchone()
assert row[0] == "harvested"
def test_harvest_decreases_live_female_count(
self, seeded_db, services, now_utc, harvest_scenario
):
"""Strip 2 live female count should decrease by 2."""
strip2_id = harvest_scenario["strip2_id"]
strip2_animal_ids = harvest_scenario["strip2_animal_ids"]
animals_to_harvest = strip2_animal_ids[:2]
# Count before harvest
count_before = seeded_db.execute(
"""SELECT COUNT(*) FROM live_animals_by_location
WHERE location_id = ? AND sex = 'female'""",
(strip2_id,),
).fetchone()[0]
assert count_before == 5
yield_items = [
{
"product_code": "meat.part.breast.duck",
"unit": "kg",
"quantity": 2,
"weight_kg": 1.4,
},
{"product_code": "fat.rendered.duck", "unit": "kg", "quantity": 1, "weight_kg": 0.3},
]
outcome_payload = AnimalOutcomePayload(
outcome="harvest",
resolved_ids=animals_to_harvest,
yield_items=yield_items,
)
services["animal_service"].record_outcome(outcome_payload, now_utc, "test_user")
# Count after harvest
count_after = seeded_db.execute(
"""SELECT COUNT(*) FROM live_animals_by_location
WHERE location_id = ? AND sex = 'female'""",
(strip2_id,),
).fetchone()[0]
assert count_after == 3
def test_harvest_yields_present_in_event(self, seeded_db, services, now_utc, harvest_scenario):
"""Yields should be present in the event payload for history/export."""
strip2_animal_ids = harvest_scenario["strip2_animal_ids"]
animals_to_harvest = strip2_animal_ids[:2]
yield_items = [
{
"product_code": "meat.part.breast.duck",
"unit": "kg",
"quantity": 2,
"weight_kg": 1.4,
},
{"product_code": "fat.rendered.duck", "unit": "kg", "quantity": 1, "weight_kg": 0.3},
]
outcome_payload = AnimalOutcomePayload(
outcome="harvest",
resolved_ids=animals_to_harvest,
yield_items=yield_items,
)
event = services["animal_service"].record_outcome(outcome_payload, now_utc, "test_user")
# Verify yields are in payload
assert "yield_items" in event.payload
assert len(event.payload["yield_items"]) == 2
# Verify yield details
yields = event.payload["yield_items"]
meat_yield = next(y for y in yields if y["product_code"] == "meat.part.breast.duck")
assert meat_yield["quantity"] == 2
assert abs(meat_yield["weight_kg"] - 1.4) < 0.001
fat_yield = next(y for y in yields if y["product_code"] == "fat.rendered.duck")
assert fat_yield["quantity"] == 1
assert abs(fat_yield["weight_kg"] - 0.3) < 0.001
def test_harvest_egg_stats_unchanged(self, seeded_db, services, now_utc, harvest_scenario):
"""EggStats should remain unchanged after harvest.
Harvest yields are stored in the event payload, not as collected products.
The PRODUCT_COLLECTED event type should not be created by harvest.
"""
from animaltrack.events.types import PRODUCT_COLLECTED
strip2_animal_ids = harvest_scenario["strip2_animal_ids"]
animals_to_harvest = strip2_animal_ids[:2]
# Count PRODUCT_COLLECTED events before harvest
events_before = seeded_db.execute(
"SELECT COUNT(*) FROM events WHERE type = ?",
(PRODUCT_COLLECTED,),
).fetchone()[0]
yield_items = [
{
"product_code": "meat.part.breast.duck",
"unit": "kg",
"quantity": 2,
"weight_kg": 1.4,
},
{"product_code": "fat.rendered.duck", "unit": "kg", "quantity": 1, "weight_kg": 0.3},
]
outcome_payload = AnimalOutcomePayload(
outcome="harvest",
resolved_ids=animals_to_harvest,
yield_items=yield_items,
)
services["animal_service"].record_outcome(outcome_payload, now_utc, "test_user")
# Count PRODUCT_COLLECTED events after harvest
events_after = seeded_db.execute(
"SELECT COUNT(*) FROM events WHERE type = ?",
(PRODUCT_COLLECTED,),
).fetchone()[0]
# Verify no new PRODUCT_COLLECTED events were created
# (yields are in ANIMAL_OUTCOME payload, not separate PRODUCT_COLLECTED events)
assert events_before == events_after
def test_harvest_other_animals_unaffected(self, seeded_db, services, now_utc, harvest_scenario):
"""Animals not harvested should remain unaffected."""
strip1_id = harvest_scenario["strip1_id"]
strip1_animal_ids = harvest_scenario["strip1_animal_ids"]
strip2_animal_ids = harvest_scenario["strip2_animal_ids"]
animals_to_harvest = strip2_animal_ids[:2]
remaining_strip2_animals = strip2_animal_ids[2:]
yield_items = [
{
"product_code": "meat.part.breast.duck",
"unit": "kg",
"quantity": 2,
"weight_kg": 1.4,
},
{"product_code": "fat.rendered.duck", "unit": "kg", "quantity": 1, "weight_kg": 0.3},
]
outcome_payload = AnimalOutcomePayload(
outcome="harvest",
resolved_ids=animals_to_harvest,
yield_items=yield_items,
)
services["animal_service"].record_outcome(outcome_payload, now_utc, "test_user")
# Verify Strip 1 animals still alive (5)
strip1_count = seeded_db.execute(
"""SELECT COUNT(*) FROM live_animals_by_location
WHERE location_id = ?""",
(strip1_id,),
).fetchone()[0]
assert strip1_count == 5
# Verify remaining Strip 2 animals still alive (3)
for animal_id in remaining_strip2_animals:
row = seeded_db.execute(
"SELECT status FROM animal_registry WHERE animal_id = ?",
(animal_id,),
).fetchone()
assert row[0] == "alive"
# Verify Strip 1 animals all alive
for animal_id in strip1_animal_ids:
row = seeded_db.execute(
"SELECT status FROM animal_registry WHERE animal_id = ?",
(animal_id,),
).fetchone()
assert row[0] == "alive"