# 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"