# ABOUTME: Tests for Animal Action web routes. # ABOUTME: Covers cohort creation, hatch recording, and other animal actions. import os import pytest from starlette.testclient import TestClient 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) 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.""" row = seeded_db.execute("SELECT id FROM locations WHERE name = 'Nursery 1'").fetchone() return row[0] # ============================================================================= # Cohort Creation Tests # ============================================================================= class TestCohortFormRendering: """Tests for GET /actions/cohort form rendering.""" def test_cohort_form_renders(self, client): """GET /actions/cohort returns 200 with form elements.""" resp = client.get("/actions/cohort") assert resp.status_code == 200 assert "Create Animal Cohort" in resp.text or "Cohort" in resp.text def test_cohort_form_has_species_dropdown(self, client): """Form has species dropdown.""" resp = client.get("/actions/cohort") assert resp.status_code == 200 assert 'name="species"' in resp.text def test_cohort_form_has_location_dropdown(self, client): """Form has location dropdown with seeded locations.""" resp = client.get("/actions/cohort") assert resp.status_code == 200 assert "Strip 1" in resp.text assert 'name="location_id"' in resp.text def test_cohort_form_has_count_field(self, client): """Form has count input field.""" resp = client.get("/actions/cohort") assert resp.status_code == 200 assert 'name="count"' in resp.text def test_cohort_form_has_life_stage_dropdown(self, client): """Form has life stage dropdown.""" resp = client.get("/actions/cohort") assert resp.status_code == 200 assert 'name="life_stage"' in resp.text def test_cohort_form_has_origin_dropdown(self, client): """Form has origin dropdown.""" resp = client.get("/actions/cohort") assert resp.status_code == 200 assert 'name="origin"' in resp.text class TestCohortCreationSuccess: """Tests for successful POST /actions/animal-cohort.""" def test_cohort_creates_event(self, client, seeded_db, location_strip1_id): """POST creates AnimalCohortCreated event when valid.""" resp = client.post( "/actions/animal-cohort", data={ "species": "duck", "location_id": location_strip1_id, "count": "5", "life_stage": "adult", "sex": "female", "origin": "purchased", "nonce": "test-cohort-nonce-1", }, ) assert resp.status_code == 200 # Verify event was created event_row = seeded_db.execute( "SELECT type FROM events WHERE type = 'AnimalCohortCreated' ORDER BY id DESC LIMIT 1" ).fetchone() assert event_row is not None assert event_row[0] == "AnimalCohortCreated" def test_cohort_creates_animals(self, client, seeded_db, location_strip1_id): """POST creates the correct number of animals in registry.""" # Count animals before count_before = seeded_db.execute("SELECT COUNT(*) FROM animal_registry").fetchone()[0] resp = client.post( "/actions/animal-cohort", data={ "species": "duck", "location_id": location_strip1_id, "count": "3", "life_stage": "juvenile", "sex": "unknown", "origin": "hatched", "nonce": "test-cohort-nonce-2", }, ) assert resp.status_code == 200 # Count animals after count_after = seeded_db.execute("SELECT COUNT(*) FROM animal_registry").fetchone()[0] assert count_after == count_before + 3 def test_cohort_success_returns_toast(self, client, seeded_db, location_strip1_id): """Successful cohort creation returns HX-Trigger with toast.""" resp = client.post( "/actions/animal-cohort", data={ "species": "duck", "location_id": location_strip1_id, "count": "2", "life_stage": "adult", "sex": "male", "origin": "purchased", "nonce": "test-cohort-nonce-3", }, ) assert resp.status_code == 200 assert "HX-Trigger" in resp.headers assert "showToast" in resp.headers["HX-Trigger"] class TestCohortCreationValidation: """Tests for validation errors in POST /actions/animal-cohort.""" def test_cohort_missing_species_returns_422(self, client, location_strip1_id): """Missing species returns 422.""" resp = client.post( "/actions/animal-cohort", data={ # Missing species "location_id": location_strip1_id, "count": "5", "life_stage": "adult", "origin": "purchased", "nonce": "test-cohort-nonce-4", }, ) assert resp.status_code == 422 def test_cohort_missing_location_returns_422(self, client): """Missing location returns 422.""" resp = client.post( "/actions/animal-cohort", data={ "species": "duck", # Missing location_id "count": "5", "life_stage": "adult", "origin": "purchased", "nonce": "test-cohort-nonce-5", }, ) assert resp.status_code == 422 def test_cohort_invalid_count_returns_422(self, client, location_strip1_id): """Invalid count (0 or negative) returns 422.""" resp = client.post( "/actions/animal-cohort", data={ "species": "duck", "location_id": location_strip1_id, "count": "0", # Invalid "life_stage": "adult", "origin": "purchased", "nonce": "test-cohort-nonce-6", }, ) assert resp.status_code == 422 def test_cohort_missing_life_stage_returns_422(self, client, location_strip1_id): """Missing life stage returns 422.""" resp = client.post( "/actions/animal-cohort", data={ "species": "duck", "location_id": location_strip1_id, "count": "5", # Missing life_stage "origin": "purchased", "nonce": "test-cohort-nonce-7", }, ) assert resp.status_code == 422 def test_cohort_missing_origin_returns_422(self, client, location_strip1_id): """Missing origin returns 422.""" resp = client.post( "/actions/animal-cohort", data={ "species": "duck", "location_id": location_strip1_id, "count": "5", "life_stage": "adult", # Missing origin "nonce": "test-cohort-nonce-8", }, ) assert resp.status_code == 422 # ============================================================================= # Hatch Recording Tests # ============================================================================= class TestHatchFormRendering: """Tests for GET /actions/hatch form rendering.""" def test_hatch_form_renders(self, client): """GET /actions/hatch returns 200 with form elements.""" resp = client.get("/actions/hatch") assert resp.status_code == 200 assert "Record Hatch" in resp.text or "Hatch" in resp.text def test_hatch_form_has_species_dropdown(self, client): """Form has species dropdown.""" resp = client.get("/actions/hatch") assert resp.status_code == 200 assert 'name="species"' in resp.text def test_hatch_form_has_location_dropdown(self, client): """Form has location dropdown.""" resp = client.get("/actions/hatch") assert resp.status_code == 200 assert 'name="location_id"' in resp.text def test_hatch_form_has_hatched_live_field(self, client): """Form has hatched live count field.""" resp = client.get("/actions/hatch") assert resp.status_code == 200 assert 'name="hatched_live"' in resp.text def test_hatch_form_has_brood_location_dropdown(self, client): """Form has optional brood location dropdown.""" resp = client.get("/actions/hatch") assert resp.status_code == 200 assert 'name="assigned_brood_location_id"' in resp.text class TestHatchRecordingSuccess: """Tests for successful POST /actions/hatch-recorded.""" def test_hatch_creates_event(self, client, seeded_db, location_strip1_id): """POST creates HatchRecorded event when valid.""" resp = client.post( "/actions/hatch-recorded", data={ "species": "duck", "location_id": location_strip1_id, "hatched_live": "4", "nonce": "test-hatch-nonce-1", }, ) assert resp.status_code == 200 # Verify event was created event_row = seeded_db.execute( "SELECT type FROM events WHERE type = 'HatchRecorded' ORDER BY id DESC LIMIT 1" ).fetchone() assert event_row is not None assert event_row[0] == "HatchRecorded" def test_hatch_creates_hatchlings(self, client, seeded_db, location_strip1_id): """POST creates the correct number of hatchling animals.""" # Count animals before count_before = seeded_db.execute("SELECT COUNT(*) FROM animal_registry").fetchone()[0] resp = client.post( "/actions/hatch-recorded", data={ "species": "duck", "location_id": location_strip1_id, "hatched_live": "6", "nonce": "test-hatch-nonce-2", }, ) assert resp.status_code == 200 # Count animals after count_after = seeded_db.execute("SELECT COUNT(*) FROM animal_registry").fetchone()[0] assert count_after == count_before + 6 def test_hatch_with_brood_location( self, client, seeded_db, location_strip1_id, location_nursery1_id ): """POST with brood location places hatchlings at brood location.""" resp = client.post( "/actions/hatch-recorded", data={ "species": "duck", "location_id": location_strip1_id, "assigned_brood_location_id": location_nursery1_id, "hatched_live": "3", "nonce": "test-hatch-nonce-3", }, ) assert resp.status_code == 200 # Verify hatchlings are at brood location count_at_nursery = seeded_db.execute( "SELECT COUNT(*) FROM animal_registry WHERE location_id = ? AND life_stage = 'hatchling'", (location_nursery1_id,), ).fetchone()[0] assert count_at_nursery >= 3 def test_hatch_success_returns_toast(self, client, seeded_db, location_strip1_id): """Successful hatch recording returns HX-Trigger with toast.""" resp = client.post( "/actions/hatch-recorded", data={ "species": "duck", "location_id": location_strip1_id, "hatched_live": "2", "nonce": "test-hatch-nonce-4", }, ) assert resp.status_code == 200 assert "HX-Trigger" in resp.headers assert "showToast" in resp.headers["HX-Trigger"] class TestHatchRecordingValidation: """Tests for validation errors in POST /actions/hatch-recorded.""" def test_hatch_missing_species_returns_422(self, client, location_strip1_id): """Missing species returns 422.""" resp = client.post( "/actions/hatch-recorded", data={ # Missing species "location_id": location_strip1_id, "hatched_live": "4", "nonce": "test-hatch-nonce-5", }, ) assert resp.status_code == 422 def test_hatch_missing_location_returns_422(self, client): """Missing location returns 422.""" resp = client.post( "/actions/hatch-recorded", data={ "species": "duck", # Missing location_id "hatched_live": "4", "nonce": "test-hatch-nonce-6", }, ) assert resp.status_code == 422 def test_hatch_invalid_count_returns_422(self, client, location_strip1_id): """Invalid count (0 or negative) returns 422.""" resp = client.post( "/actions/hatch-recorded", data={ "species": "duck", "location_id": location_strip1_id, "hatched_live": "0", # Invalid "nonce": "test-hatch-nonce-7", }, ) assert resp.status_code == 422 # ============================================================================= # Animal Promotion Tests # ============================================================================= @pytest.fixture def unpromoted_animal_id(seeded_db, client, location_strip1_id): """Create an unpromoted animal for testing.""" # Create a cohort with one animal resp = client.post( "/actions/animal-cohort", data={ "species": "duck", "location_id": location_strip1_id, "count": "1", "life_stage": "adult", "sex": "female", "origin": "purchased", "nonce": "test-fixture-cohort-1", }, ) assert resp.status_code == 200 # Get the animal ID row = seeded_db.execute( "SELECT animal_id FROM animal_registry WHERE identified = 0 ORDER BY animal_id DESC LIMIT 1" ).fetchone() return row[0] class TestPromoteFormRendering: """Tests for GET /actions/promote/{animal_id} form rendering.""" def test_promote_form_renders(self, client, unpromoted_animal_id): """GET /actions/promote/{animal_id} returns 200 with form elements.""" resp = client.get(f"/actions/promote/{unpromoted_animal_id}") assert resp.status_code == 200 assert "Promote" in resp.text def test_promote_form_shows_animal_info(self, client, unpromoted_animal_id): """Form shows animal information.""" resp = client.get(f"/actions/promote/{unpromoted_animal_id}") assert resp.status_code == 200 # Should show animal ID (truncated) assert unpromoted_animal_id[:8] in resp.text def test_promote_form_has_nickname_field(self, client, unpromoted_animal_id): """Form has nickname input field.""" resp = client.get(f"/actions/promote/{unpromoted_animal_id}") assert resp.status_code == 200 assert 'name="nickname"' in resp.text def test_promote_form_has_sex_dropdown(self, client, unpromoted_animal_id): """Form has sex dropdown.""" resp = client.get(f"/actions/promote/{unpromoted_animal_id}") assert resp.status_code == 200 assert 'name="sex"' in resp.text def test_promote_form_not_found(self, client): """Non-existent animal returns 404.""" resp = client.get("/actions/promote/01ABCDEFGHIJKLMNOPQRSTUV00") assert resp.status_code == 404 class TestPromoteSuccess: """Tests for successful POST /actions/animal-promote.""" def test_promote_creates_event(self, client, seeded_db, unpromoted_animal_id): """POST creates AnimalPromoted event when valid.""" resp = client.post( "/actions/animal-promote", data={ "animal_id": unpromoted_animal_id, "nickname": "Daffy", "nonce": "test-promote-nonce-1", }, follow_redirects=False, ) # Should redirect assert resp.status_code == 303 # Verify event was created event_row = seeded_db.execute( "SELECT type FROM events WHERE type = 'AnimalPromoted' ORDER BY id DESC LIMIT 1" ).fetchone() assert event_row is not None assert event_row[0] == "AnimalPromoted" def test_promote_sets_identified(self, client, seeded_db, unpromoted_animal_id): """POST sets identified flag on animal.""" resp = client.post( "/actions/animal-promote", data={ "animal_id": unpromoted_animal_id, "nickname": "Huey", "nonce": "test-promote-nonce-2", }, follow_redirects=False, ) assert resp.status_code == 303 # Verify identified is now true row = seeded_db.execute( "SELECT identified FROM animal_registry WHERE animal_id = ?", (unpromoted_animal_id,), ).fetchone() assert row[0] == 1 # identified = true def test_promote_sets_nickname(self, client, seeded_db, unpromoted_animal_id): """POST sets nickname on animal.""" resp = client.post( "/actions/animal-promote", data={ "animal_id": unpromoted_animal_id, "nickname": "Dewey", "nonce": "test-promote-nonce-3", }, follow_redirects=False, ) assert resp.status_code == 303 # Verify nickname is set row = seeded_db.execute( "SELECT nickname FROM animal_registry WHERE animal_id = ?", (unpromoted_animal_id,), ).fetchone() assert row[0] == "Dewey" class TestPromoteValidation: """Tests for validation errors in POST /actions/animal-promote.""" def test_promote_missing_animal_id_returns_422(self, client): """Missing animal_id returns 422.""" resp = client.post( "/actions/animal-promote", data={ "nickname": "Test", "nonce": "test-promote-nonce-4", }, ) assert resp.status_code == 422 def test_promote_not_found_returns_404(self, client): """Non-existent animal returns 404.""" resp = client.post( "/actions/animal-promote", data={ "animal_id": "01ABCDEFGHIJKLMNOPQRSTUV00", "nickname": "Test", "nonce": "test-promote-nonce-5", }, ) assert resp.status_code == 404 # ============================================================================= # Add Tag Tests # ============================================================================= @pytest.fixture def animals_for_tagging(seeded_db, client, location_strip1_id): """Create animals for tag testing and return their IDs.""" # Create a cohort with 3 animals resp = client.post( "/actions/animal-cohort", data={ "species": "duck", "location_id": location_strip1_id, "count": "3", "life_stage": "adult", "sex": "female", "origin": "purchased", "nonce": "test-tag-fixture-cohort-1", }, ) assert resp.status_code == 200 # Get the animal IDs rows = seeded_db.execute( "SELECT animal_id FROM animal_registry WHERE origin = 'purchased' AND life_stage = 'adult' ORDER BY animal_id DESC LIMIT 3" ).fetchall() return [row[0] for row in rows] class TestTagAddFormRendering: """Tests for GET /actions/tag-add form rendering.""" def test_tag_add_form_renders(self, client): """GET /actions/tag-add returns 200 with form elements.""" resp = client.get("/actions/tag-add") assert resp.status_code == 200 assert "Add Tag" in resp.text def test_tag_add_form_has_filter_field(self, client): """Form has filter input field.""" resp = client.get("/actions/tag-add") assert resp.status_code == 200 assert 'name="filter"' in resp.text def test_tag_add_form_has_tag_field(self, client): """Form has tag input field.""" resp = client.get("/actions/tag-add") assert resp.status_code == 200 assert 'name="tag"' in resp.text def test_tag_add_form_with_filter_shows_selection(self, client, animals_for_tagging): """Form with filter shows selection preview.""" # Use species filter which is a valid filter field resp = client.get("/actions/tag-add?filter=species:duck") assert resp.status_code == 200 # Should show selection count assert "selected" in resp.text.lower() class TestTagAddSuccess: """Tests for successful POST /actions/animal-tag-add.""" def test_tag_add_creates_event(self, client, seeded_db, animals_for_tagging): """POST creates AnimalTagged event when valid.""" import time from animaltrack.selection import compute_roster_hash roster_hash = compute_roster_hash(animals_for_tagging, None) ts_utc = int(time.time() * 1000) resp = client.post( "/actions/animal-tag-add", data={ "filter": "species:duck", "tag": "test-tag", "resolved_ids": animals_for_tagging, "roster_hash": roster_hash, "ts_utc": str(ts_utc), "nonce": "test-tag-nonce-1", }, ) assert resp.status_code == 200 # Verify event was created event_row = seeded_db.execute( "SELECT type FROM events WHERE type = 'AnimalTagged' ORDER BY id DESC LIMIT 1" ).fetchone() assert event_row is not None assert event_row[0] == "AnimalTagged" def test_tag_add_creates_tag_intervals(self, client, seeded_db, animals_for_tagging): """POST creates tag intervals for animals.""" import time from animaltrack.selection import compute_roster_hash roster_hash = compute_roster_hash(animals_for_tagging, None) ts_utc = int(time.time() * 1000) resp = client.post( "/actions/animal-tag-add", data={ "filter": "species:duck", "tag": "layer-birds", "resolved_ids": animals_for_tagging, "roster_hash": roster_hash, "ts_utc": str(ts_utc), "nonce": "test-tag-nonce-2", }, ) assert resp.status_code == 200 # Verify tag intervals were created tag_count = seeded_db.execute( "SELECT COUNT(*) FROM animal_tag_intervals WHERE tag = 'layer-birds' AND end_utc IS NULL" ).fetchone()[0] assert tag_count >= len(animals_for_tagging) def test_tag_add_success_returns_toast(self, client, seeded_db, animals_for_tagging): """Successful tag add returns HX-Trigger with toast.""" import time from animaltrack.selection import compute_roster_hash roster_hash = compute_roster_hash(animals_for_tagging, None) ts_utc = int(time.time() * 1000) resp = client.post( "/actions/animal-tag-add", data={ "filter": "species:duck", "tag": "test-tag-toast", "resolved_ids": animals_for_tagging, "roster_hash": roster_hash, "ts_utc": str(ts_utc), "nonce": "test-tag-nonce-3", }, ) assert resp.status_code == 200 assert "HX-Trigger" in resp.headers assert "showToast" in resp.headers["HX-Trigger"] class TestTagAddValidation: """Tests for validation errors in POST /actions/animal-tag-add.""" def test_tag_add_missing_tag_returns_422(self, client, animals_for_tagging): """Missing tag returns 422.""" import time from animaltrack.selection import compute_roster_hash roster_hash = compute_roster_hash(animals_for_tagging, None) ts_utc = int(time.time() * 1000) resp = client.post( "/actions/animal-tag-add", data={ "filter": "species:duck", # Missing tag "resolved_ids": animals_for_tagging, "roster_hash": roster_hash, "ts_utc": str(ts_utc), "nonce": "test-tag-nonce-4", }, ) assert resp.status_code == 422 def test_tag_add_no_animals_returns_422(self, client): """No animals selected returns 422.""" import time ts_utc = int(time.time() * 1000) resp = client.post( "/actions/animal-tag-add", data={ "filter": "", "tag": "test-tag", # No resolved_ids "roster_hash": "", "ts_utc": str(ts_utc), "nonce": "test-tag-nonce-5", }, ) assert resp.status_code == 422 # ============================================================================= # End Tag Tests # ============================================================================= @pytest.fixture def tagged_animals(seeded_db, client, animals_for_tagging): """Tag animals and return their IDs.""" import time from animaltrack.selection import compute_roster_hash roster_hash = compute_roster_hash(animals_for_tagging, None) ts_utc = int(time.time() * 1000) # First tag the animals resp = client.post( "/actions/animal-tag-add", data={ "filter": "species:duck", "tag": "test-end-tag", "resolved_ids": animals_for_tagging, "roster_hash": roster_hash, "ts_utc": str(ts_utc), "nonce": "test-fixture-tag-1", }, ) assert resp.status_code == 200 return animals_for_tagging class TestTagEndFormRendering: """Tests for GET /actions/tag-end form rendering.""" def test_tag_end_form_renders(self, client): """GET /actions/tag-end returns 200 with form elements.""" resp = client.get("/actions/tag-end") assert resp.status_code == 200 assert "End Tag" in resp.text def test_tag_end_form_has_filter_field(self, client): """Form has filter input field.""" resp = client.get("/actions/tag-end") assert resp.status_code == 200 assert 'name="filter"' in resp.text class TestTagEndSuccess: """Tests for successful POST /actions/animal-tag-end.""" def test_tag_end_creates_event(self, client, seeded_db, tagged_animals): """POST creates AnimalTagEnded event when valid.""" import time from animaltrack.selection import compute_roster_hash roster_hash = compute_roster_hash(tagged_animals, None) ts_utc = int(time.time() * 1000) resp = client.post( "/actions/animal-tag-end", data={ "filter": "species:duck", "tag": "test-end-tag", "resolved_ids": tagged_animals, "roster_hash": roster_hash, "ts_utc": str(ts_utc), "nonce": "test-tag-end-nonce-1", }, ) assert resp.status_code == 200 # Verify event was created event_row = seeded_db.execute( "SELECT type FROM events WHERE type = 'AnimalTagEnded' ORDER BY id DESC LIMIT 1" ).fetchone() assert event_row is not None assert event_row[0] == "AnimalTagEnded" def test_tag_end_closes_intervals(self, client, seeded_db, tagged_animals): """POST closes tag intervals for animals.""" import time from animaltrack.selection import compute_roster_hash roster_hash = compute_roster_hash(tagged_animals, None) ts_utc = int(time.time() * 1000) # Verify intervals are open before open_before = seeded_db.execute( "SELECT COUNT(*) FROM animal_tag_intervals WHERE tag = 'test-end-tag' AND end_utc IS NULL" ).fetchone()[0] assert open_before >= len(tagged_animals) resp = client.post( "/actions/animal-tag-end", data={ "filter": "species:duck", "tag": "test-end-tag", "resolved_ids": tagged_animals, "roster_hash": roster_hash, "ts_utc": str(ts_utc), "nonce": "test-tag-end-nonce-2", }, ) assert resp.status_code == 200 # Verify intervals are closed after open_after = seeded_db.execute( "SELECT COUNT(*) FROM animal_tag_intervals WHERE tag = 'test-end-tag' AND end_utc IS NULL" ).fetchone()[0] assert open_after == 0 def test_tag_end_success_returns_toast(self, client, seeded_db, tagged_animals): """Successful tag end returns HX-Trigger with toast.""" import time from animaltrack.selection import compute_roster_hash roster_hash = compute_roster_hash(tagged_animals, None) ts_utc = int(time.time() * 1000) resp = client.post( "/actions/animal-tag-end", data={ "filter": "species:duck", "tag": "test-end-tag", "resolved_ids": tagged_animals, "roster_hash": roster_hash, "ts_utc": str(ts_utc), "nonce": "test-tag-end-nonce-3", }, ) assert resp.status_code == 200 assert "HX-Trigger" in resp.headers assert "showToast" in resp.headers["HX-Trigger"] class TestTagEndValidation: """Tests for validation errors in POST /actions/animal-tag-end.""" def test_tag_end_missing_tag_returns_422(self, client, tagged_animals): """Missing tag returns 422.""" import time from animaltrack.selection import compute_roster_hash roster_hash = compute_roster_hash(tagged_animals, None) ts_utc = int(time.time() * 1000) resp = client.post( "/actions/animal-tag-end", data={ "filter": "species:duck", # Missing tag "resolved_ids": tagged_animals, "roster_hash": roster_hash, "ts_utc": str(ts_utc), "nonce": "test-tag-end-nonce-4", }, ) assert resp.status_code == 422 def test_tag_end_no_animals_returns_422(self, client): """No animals selected returns 422.""" import time ts_utc = int(time.time() * 1000) resp = client.post( "/actions/animal-tag-end", data={ "filter": "", "tag": "some-tag", # No resolved_ids "roster_hash": "", "ts_utc": str(ts_utc), "nonce": "test-tag-end-nonce-5", }, ) assert resp.status_code == 422 # ============================================================================= # Update Attributes Tests # ============================================================================= class TestAttrsFormRendering: """Tests for GET /actions/attrs form rendering.""" def test_attrs_form_renders(self, client): """GET /actions/attrs returns 200 with form elements.""" resp = client.get("/actions/attrs") assert resp.status_code == 200 assert "Update Attributes" in resp.text def test_attrs_form_has_filter_field(self, client): """Form has filter input field.""" resp = client.get("/actions/attrs") assert resp.status_code == 200 assert 'name="filter"' in resp.text def test_attrs_form_has_sex_dropdown(self, client): """Form has sex dropdown.""" resp = client.get("/actions/attrs") assert resp.status_code == 200 assert 'name="sex"' in resp.text def test_attrs_form_has_life_stage_dropdown(self, client): """Form has life_stage dropdown.""" resp = client.get("/actions/attrs") assert resp.status_code == 200 assert 'name="life_stage"' in resp.text class TestAttrsSuccess: """Tests for successful POST /actions/animal-attrs.""" def test_attrs_creates_event(self, client, seeded_db, animals_for_tagging): """POST creates AnimalAttributesUpdated event when valid.""" import time from animaltrack.selection import compute_roster_hash roster_hash = compute_roster_hash(animals_for_tagging, None) ts_utc = int(time.time() * 1000) resp = client.post( "/actions/animal-attrs", data={ "filter": "species:duck", "sex": "male", "resolved_ids": animals_for_tagging, "roster_hash": roster_hash, "ts_utc": str(ts_utc), "nonce": "test-attrs-nonce-1", }, ) assert resp.status_code == 200 # Verify event was created event_row = seeded_db.execute( "SELECT type FROM events WHERE type = 'AnimalAttributesUpdated' ORDER BY id DESC LIMIT 1" ).fetchone() assert event_row is not None assert event_row[0] == "AnimalAttributesUpdated" def test_attrs_updates_registry(self, client, seeded_db, animals_for_tagging): """POST updates animal_registry with new attributes.""" import time from animaltrack.selection import compute_roster_hash roster_hash = compute_roster_hash(animals_for_tagging, None) ts_utc = int(time.time() * 1000) resp = client.post( "/actions/animal-attrs", data={ "filter": "species:duck", "life_stage": "adult", "resolved_ids": animals_for_tagging, "roster_hash": roster_hash, "ts_utc": str(ts_utc), "nonce": "test-attrs-nonce-2", }, ) assert resp.status_code == 200 # Verify attribute was updated adult_count = seeded_db.execute( "SELECT COUNT(*) FROM animal_registry WHERE animal_id IN ({}) AND life_stage = 'adult'".format( ",".join("?" * len(animals_for_tagging)) ), animals_for_tagging, ).fetchone()[0] assert adult_count == len(animals_for_tagging) def test_attrs_success_returns_toast(self, client, seeded_db, animals_for_tagging): """Successful attrs update returns HX-Trigger with toast.""" import time from animaltrack.selection import compute_roster_hash roster_hash = compute_roster_hash(animals_for_tagging, None) ts_utc = int(time.time() * 1000) resp = client.post( "/actions/animal-attrs", data={ "filter": "species:duck", "repro_status": "intact", "resolved_ids": animals_for_tagging, "roster_hash": roster_hash, "ts_utc": str(ts_utc), "nonce": "test-attrs-nonce-3", }, ) assert resp.status_code == 200 assert "HX-Trigger" in resp.headers assert "showToast" in resp.headers["HX-Trigger"] class TestAttrsValidation: """Tests for validation errors in POST /actions/animal-attrs.""" def test_attrs_no_attribute_returns_422(self, client, animals_for_tagging): """No attribute selected returns 422.""" import time from animaltrack.selection import compute_roster_hash roster_hash = compute_roster_hash(animals_for_tagging, None) ts_utc = int(time.time() * 1000) resp = client.post( "/actions/animal-attrs", data={ "filter": "species:duck", # No sex, life_stage, or repro_status "resolved_ids": animals_for_tagging, "roster_hash": roster_hash, "ts_utc": str(ts_utc), "nonce": "test-attrs-nonce-4", }, ) assert resp.status_code == 422 def test_attrs_no_animals_returns_422(self, client): """No animals selected returns 422.""" import time ts_utc = int(time.time() * 1000) resp = client.post( "/actions/animal-attrs", data={ "filter": "", "sex": "male", # No resolved_ids "roster_hash": "", "ts_utc": str(ts_utc), "nonce": "test-attrs-nonce-5", }, ) assert resp.status_code == 422 # ============================================================================= # Record Outcome Form Tests # ============================================================================= class TestOutcomeFormRendering: """Tests for GET /actions/outcome form rendering.""" def test_outcome_form_renders(self, client): """Outcome form page renders with 200.""" resp = client.get("/actions/outcome") assert resp.status_code == 200 assert "Record Outcome" in resp.text def test_outcome_form_has_filter_field(self, client): """Outcome form has filter input field.""" resp = client.get("/actions/outcome") assert 'name="filter"' in resp.text def test_outcome_form_has_outcome_dropdown(self, client): """Outcome form has outcome dropdown.""" resp = client.get("/actions/outcome") assert 'name="outcome"' in resp.text assert "death" in resp.text.lower() assert "harvest" in resp.text.lower() def test_outcome_form_has_yield_section(self, client): """Outcome form has yield items section.""" resp = client.get("/actions/outcome") assert "Yield Items" in resp.text class TestOutcomeSuccess: """Tests for successful POST /actions/animal-outcome.""" def test_outcome_creates_event(self, client, seeded_db, animals_for_tagging): """Recording outcome creates AnimalOutcome event.""" import time from animaltrack.selection import compute_roster_hash roster_hash = compute_roster_hash(animals_for_tagging, None) ts_utc = int(time.time() * 1000) resp = client.post( "/actions/animal-outcome", data={ "filter": "species:duck", "outcome": "death", "reason": "old age", "resolved_ids": animals_for_tagging, "roster_hash": roster_hash, "ts_utc": str(ts_utc), "nonce": "test-outcome-nonce-1", }, ) assert resp.status_code == 200 # Verify event was created event_row = seeded_db.execute( "SELECT type FROM events WHERE type = 'AnimalOutcome'" ).fetchone() assert event_row is not None def test_outcome_updates_status(self, client, seeded_db, animals_for_tagging): """Recording outcome updates animal status in registry.""" import time from animaltrack.selection import compute_roster_hash roster_hash = compute_roster_hash(animals_for_tagging, None) ts_utc = int(time.time() * 1000) resp = client.post( "/actions/animal-outcome", data={ "filter": "species:duck", "outcome": "harvest", "resolved_ids": animals_for_tagging, "roster_hash": roster_hash, "ts_utc": str(ts_utc), "nonce": "test-outcome-nonce-2", }, ) assert resp.status_code == 200 # Verify status was updated harvested_count = seeded_db.execute( "SELECT COUNT(*) FROM animal_registry WHERE animal_id IN ({}) AND status = 'harvested'".format( ",".join("?" * len(animals_for_tagging)) ), animals_for_tagging, ).fetchone()[0] assert harvested_count == len(animals_for_tagging) def test_outcome_success_returns_toast(self, client, seeded_db, animals_for_tagging): """Successful outcome recording returns HX-Trigger with toast.""" import time from animaltrack.selection import compute_roster_hash roster_hash = compute_roster_hash(animals_for_tagging, None) ts_utc = int(time.time() * 1000) resp = client.post( "/actions/animal-outcome", data={ "filter": "species:duck", "outcome": "sold", "resolved_ids": animals_for_tagging, "roster_hash": roster_hash, "ts_utc": str(ts_utc), "nonce": "test-outcome-nonce-3", }, ) assert resp.status_code == 200 assert "HX-Trigger" in resp.headers assert "showToast" in resp.headers["HX-Trigger"] class TestOutcomeValidation: """Tests for validation errors in POST /actions/animal-outcome.""" def test_outcome_no_outcome_returns_422(self, client, animals_for_tagging): """No outcome selected returns 422.""" import time from animaltrack.selection import compute_roster_hash roster_hash = compute_roster_hash(animals_for_tagging, None) ts_utc = int(time.time() * 1000) resp = client.post( "/actions/animal-outcome", data={ "filter": "species:duck", # No outcome "resolved_ids": animals_for_tagging, "roster_hash": roster_hash, "ts_utc": str(ts_utc), "nonce": "test-outcome-nonce-4", }, ) assert resp.status_code == 422 def test_outcome_no_animals_returns_422(self, client): """No animals selected returns 422.""" import time ts_utc = int(time.time() * 1000) resp = client.post( "/actions/animal-outcome", data={ "filter": "", "outcome": "death", # No resolved_ids "roster_hash": "", "ts_utc": str(ts_utc), "nonce": "test-outcome-nonce-5", }, ) assert resp.status_code == 422 # ============================================================================= # Status Correct Form Tests (Admin-Only) # ============================================================================= @pytest.fixture def admin_client(seeded_db): """Test client with admin role (dev mode - bypasses CSRF, auto-admin auth).""" from animaltrack.web.app import create_app # Use dev_mode=True so CSRF is bypassed and auth is auto-admin settings = make_test_settings(trusted_proxy_ips="testclient", dev_mode=True) app, rt = create_app(settings=settings, db=seeded_db) return TestClient(app, raise_server_exceptions=True) @pytest.fixture def user_client(seeded_db): """Test client with regular (recorder) role.""" import time from animaltrack.models.reference import User, UserRole from animaltrack.repositories.users import UserRepository from animaltrack.web.app import create_app # Create recorder user in database user_repo = UserRepository(seeded_db) now = int(time.time() * 1000) recorder_user = User( username="recorder", role=UserRole.RECORDER, active=True, created_at_utc=now, updated_at_utc=now, ) user_repo.upsert(recorder_user) # Use dev_mode=False so auth_before checks the database settings = make_test_settings(trusted_proxy_ips="testclient", dev_mode=False) app, rt = create_app(settings=settings, db=seeded_db) # Create client that sends the auth header # raise_server_exceptions=False to get 403 instead of exception client = TestClient(app, raise_server_exceptions=False) client.headers["X-Oidc-Username"] = "recorder" return client class TestStatusCorrectFormRendering: """Tests for GET /actions/status-correct form rendering.""" def test_status_correct_form_renders_for_admin(self, admin_client): """Status correct form page renders for admin.""" resp = admin_client.get("/actions/status-correct") assert resp.status_code == 200 assert "Correct" in resp.text def test_status_correct_form_has_filter_field(self, admin_client): """Status correct form has filter input field.""" resp = admin_client.get("/actions/status-correct") assert 'name="filter"' in resp.text def test_status_correct_form_has_status_dropdown(self, admin_client): """Status correct form has status dropdown.""" resp = admin_client.get("/actions/status-correct") assert 'name="new_status"' in resp.text assert "alive" in resp.text.lower() def test_status_correct_form_requires_reason(self, admin_client): """Status correct form has required reason field.""" resp = admin_client.get("/actions/status-correct") assert 'name="reason"' in resp.text def test_status_correct_returns_403_for_user(self, user_client): """Status correct returns 403 for non-admin user.""" resp = user_client.get("/actions/status-correct") assert resp.status_code == 403 class TestStatusCorrectSuccess: """Tests for successful POST /actions/animal-status-correct.""" def test_status_correct_creates_event(self, admin_client, seeded_db, animals_for_tagging): """Correcting status creates AnimalStatusCorrected event.""" import time from animaltrack.selection import compute_roster_hash roster_hash = compute_roster_hash(animals_for_tagging, None) ts_utc = int(time.time() * 1000) resp = admin_client.post( "/actions/animal-status-correct", data={ "filter": "species:duck", "new_status": "dead", "reason": "Data entry error", "resolved_ids": animals_for_tagging, "roster_hash": roster_hash, "ts_utc": str(ts_utc), "nonce": "test-status-correct-nonce-1", }, ) assert resp.status_code == 200 # Verify event was created event_row = seeded_db.execute( "SELECT type FROM events WHERE type = 'AnimalStatusCorrected'" ).fetchone() assert event_row is not None def test_status_correct_updates_status(self, admin_client, seeded_db, animals_for_tagging): """Correcting status updates animal status in registry.""" import time from animaltrack.selection import compute_roster_hash roster_hash = compute_roster_hash(animals_for_tagging, None) ts_utc = int(time.time() * 1000) resp = admin_client.post( "/actions/animal-status-correct", data={ "filter": "species:duck", "new_status": "dead", "reason": "Mis-identified animal", "resolved_ids": animals_for_tagging, "roster_hash": roster_hash, "ts_utc": str(ts_utc), "nonce": "test-status-correct-nonce-2", }, ) assert resp.status_code == 200 # Verify status was updated dead_count = seeded_db.execute( "SELECT COUNT(*) FROM animal_registry WHERE animal_id IN ({}) AND status = 'dead'".format( ",".join("?" * len(animals_for_tagging)) ), animals_for_tagging, ).fetchone()[0] assert dead_count == len(animals_for_tagging) class TestStatusCorrectValidation: """Tests for validation errors in POST /actions/animal-status-correct.""" def test_status_correct_no_status_returns_422(self, admin_client, animals_for_tagging): """No status selected returns 422.""" import time from animaltrack.selection import compute_roster_hash roster_hash = compute_roster_hash(animals_for_tagging, None) ts_utc = int(time.time() * 1000) resp = admin_client.post( "/actions/animal-status-correct", data={ "filter": "species:duck", # No new_status "reason": "Some reason", "resolved_ids": animals_for_tagging, "roster_hash": roster_hash, "ts_utc": str(ts_utc), "nonce": "test-status-correct-nonce-3", }, ) assert resp.status_code == 422 def test_status_correct_no_reason_returns_422(self, admin_client, animals_for_tagging): """No reason provided returns 422.""" import time from animaltrack.selection import compute_roster_hash roster_hash = compute_roster_hash(animals_for_tagging, None) ts_utc = int(time.time() * 1000) resp = admin_client.post( "/actions/animal-status-correct", data={ "filter": "species:duck", "new_status": "dead", # No reason "resolved_ids": animals_for_tagging, "roster_hash": roster_hash, "ts_utc": str(ts_utc), "nonce": "test-status-correct-nonce-4", }, ) assert resp.status_code == 422 def test_status_correct_no_animals_returns_422(self, admin_client): """No animals selected returns 422.""" import time ts_utc = int(time.time() * 1000) resp = admin_client.post( "/actions/animal-status-correct", data={ "filter": "", "new_status": "dead", "reason": "Data entry error", # No resolved_ids "roster_hash": "", "ts_utc": str(ts_utc), "nonce": "test-status-correct-nonce-5", }, ) assert resp.status_code == 422 def test_status_correct_returns_403_for_user(self, user_client, animals_for_tagging): """Status correct returns 403 for non-admin user.""" import time from animaltrack.selection import compute_roster_hash roster_hash = compute_roster_hash(animals_for_tagging, None) ts_utc = int(time.time() * 1000) resp = user_client.post( "/actions/animal-status-correct", data={ "filter": "species:duck", "new_status": "dead", "reason": "Data entry error", "resolved_ids": animals_for_tagging, "roster_hash": roster_hash, "ts_utc": str(ts_utc), "nonce": "test-status-correct-nonce-6", }, ) assert resp.status_code == 403