# ABOUTME: Tests for Egg Quick Capture web routes. # ABOUTME: Covers GET / form rendering and POST /actions/product-collected. import os import time import pytest from starlette.testclient import TestClient 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.projections.products import ProductsProjection from animaltrack.services.animal import AnimalService def make_test_settings( csrf_secret: str = "test-secret", trusted_proxy_ips: str = "127.0.0.1", dev_mode: bool = True, ): """Create Settings for testing by setting env vars temporarily.""" from animaltrack.config import Settings old_env = os.environ.copy() try: os.environ["CSRF_SECRET"] = csrf_secret os.environ["TRUSTED_PROXY_IPS"] = trusted_proxy_ips os.environ["DEV_MODE"] = str(dev_mode).lower() return Settings() finally: os.environ.clear() os.environ.update(old_env) @pytest.fixture def client(seeded_db): """Create a test client for the app.""" from animaltrack.web.app import create_app settings = make_test_settings(trusted_proxy_ips="testclient") app, rt = create_app(settings=settings, db=seeded_db) # Use raise_server_exceptions=True to see actual errors return TestClient(app, raise_server_exceptions=True) @pytest.fixture def location_strip1_id(seeded_db): """Get Strip 1 location ID from seeded data.""" row = seeded_db.execute("SELECT id FROM locations WHERE name = 'Strip 1'").fetchone() return row[0] @pytest.fixture def location_nursery1_id(seeded_db): """Get Nursery 1 location ID from seeded data (no ducks here).""" row = seeded_db.execute("SELECT id FROM locations WHERE name = 'Nursery 1'").fetchone() return row[0] @pytest.fixture def ducks_at_strip1(seeded_db, location_strip1_id): """Create ducks at Strip 1 for testing egg collection.""" event_store = EventStore(seeded_db) registry = ProjectionRegistry() registry.register(AnimalRegistryProjection(seeded_db)) registry.register(EventAnimalsProjection(seeded_db)) registry.register(IntervalProjection(seeded_db)) registry.register(ProductsProjection(seeded_db)) animal_service = AnimalService(seeded_db, event_store, registry) payload = AnimalCohortCreatedPayload( species="duck", count=5, life_stage="adult", sex="female", location_id=location_strip1_id, origin="purchased", ) ts_utc = int(time.time() * 1000) event = animal_service.create_cohort(payload, ts_utc, "test_user") return event.entity_refs["animal_ids"] class TestEggFormRendering: """Tests for GET / egg capture form.""" def test_egg_form_renders(self, client): """GET / returns 200 with form elements.""" resp = client.get("/") assert resp.status_code == 200 assert "Record Eggs" in resp.text or "Egg" in resp.text def test_egg_form_shows_locations(self, client): """Form has location dropdown with seeded locations.""" resp = client.get("/") assert resp.status_code == 200 # Check for seeded location names in the response assert "Strip 1" in resp.text assert "Strip 2" in resp.text def test_egg_form_has_quantity_field(self, client): """Form has quantity input field.""" resp = client.get("/") assert resp.status_code == 200 assert 'name="quantity"' in resp.text or 'id="quantity"' in resp.text class TestEggCollection: """Tests for POST /actions/product-collected.""" def test_egg_collection_creates_event( self, client, seeded_db, location_strip1_id, ducks_at_strip1 ): """POST creates ProductCollected event when ducks exist at location.""" resp = client.post( "/actions/product-collected", data={ "location_id": location_strip1_id, "quantity": "12", "notes": "Morning collection", "nonce": "test-nonce-123", }, ) # Should succeed (200 or redirect) assert resp.status_code in [200, 302, 303] # Verify event was created in database event_row = seeded_db.execute( "SELECT type, payload FROM events WHERE type = 'ProductCollected' ORDER BY id DESC LIMIT 1" ).fetchone() assert event_row is not None assert event_row[0] == "ProductCollected" def test_egg_collection_quantity_zero_accepted( self, client, seeded_db, location_strip1_id, ducks_at_strip1 ): """quantity=0 is accepted (checked coop, found no eggs).""" resp = client.post( "/actions/product-collected", data={ "location_id": location_strip1_id, "quantity": "0", "nonce": "test-nonce-456", }, ) assert resp.status_code in [200, 302, 303] # Verify event was created with quantity=0 event_row = seeded_db.execute( "SELECT payload FROM events WHERE type = 'ProductCollected' ORDER BY id DESC LIMIT 1" ).fetchone() assert event_row is not None import json payload = json.loads(event_row[0]) assert payload["quantity"] == 0 def test_egg_collection_validation_quantity_negative( self, client, location_strip1_id, ducks_at_strip1 ): """quantity=-1 returns 422.""" resp = client.post( "/actions/product-collected", data={ "location_id": location_strip1_id, "quantity": "-1", "nonce": "test-nonce-789", }, ) assert resp.status_code == 422 def test_egg_collection_validation_location_missing(self, client, ducks_at_strip1): """Missing location returns 422.""" resp = client.post( "/actions/product-collected", data={ "quantity": "12", "nonce": "test-nonce-abc", }, ) assert resp.status_code == 422 def test_egg_collection_no_ducks_at_location(self, client, location_nursery1_id): """POST to location with no ducks returns 422.""" resp = client.post( "/actions/product-collected", data={ "location_id": location_nursery1_id, "quantity": "12", "nonce": "test-nonce-def", }, ) assert resp.status_code == 422 # Error message should indicate no ducks assert "duck" in resp.text.lower() or "animal" in resp.text.lower() def test_egg_collection_location_sticks( self, client, seeded_db, location_strip1_id, ducks_at_strip1 ): """After successful POST, returned form shows same location selected.""" resp = client.post( "/actions/product-collected", data={ "location_id": location_strip1_id, "quantity": "6", "nonce": "test-nonce-ghi", }, ) assert resp.status_code == 200 # The response should contain the form with the location pre-selected # Check for "selected" attribute on the option with our location_id assert "selected" in resp.text and location_strip1_id in resp.text class TestEggsRecentEvents: """Tests for recent events display on eggs page.""" def test_harvest_tab_shows_recent_events_section(self, client): """Harvest tab shows Recent Harvests section.""" resp = client.get("/") assert resp.status_code == 200 assert "Recent Harvests" in resp.text def test_sell_tab_shows_recent_events_section(self, client): """Sell tab shows Recent Sales section.""" resp = client.get("/?tab=sell") assert resp.status_code == 200 assert "Recent Sales" in resp.text def test_harvest_event_appears_in_recent( self, client, seeded_db, location_strip1_id, ducks_at_strip1 ): """Newly created harvest event appears in recent events list.""" resp = client.post( "/actions/product-collected", data={ "location_id": location_strip1_id, "quantity": "12", "nonce": "test-nonce-recent-1", }, ) assert resp.status_code == 200 # Recent events should include the newly created event # Check for event link pattern assert "/events/" in resp.text def test_harvest_event_links_to_detail( self, client, seeded_db, location_strip1_id, ducks_at_strip1 ): """Harvest events in recent list link to event detail page.""" # Create an event resp = client.post( "/actions/product-collected", data={ "location_id": location_strip1_id, "quantity": "8", "nonce": "test-nonce-recent-2", }, ) assert resp.status_code == 200 # Get the event ID from DB event_row = seeded_db.execute( "SELECT id FROM events WHERE type = 'ProductCollected' ORDER BY id DESC LIMIT 1" ).fetchone() event_id = event_row[0] # The response should contain a link to the event detail assert f"/events/{event_id}" in resp.text class TestEggCollectionAnimalFiltering: """Tests that egg collection only associates adult females.""" def test_egg_collection_excludes_males_and_juveniles( self, client, seeded_db, location_strip1_id ): """Egg collection only associates adult female ducks, not males or juveniles.""" # Setup: Create mixed animals at location event_store = EventStore(seeded_db) registry = ProjectionRegistry() registry.register(AnimalRegistryProjection(seeded_db)) registry.register(EventAnimalsProjection(seeded_db)) registry.register(IntervalProjection(seeded_db)) registry.register(ProductsProjection(seeded_db)) animal_service = AnimalService(seeded_db, event_store, registry) ts_utc = int(time.time() * 1000) # Create adult female (should be included) female_payload = AnimalCohortCreatedPayload( species="duck", count=1, life_stage="adult", sex="female", location_id=location_strip1_id, origin="purchased", ) female_event = animal_service.create_cohort(female_payload, ts_utc, "test_user") female_id = female_event.entity_refs["animal_ids"][0] # Create adult male (should be excluded) male_payload = AnimalCohortCreatedPayload( species="duck", count=1, life_stage="adult", sex="male", location_id=location_strip1_id, origin="purchased", ) male_event = animal_service.create_cohort(male_payload, ts_utc, "test_user") male_id = male_event.entity_refs["animal_ids"][0] # Create juvenile female (should be excluded) juvenile_payload = AnimalCohortCreatedPayload( species="duck", count=1, life_stage="juvenile", sex="female", location_id=location_strip1_id, origin="purchased", ) juvenile_event = animal_service.create_cohort(juvenile_payload, ts_utc, "test_user") juvenile_id = juvenile_event.entity_refs["animal_ids"][0] # Collect eggs resp = client.post( "/actions/product-collected", data={ "location_id": location_strip1_id, "quantity": "6", "nonce": "test-nonce-filter", }, ) assert resp.status_code == 200 # Get the egg collection event event_row = seeded_db.execute( "SELECT id FROM events WHERE type = 'ProductCollected' ORDER BY id DESC LIMIT 1" ).fetchone() event_id = event_row[0] # Check which animals are associated with the event animal_rows = seeded_db.execute( "SELECT animal_id FROM event_animals WHERE event_id = ?", (event_id,), ).fetchall() associated_ids = {row[0] for row in animal_rows} # Only the adult female should be associated assert female_id in associated_ids, "Adult female should be associated with egg collection" assert male_id not in associated_ids, "Male should NOT be associated with egg collection" assert juvenile_id not in associated_ids, ( "Juvenile should NOT be associated with egg collection" ) assert len(associated_ids) == 1, "Only adult females should be associated"