Petru Paler
abf78ec98a
Add ability to specify custom date/time when recording events, enabling historical data entry. Forms show "Now - Set custom date" with a collapsible datetime picker that converts to milliseconds. - Add event_datetime_field() component in templates/actions.py - Add datetime picker to all event forms (cohort, hatch, outcome, tag, attrs, move, feed) - Add _parse_ts_utc() helper to parse form timestamp or use current - Add tests for backdating functionality 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
1686 lines
55 KiB
Python
1686 lines
55 KiB
Python
# 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
|
|
|
|
|
|
# =============================================================================
|
|
# Backdating Tests
|
|
# =============================================================================
|
|
|
|
|
|
class TestParseTsUtcHelper:
|
|
"""Tests for the _parse_ts_utc helper function."""
|
|
|
|
def test_parse_ts_utc_returns_current_for_none(self):
|
|
"""Returns current time when value is None."""
|
|
import time
|
|
|
|
from animaltrack.web.routes.actions import _parse_ts_utc
|
|
|
|
before = int(time.time() * 1000)
|
|
result = _parse_ts_utc(None)
|
|
after = int(time.time() * 1000)
|
|
|
|
assert before <= result <= after
|
|
|
|
def test_parse_ts_utc_returns_current_for_zero(self):
|
|
"""Returns current time when value is '0'."""
|
|
import time
|
|
|
|
from animaltrack.web.routes.actions import _parse_ts_utc
|
|
|
|
before = int(time.time() * 1000)
|
|
result = _parse_ts_utc("0")
|
|
after = int(time.time() * 1000)
|
|
|
|
assert before <= result <= after
|
|
|
|
def test_parse_ts_utc_returns_current_for_empty(self):
|
|
"""Returns current time when value is empty string."""
|
|
import time
|
|
|
|
from animaltrack.web.routes.actions import _parse_ts_utc
|
|
|
|
before = int(time.time() * 1000)
|
|
result = _parse_ts_utc("")
|
|
after = int(time.time() * 1000)
|
|
|
|
assert before <= result <= after
|
|
|
|
def test_parse_ts_utc_returns_provided_value(self):
|
|
"""Returns the provided timestamp when valid."""
|
|
from animaltrack.web.routes.actions import _parse_ts_utc
|
|
|
|
# Use a past timestamp
|
|
past_ts = 1700000000000 # Some time in 2023
|
|
|
|
result = _parse_ts_utc(str(past_ts))
|
|
|
|
assert result == past_ts
|
|
|
|
def test_parse_ts_utc_returns_current_for_invalid(self):
|
|
"""Returns current time when value is invalid."""
|
|
import time
|
|
|
|
from animaltrack.web.routes.actions import _parse_ts_utc
|
|
|
|
before = int(time.time() * 1000)
|
|
result = _parse_ts_utc("not-a-number")
|
|
after = int(time.time() * 1000)
|
|
|
|
assert before <= result <= after
|
|
|
|
|
|
class TestBackdatingCohort:
|
|
"""Tests for backdating cohort creation."""
|
|
|
|
def test_cohort_uses_provided_timestamp(self, client, seeded_db, location_strip1_id):
|
|
"""POST uses provided ts_utc for backdating."""
|
|
# Use a past timestamp (Feb 13, 2025 in ms)
|
|
backdated_ts = 1739404800000
|
|
|
|
resp = client.post(
|
|
"/actions/animal-cohort",
|
|
data={
|
|
"species": "duck",
|
|
"location_id": location_strip1_id,
|
|
"count": "2",
|
|
"life_stage": "adult",
|
|
"sex": "female",
|
|
"origin": "purchased",
|
|
"ts_utc": str(backdated_ts),
|
|
"nonce": "test-backdate-cohort-1",
|
|
},
|
|
)
|
|
|
|
assert resp.status_code == 200
|
|
|
|
# Verify the event was created with the backdated timestamp
|
|
event_row = seeded_db.execute(
|
|
"SELECT ts_utc FROM events WHERE type = 'AnimalCohortCreated' ORDER BY id DESC LIMIT 1"
|
|
).fetchone()
|
|
assert event_row is not None
|
|
assert event_row[0] == backdated_ts
|
|
|
|
def test_cohort_uses_current_time_when_ts_utc_zero(self, client, seeded_db, location_strip1_id):
|
|
"""POST uses current time when ts_utc is 0."""
|
|
import time
|
|
|
|
before = int(time.time() * 1000)
|
|
|
|
resp = client.post(
|
|
"/actions/animal-cohort",
|
|
data={
|
|
"species": "duck",
|
|
"location_id": location_strip1_id,
|
|
"count": "2",
|
|
"life_stage": "adult",
|
|
"sex": "female",
|
|
"origin": "purchased",
|
|
"ts_utc": "0",
|
|
"nonce": "test-backdate-cohort-2",
|
|
},
|
|
)
|
|
|
|
after = int(time.time() * 1000)
|
|
|
|
assert resp.status_code == 200
|
|
|
|
# Verify the event was created with a current timestamp
|
|
event_row = seeded_db.execute(
|
|
"SELECT ts_utc FROM events WHERE type = 'AnimalCohortCreated' ORDER BY id DESC LIMIT 1"
|
|
).fetchone()
|
|
assert event_row is not None
|
|
assert before <= event_row[0] <= after
|