Petru Paler
301b925be3
Add GET /animals/{animal_id} route to display individual animal details:
- Header summary with species, location, status, tags
- Event timeline showing all events affecting the animal (newest first)
- Quick actions card (Move functional, others disabled for now)
- Merge info alert for animals that have been merged
🤖 Generated with [Claude Code](https://claude.com/claude-code)
Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
354 lines
12 KiB
Python
354 lines
12 KiB
Python
# ABOUTME: Tests for Animal Detail web routes.
|
|
# ABOUTME: Covers GET /animals/{animal_id} rendering, timeline, merge info, and quick actions.
|
|
|
|
import os
|
|
import time
|
|
|
|
import pytest
|
|
from starlette.testclient import TestClient
|
|
|
|
from animaltrack.events.payloads import (
|
|
AnimalCohortCreatedPayload,
|
|
AnimalMergedPayload,
|
|
AnimalMovedPayload,
|
|
AnimalTaggedPayload,
|
|
)
|
|
from animaltrack.events.store import EventStore
|
|
from animaltrack.projections import ProjectionRegistry
|
|
from animaltrack.projections.animal_registry import AnimalRegistryProjection
|
|
from animaltrack.projections.event_animals import EventAnimalsProjection
|
|
from animaltrack.projections.intervals import IntervalProjection
|
|
from animaltrack.projections.tags import TagProjection
|
|
from animaltrack.services.animal import AnimalService
|
|
|
|
|
|
def make_test_settings(
|
|
csrf_secret: str = "test-secret",
|
|
trusted_proxy_ips: str = "127.0.0.1",
|
|
dev_mode: bool = True,
|
|
):
|
|
"""Create Settings for testing by setting env vars temporarily."""
|
|
from animaltrack.config import Settings
|
|
|
|
old_env = os.environ.copy()
|
|
try:
|
|
os.environ["CSRF_SECRET"] = csrf_secret
|
|
os.environ["TRUSTED_PROXY_IPS"] = trusted_proxy_ips
|
|
os.environ["DEV_MODE"] = str(dev_mode).lower()
|
|
return Settings()
|
|
finally:
|
|
os.environ.clear()
|
|
os.environ.update(old_env)
|
|
|
|
|
|
@pytest.fixture
|
|
def client(seeded_db):
|
|
"""Create a test client for the app."""
|
|
from animaltrack.web.app import create_app
|
|
|
|
settings = make_test_settings(trusted_proxy_ips="testclient")
|
|
app, rt = create_app(settings=settings, db=seeded_db)
|
|
return TestClient(app, raise_server_exceptions=True)
|
|
|
|
|
|
@pytest.fixture
|
|
def projection_registry(seeded_db):
|
|
"""Create a ProjectionRegistry with animal projections registered."""
|
|
registry = ProjectionRegistry()
|
|
registry.register(AnimalRegistryProjection(seeded_db))
|
|
registry.register(EventAnimalsProjection(seeded_db))
|
|
registry.register(IntervalProjection(seeded_db))
|
|
registry.register(TagProjection(seeded_db))
|
|
return registry
|
|
|
|
|
|
@pytest.fixture
|
|
def animal_service(seeded_db, projection_registry):
|
|
"""Create an AnimalService for testing."""
|
|
event_store = EventStore(seeded_db)
|
|
return AnimalService(seeded_db, event_store, projection_registry)
|
|
|
|
|
|
@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_strip2_id(seeded_db):
|
|
"""Get Strip 2 location ID from seeded data."""
|
|
row = seeded_db.execute("SELECT id FROM locations WHERE name = 'Strip 2'").fetchone()
|
|
return row[0]
|
|
|
|
|
|
@pytest.fixture
|
|
def single_duck(seeded_db, animal_service, location_strip1_id):
|
|
"""Create a single duck at Strip 1."""
|
|
payload = AnimalCohortCreatedPayload(
|
|
species="duck",
|
|
count=1,
|
|
life_stage="adult",
|
|
sex="female",
|
|
location_id=location_strip1_id,
|
|
origin="purchased",
|
|
)
|
|
ts_utc = int(time.time() * 1000)
|
|
event = animal_service.create_cohort(payload, ts_utc, "test_user")
|
|
return event.entity_refs["animal_ids"][0]
|
|
|
|
|
|
@pytest.fixture
|
|
def promoted_duck(seeded_db, single_duck):
|
|
"""Create a duck with nickname (identified)."""
|
|
seeded_db.execute(
|
|
"UPDATE animal_registry SET identified = 1, nickname = 'Daisy' WHERE animal_id = ?",
|
|
(single_duck,),
|
|
)
|
|
return single_duck
|
|
|
|
|
|
@pytest.fixture
|
|
def tagged_duck(seeded_db, animal_service, single_duck):
|
|
"""Create a duck with a tag."""
|
|
ts_utc = int(time.time() * 1000)
|
|
tag_payload = AnimalTaggedPayload(tag="layer", resolved_ids=[single_duck])
|
|
animal_service.add_tag(tag_payload, ts_utc + 1000, "test_user")
|
|
return single_duck
|
|
|
|
|
|
@pytest.fixture
|
|
def duck_with_history(seeded_db, animal_service, location_strip1_id, location_strip2_id):
|
|
"""Create a duck with multiple events (cohort creation, move, tag)."""
|
|
ts_utc = int(time.time() * 1000)
|
|
|
|
# Create cohort
|
|
payload = AnimalCohortCreatedPayload(
|
|
species="duck",
|
|
count=1,
|
|
life_stage="adult",
|
|
sex="female",
|
|
location_id=location_strip1_id,
|
|
origin="purchased",
|
|
)
|
|
event = animal_service.create_cohort(payload, ts_utc, "test_user")
|
|
animal_id = event.entity_refs["animal_ids"][0]
|
|
|
|
# Move to Strip 2
|
|
move_payload = AnimalMovedPayload(
|
|
resolved_ids=[animal_id],
|
|
to_location_id=location_strip2_id,
|
|
)
|
|
animal_service.move_animals(move_payload, ts_utc + 1000, "test_user")
|
|
|
|
# Add a tag
|
|
tag_payload = AnimalTaggedPayload(tag="breeder", resolved_ids=[animal_id])
|
|
animal_service.add_tag(tag_payload, ts_utc + 2000, "test_user")
|
|
|
|
return animal_id
|
|
|
|
|
|
@pytest.fixture
|
|
def merged_duck(seeded_db, animal_service, location_strip1_id):
|
|
"""Create two ducks and merge one into the other."""
|
|
ts_utc = int(time.time() * 1000)
|
|
|
|
# Create survivor
|
|
payload1 = AnimalCohortCreatedPayload(
|
|
species="duck",
|
|
count=1,
|
|
life_stage="adult",
|
|
sex="female",
|
|
location_id=location_strip1_id,
|
|
origin="purchased",
|
|
)
|
|
event1 = animal_service.create_cohort(payload1, ts_utc, "test_user")
|
|
survivor_id = event1.entity_refs["animal_ids"][0]
|
|
|
|
# Give survivor a nickname
|
|
seeded_db.execute(
|
|
"UPDATE animal_registry SET identified = 1, nickname = 'Survivor' WHERE animal_id = ?",
|
|
(survivor_id,),
|
|
)
|
|
|
|
# Create alias
|
|
payload2 = AnimalCohortCreatedPayload(
|
|
species="duck",
|
|
count=1,
|
|
life_stage="adult",
|
|
sex="female",
|
|
location_id=location_strip1_id,
|
|
origin="purchased",
|
|
)
|
|
event2 = animal_service.create_cohort(payload2, ts_utc + 1000, "test_user")
|
|
alias_id = event2.entity_refs["animal_ids"][0]
|
|
|
|
# Merge alias into survivor
|
|
merge_payload = AnimalMergedPayload(
|
|
survivor_animal_id=survivor_id,
|
|
merged_animal_ids=[alias_id],
|
|
)
|
|
animal_service.merge_animals(merge_payload, ts_utc + 2000, "test_user")
|
|
|
|
return {"alias_id": alias_id, "survivor_id": survivor_id}
|
|
|
|
|
|
@pytest.fixture
|
|
def dead_duck(seeded_db, single_duck):
|
|
"""Create a dead duck."""
|
|
seeded_db.execute(
|
|
"UPDATE animal_registry SET status = 'dead' WHERE animal_id = ?",
|
|
(single_duck,),
|
|
)
|
|
return single_duck
|
|
|
|
|
|
class TestAnimalDetailRendering:
|
|
"""Tests for basic page rendering."""
|
|
|
|
def test_detail_renders_for_valid_animal(self, client, single_duck):
|
|
"""GET /animals/{id} returns 200 with valid animal."""
|
|
response = client.get(
|
|
f"/animals/{single_duck}",
|
|
headers={"X-Oidc-Username": "test_user"},
|
|
)
|
|
assert response.status_code == 200
|
|
assert "Animal Details" in response.text
|
|
|
|
def test_detail_returns_404_for_invalid_id(self, client):
|
|
"""GET /animals/{invalid_id} returns 404."""
|
|
response = client.get(
|
|
"/animals/00000000000000000000000000",
|
|
headers={"X-Oidc-Username": "test_user"},
|
|
)
|
|
assert response.status_code == 404
|
|
|
|
def test_detail_shows_animal_info(self, client, single_duck):
|
|
"""Detail page shows species, location, status, etc."""
|
|
response = client.get(
|
|
f"/animals/{single_duck}",
|
|
headers={"X-Oidc-Username": "test_user"},
|
|
)
|
|
assert response.status_code == 200
|
|
assert "Duck" in response.text
|
|
assert "Strip 1" in response.text
|
|
assert "alive" in response.text.lower()
|
|
|
|
def test_detail_shows_nickname_if_identified(self, client, promoted_duck):
|
|
"""Detail page shows nickname for identified animals."""
|
|
response = client.get(
|
|
f"/animals/{promoted_duck}",
|
|
headers={"X-Oidc-Username": "test_user"},
|
|
)
|
|
assert response.status_code == 200
|
|
assert "Daisy" in response.text
|
|
|
|
def test_detail_shows_truncated_id_if_not_identified(self, client, single_duck):
|
|
"""Detail page shows truncated ID for unidentified animals."""
|
|
response = client.get(
|
|
f"/animals/{single_duck}",
|
|
headers={"X-Oidc-Username": "test_user"},
|
|
)
|
|
assert response.status_code == 200
|
|
# Should show first 8 chars of ID
|
|
assert single_duck[:8] in response.text
|
|
|
|
def test_detail_shows_current_tags(self, client, tagged_duck):
|
|
"""Detail page shows active tags."""
|
|
response = client.get(
|
|
f"/animals/{tagged_duck}",
|
|
headers={"X-Oidc-Username": "test_user"},
|
|
)
|
|
assert response.status_code == 200
|
|
assert "layer" in response.text
|
|
|
|
|
|
class TestAnimalTimeline:
|
|
"""Tests for timeline display."""
|
|
|
|
def test_timeline_shows_events(self, client, duck_with_history):
|
|
"""Timeline shows events affecting the animal."""
|
|
response = client.get(
|
|
f"/animals/{duck_with_history}",
|
|
headers={"X-Oidc-Username": "test_user"},
|
|
)
|
|
assert response.status_code == 200
|
|
assert "Event Timeline" in response.text
|
|
assert "AnimalCohortCreated" in response.text
|
|
assert "AnimalMoved" in response.text
|
|
assert "AnimalTagged" in response.text
|
|
|
|
def test_timeline_ordered_newest_first(self, client, duck_with_history):
|
|
"""Timeline events are ordered newest first."""
|
|
response = client.get(
|
|
f"/animals/{duck_with_history}",
|
|
headers={"X-Oidc-Username": "test_user"},
|
|
)
|
|
assert response.status_code == 200
|
|
# AnimalTagged should appear before AnimalMoved which should appear before AnimalCohortCreated
|
|
tagged_pos = response.text.find("AnimalTagged")
|
|
moved_pos = response.text.find("AnimalMoved")
|
|
created_pos = response.text.find("AnimalCohortCreated")
|
|
assert tagged_pos < moved_pos < created_pos
|
|
|
|
|
|
class TestMergeInfo:
|
|
"""Tests for merged animal handling."""
|
|
|
|
def test_merged_animal_shows_alert(self, client, merged_duck):
|
|
"""Merged animals show merge alert with survivor link."""
|
|
alias_id = merged_duck["alias_id"]
|
|
response = client.get(
|
|
f"/animals/{alias_id}",
|
|
headers={"X-Oidc-Username": "test_user"},
|
|
)
|
|
assert response.status_code == 200
|
|
assert "merged into" in response.text.lower()
|
|
assert "Survivor" in response.text
|
|
|
|
def test_merged_alert_links_to_survivor(self, client, merged_duck):
|
|
"""Merge alert links to the survivor animal."""
|
|
alias_id = merged_duck["alias_id"]
|
|
survivor_id = merged_duck["survivor_id"]
|
|
response = client.get(
|
|
f"/animals/{alias_id}",
|
|
headers={"X-Oidc-Username": "test_user"},
|
|
)
|
|
assert response.status_code == 200
|
|
assert f"/animals/{survivor_id}" in response.text
|
|
|
|
def test_alive_animal_no_merge_alert(self, client, single_duck):
|
|
"""Alive animals don't show merge alert."""
|
|
response = client.get(
|
|
f"/animals/{single_duck}",
|
|
headers={"X-Oidc-Username": "test_user"},
|
|
)
|
|
assert response.status_code == 200
|
|
assert "merged into" not in response.text.lower()
|
|
|
|
|
|
class TestQuickActions:
|
|
"""Tests for quick actions card."""
|
|
|
|
def test_alive_animal_shows_actions(self, client, single_duck):
|
|
"""Alive animals show available actions."""
|
|
response = client.get(
|
|
f"/animals/{single_duck}",
|
|
headers={"X-Oidc-Username": "test_user"},
|
|
)
|
|
assert response.status_code == 200
|
|
assert "Quick Actions" in response.text
|
|
assert "Move" in response.text
|
|
|
|
def test_dead_animal_no_actions(self, client, dead_duck):
|
|
"""Dead/harvested animals show no actions."""
|
|
response = client.get(
|
|
f"/animals/{dead_duck}",
|
|
headers={"X-Oidc-Username": "test_user"},
|
|
)
|
|
assert response.status_code == 200
|
|
# Should still show Quick Actions card but with "No actions available"
|
|
assert "Quick Actions" in response.text
|
|
assert "No actions available" in response.text
|