feat: implement Animal Detail page with timeline (Step 8.3)

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>
This commit is contained in:
2025-12-30 20:14:12 +00:00
parent bce4d099c9
commit 301b925be3
8 changed files with 1296 additions and 9 deletions

18
PLAN.md
View File

@@ -326,15 +326,15 @@ Check off items as completed. Each phase builds on the previous.
- [x] **Commit checkpoint** - [x] **Commit checkpoint**
### Step 8.3: Animal Detail Drawer ### Step 8.3: Animal Detail Drawer
- [ ] Create `web/routes/animals.py`: - [x] Create `web/routes/animals.py`:
- [ ] GET /animals/{animal_id} - [x] GET /animals/{animal_id}
- [ ] Create `web/templates/animal_detail.py`: - [x] Create `web/templates/animal_detail.py`:
- [ ] Header summary - [x] Header summary
- [ ] Timeline (newest first) - [x] Timeline (newest first)
- [ ] Quick actions - [x] Quick actions
- [ ] Create `repositories/animal_timeline.py` - [x] Create `repositories/animal_timeline.py`
- [ ] Write tests: detail renders, timeline shows events, merge info shown - [x] Write tests: detail renders, timeline shows events, merge info shown
- [ ] **Commit checkpoint** - [x] **Commit checkpoint**
--- ---

View File

@@ -0,0 +1,277 @@
# ABOUTME: Repository for animal detail and timeline queries.
# ABOUTME: Provides get_animal, get_timeline, and get_merge_info for detail views.
import json
from dataclasses import dataclass, field
from typing import Any
@dataclass
class AnimalDetail:
"""Complete animal data for detail view."""
animal_id: str
species_code: str
species_name: str
sex: str
repro_status: str
life_stage: str
status: str
location_id: str
location_name: str
origin: str
nickname: str | None
identified: bool
born_or_hatched_at: int | None
acquired_at: int | None
first_seen_utc: int
last_event_utc: int
tags: list[str] = field(default_factory=list)
@dataclass
class TimelineEvent:
"""Single event in the animal timeline."""
event_id: str
event_type: str
ts_utc: int
actor: str
summary: dict[str, Any]
@dataclass
class MergeInfo:
"""Info about a merged animal."""
survivor_animal_id: str
survivor_nickname: str | None
merged_at_utc: int
class AnimalTimelineRepository:
"""Repository for animal detail and timeline operations."""
def __init__(self, db: Any) -> None:
"""Initialize repository with database connection.
Args:
db: A fastlite database connection.
"""
self.db = db
def get_animal(self, animal_id: str) -> AnimalDetail | None:
"""Get full animal data with joined species/location names.
Args:
animal_id: The animal's ULID.
Returns:
AnimalDetail if found, None otherwise.
"""
query = """
SELECT
ar.animal_id,
ar.species_code,
s.name as species_name,
ar.sex,
ar.repro_status,
ar.life_stage,
ar.status,
ar.location_id,
l.name as location_name,
ar.origin,
ar.nickname,
ar.identified,
ar.born_or_hatched_at,
ar.acquired_at,
ar.first_seen_utc,
ar.last_event_utc
FROM animal_registry ar
JOIN species s ON ar.species_code = s.code
JOIN locations l ON ar.location_id = l.id
WHERE ar.animal_id = ?
"""
row = self.db.execute(query, (animal_id,)).fetchone()
if row is None:
return None
# Get active tags
tags_query = """
SELECT tag
FROM animal_tag_intervals
WHERE animal_id = ? AND end_utc IS NULL
ORDER BY tag
"""
tag_rows = self.db.execute(tags_query, (animal_id,)).fetchall()
tags = [r[0] for r in tag_rows]
return AnimalDetail(
animal_id=row[0],
species_code=row[1],
species_name=row[2],
sex=row[3],
repro_status=row[4],
life_stage=row[5],
status=row[6],
location_id=row[7],
location_name=row[8],
origin=row[9],
nickname=row[10],
identified=bool(row[11]),
born_or_hatched_at=row[12],
acquired_at=row[13],
first_seen_utc=row[14],
last_event_utc=row[15],
tags=tags,
)
def get_timeline(self, animal_id: str, limit: int = 50) -> list[TimelineEvent]:
"""Get events affecting this animal, newest first.
Args:
animal_id: The animal's ULID.
limit: Maximum number of events to return.
Returns:
List of TimelineEvent objects ordered by ts_utc DESC.
"""
query = """
SELECT e.id, e.type, e.ts_utc, e.actor, e.payload
FROM events e
JOIN event_animals ea ON e.id = ea.event_id
WHERE ea.animal_id = ?
ORDER BY e.ts_utc DESC
LIMIT ?
"""
rows = self.db.execute(query, (animal_id, limit)).fetchall()
events = []
for row in rows:
payload = json.loads(row[4]) if row[4] else {}
summary = self._build_summary(row[1], payload)
events.append(
TimelineEvent(
event_id=row[0],
event_type=row[1],
ts_utc=row[2],
actor=row[3],
summary=summary,
)
)
return events
def _build_summary(self, event_type: str, payload: dict[str, Any]) -> dict[str, Any]:
"""Build summary dict from event payload.
Args:
event_type: The event type string.
payload: The event payload dict.
Returns:
Summary dict with type-specific fields.
"""
summary: dict[str, Any] = {}
if event_type == "AnimalCohortCreated":
summary["species"] = payload.get("species", "")
summary["count"] = payload.get("count", 0)
summary["origin"] = payload.get("origin", "")
elif event_type == "AnimalMoved":
summary["from_location_id"] = payload.get("from_location_id", "")
summary["to_location_id"] = payload.get("to_location_id", "")
# Look up location names
from_name = self._get_location_name(payload.get("from_location_id"))
to_name = self._get_location_name(payload.get("to_location_id"))
summary["from_location_name"] = from_name
summary["to_location_name"] = to_name
elif event_type == "AnimalTagged":
summary["tag"] = payload.get("tag", "")
elif event_type == "AnimalTagEnded":
summary["tag"] = payload.get("tag", "")
elif event_type == "ProductCollected":
summary["product_code"] = payload.get("product_code", "")
summary["quantity"] = payload.get("quantity", 0)
elif event_type == "HatchRecorded":
summary["hatched_live"] = payload.get("hatched_live", 0)
summary["hatched_dead"] = payload.get("hatched_dead", 0)
elif event_type == "AnimalOutcome":
summary["outcome"] = payload.get("outcome", "")
elif event_type == "AnimalPromoted":
summary["nickname"] = payload.get("nickname", "")
elif event_type == "AnimalMerged":
summary["survivor_id"] = payload.get("survivor_animal_id", "")
# Look up survivor nickname
survivor_id = payload.get("survivor_animal_id")
if survivor_id:
row = self.db.execute(
"SELECT nickname FROM animal_registry WHERE animal_id = ?",
(survivor_id,),
).fetchone()
summary["survivor_nickname"] = row[0] if row else None
elif event_type == "AnimalStatusCorrected":
summary["new_status"] = payload.get("new_status", "")
summary["reason"] = payload.get("reason", "")
elif event_type == "AnimalAttributesUpdated":
# Extract changed attributes from payload
summary["changes"] = {}
for attr in ["sex", "life_stage", "repro_status"]:
if attr in payload:
summary["changes"][attr] = payload[attr]
return summary
def _get_location_name(self, location_id: str | None) -> str:
"""Get location name by ID.
Args:
location_id: The location ULID.
Returns:
Location name or empty string if not found.
"""
if not location_id:
return ""
row = self.db.execute(
"SELECT name FROM locations WHERE id = ?",
(location_id,),
).fetchone()
return row[0] if row else ""
def get_merge_info(self, animal_id: str) -> MergeInfo | None:
"""Get merge info if this animal was merged into another.
Args:
animal_id: The animal's ULID (the alias, not survivor).
Returns:
MergeInfo if animal was merged, None otherwise.
"""
query = """
SELECT
aa.survivor_animal_id,
ar.nickname,
aa.merged_at_utc
FROM animal_aliases aa
JOIN animal_registry ar ON aa.survivor_animal_id = ar.animal_id
WHERE aa.alias_animal_id = ?
"""
row = self.db.execute(query, (animal_id,)).fetchone()
if row is None:
return None
return MergeInfo(
survivor_animal_id=row[0],
survivor_nickname=row[1],
merged_at_utc=row[2],
)

View File

@@ -18,6 +18,7 @@ from animaltrack.web.middleware import (
request_id_before, request_id_before,
) )
from animaltrack.web.routes import ( from animaltrack.web.routes import (
register_animals_routes,
register_egg_routes, register_egg_routes,
register_events_routes, register_events_routes,
register_feed_routes, register_feed_routes,
@@ -132,6 +133,7 @@ def create_app(
# Register routes # Register routes
register_health_routes(rt, app) register_health_routes(rt, app)
register_animals_routes(rt, app)
register_egg_routes(rt, app) register_egg_routes(rt, app)
register_events_routes(rt, app) register_events_routes(rt, app)
register_feed_routes(rt, app) register_feed_routes(rt, app)

View File

@@ -1,6 +1,7 @@
# ABOUTME: Routes package for AnimalTrack web application. # ABOUTME: Routes package for AnimalTrack web application.
# ABOUTME: Contains modular route handlers for different features. # ABOUTME: Contains modular route handlers for different features.
from animaltrack.web.routes.animals import register_animals_routes
from animaltrack.web.routes.eggs import register_egg_routes from animaltrack.web.routes.eggs import register_egg_routes
from animaltrack.web.routes.events import register_events_routes from animaltrack.web.routes.events import register_events_routes
from animaltrack.web.routes.feed import register_feed_routes from animaltrack.web.routes.feed import register_feed_routes
@@ -9,6 +10,7 @@ from animaltrack.web.routes.move import register_move_routes
from animaltrack.web.routes.registry import register_registry_routes from animaltrack.web.routes.registry import register_registry_routes
__all__ = [ __all__ = [
"register_animals_routes",
"register_egg_routes", "register_egg_routes",
"register_events_routes", "register_events_routes",
"register_feed_routes", "register_feed_routes",

View File

@@ -0,0 +1,47 @@
# ABOUTME: Routes for individual animal detail views.
# ABOUTME: Handles GET /animals/{animal_id} for animal detail page.
from starlette.requests import Request
from starlette.responses import HTMLResponse
from animaltrack.repositories.animal_timeline import AnimalTimelineRepository
from animaltrack.web.templates.animal_detail import animal_detail_page
from animaltrack.web.templates.base import page
def animal_detail(request: Request, animal_id: str):
"""GET /animals/{animal_id} - Animal detail page with timeline."""
db = request.app.state.db
repo = AnimalTimelineRepository(db)
# Get animal data
animal = repo.get_animal(animal_id)
if animal is None:
return HTMLResponse(
content="<p>Animal not found</p>",
status_code=404,
)
# Get timeline events
timeline = repo.get_timeline(animal_id)
# Get merge info if applicable
merge_info = None
if animal.status == "merged_into":
merge_info = repo.get_merge_info(animal_id)
# Build title
display_name = animal.nickname or f"{animal.animal_id[:8]}..."
title = f"{display_name} - AnimalTrack"
# Full page render
return page(
animal_detail_page(animal, timeline, merge_info),
title=title,
active_nav="registry",
)
def register_animals_routes(rt, app):
"""Register animal detail routes."""
rt("/animals/{animal_id}")(animal_detail)

View File

@@ -0,0 +1,303 @@
# ABOUTME: Templates for animal detail page.
# ABOUTME: Renders header summary, timeline, merge info, and quick actions.
from datetime import UTC, datetime
from typing import Any
from fasthtml.common import H2, H3, A, Div, Li, P, Span, Ul
from monsterui.all import Button, ButtonT, Card, Grid
from animaltrack.repositories.animal_timeline import (
AnimalDetail,
MergeInfo,
TimelineEvent,
)
def format_timestamp(ts_utc: int) -> str:
"""Format timestamp for display."""
dt = datetime.fromtimestamp(ts_utc / 1000, tz=UTC)
return dt.strftime("%Y-%m-%d %H:%M")
def animal_detail_page(
animal: AnimalDetail,
timeline: list[TimelineEvent],
merge_info: MergeInfo | None = None,
) -> Div:
"""Full animal detail page."""
return Div(
back_to_registry_link(),
Grid(
Div(
animal_header_card(animal, merge_info),
animal_timeline_section(timeline),
cls="space-y-4",
),
Div(
quick_actions_card(animal),
attributes_card(animal),
cls="space-y-4",
),
cols_sm=1,
cols_lg=3,
cls="gap-4 p-4",
),
id="animal-detail",
)
def back_to_registry_link() -> Div:
"""Link back to registry."""
return Div(
A(
"< Back to Registry",
href="/registry",
cls="text-amber-500 hover:underline text-sm",
),
cls="p-4",
)
def animal_header_card(animal: AnimalDetail, merge_info: MergeInfo | None) -> Card:
"""Header card with animal summary."""
display_name = animal.nickname or f"{animal.animal_id[:8]}..."
status_badge = status_badge_component(animal.status)
tags_display = (
Div(
*[tag_badge(tag) for tag in animal.tags],
cls="flex flex-wrap gap-1 mt-2",
)
if animal.tags
else None
)
merge_alert = merge_info_alert(merge_info) if merge_info else None
return Card(
merge_alert,
Div(
H2(display_name, cls="text-xl font-bold"),
Span(animal.animal_id, cls="text-xs text-stone-400 font-mono"),
cls="mb-3",
),
Grid(
info_item("Species", animal.species_name),
info_item("Sex", animal.sex.title()),
info_item("Life Stage", animal.life_stage.replace("_", " ").title()),
info_item("Location", animal.location_name),
info_item("Origin", animal.origin.title()),
info_item("Status", status_badge),
cols_sm=2,
cols_md=3,
cls="gap-3",
),
tags_display,
header=Div(
P("Animal Details", cls="text-sm font-medium text-stone-400"),
),
)
def info_item(label: str, value: Any) -> Div:
"""Single key-value info item."""
return Div(
P(label, cls="text-xs text-stone-400"),
P(value if isinstance(value, str) else value, cls="text-sm"),
)
def status_badge_component(status: str) -> Span:
"""Badge for animal status."""
status_cls = {
"alive": "bg-green-900/50 text-green-300",
"dead": "bg-red-900/50 text-red-300",
"harvested": "bg-amber-900/50 text-amber-300",
"sold": "bg-blue-900/50 text-blue-300",
"merged_into": "bg-slate-700 text-slate-300",
}.get(status, "bg-slate-700 text-slate-300")
return Span(status, cls=f"text-xs px-2 py-0.5 rounded {status_cls}")
def tag_badge(tag: str) -> Span:
"""Badge for a tag."""
return Span(tag, cls="text-xs px-2 py-0.5 rounded bg-indigo-900/50 text-indigo-300")
def merge_info_alert(merge_info: MergeInfo) -> Div:
"""Alert showing merge information."""
survivor_display = merge_info.survivor_nickname or merge_info.survivor_animal_id[:8]
return Div(
P(
"This animal was merged into ",
A(
survivor_display,
href=f"/animals/{merge_info.survivor_animal_id}",
cls="text-amber-500 hover:underline",
),
cls="text-sm",
),
cls="bg-slate-800 border border-slate-600 rounded p-3 mb-4",
)
def quick_actions_card(animal: AnimalDetail) -> Card:
"""Quick actions available for this animal."""
actions = []
if animal.status == "alive":
actions.append(
A(
Button("Move", cls=ButtonT.default + " w-full"),
href=f"/move?filter=animal_id:{animal.animal_id}",
)
)
actions.append(
Button("Add Tag", cls=ButtonT.default + " w-full", disabled=True),
)
if not animal.identified:
actions.append(
Button("Promote", cls=ButtonT.default + " w-full", disabled=True),
)
actions.append(
Button("Record Outcome", cls=ButtonT.destructive + " w-full", disabled=True),
)
return Card(
Div(*actions, cls="space-y-2")
if actions
else P("No actions available", cls="text-sm text-stone-400"),
header=Div(P("Quick Actions", cls="text-sm font-medium text-stone-400")),
)
def attributes_card(animal: AnimalDetail) -> Card:
"""Detailed attributes card."""
first_seen = format_timestamp(animal.first_seen_utc)
last_event = format_timestamp(animal.last_event_utc)
born_at = (
format_timestamp(animal.born_or_hatched_at) if animal.born_or_hatched_at else "Unknown"
)
return Card(
info_item("Identified", "Yes" if animal.identified else "No"),
info_item("Repro Status", animal.repro_status.title()),
info_item("Born/Hatched", born_at),
info_item("First Seen", first_seen),
info_item("Last Event", last_event),
header=Div(P("Attributes", cls="text-sm font-medium text-stone-400")),
)
def animal_timeline_section(timeline: list[TimelineEvent]) -> Div:
"""Timeline section with event history."""
return Div(
H3("Event Timeline", cls="text-lg font-semibold mb-4"),
animal_timeline_list(timeline),
cls="bg-slate-900/50 rounded-lg p-4",
id="timeline-section",
)
def animal_timeline_list(timeline: list[TimelineEvent]) -> Ul:
"""Timeline list."""
if not timeline:
return Div(
P("No events recorded yet.", cls="text-stone-500 text-sm"),
cls="p-4 text-center",
)
items = [timeline_event_item(event) for event in timeline]
return Ul(*items, cls="space-y-3", id="event-timeline")
def timeline_event_item(event: TimelineEvent) -> Li:
"""Single timeline event item."""
badge_cls = event_type_badge_class(event.event_type)
summary_text = format_timeline_summary(event.event_type, event.summary)
time_str = format_timestamp(event.ts_utc)
return Li(
Div(
Span(event.event_type, cls=f"text-xs font-medium px-2 py-1 rounded {badge_cls}"),
Span(time_str, cls="text-xs text-stone-500 ml-2"),
cls="flex items-center gap-2 mb-1",
),
P(summary_text, cls="text-sm text-stone-300"),
P(f"by {event.actor}", cls="text-xs text-stone-500"),
cls="py-3 border-b border-stone-700 last:border-0",
)
def event_type_badge_class(event_type: str) -> str:
"""Get badge color class for event type."""
type_colors = {
"AnimalCohortCreated": "bg-green-900/50 text-green-300",
"AnimalMoved": "bg-purple-900/50 text-purple-300",
"AnimalAttributesUpdated": "bg-blue-900/50 text-blue-300",
"AnimalTagged": "bg-indigo-900/50 text-indigo-300",
"AnimalTagEnded": "bg-slate-700 text-slate-300",
"ProductCollected": "bg-amber-900/50 text-amber-300",
"HatchRecorded": "bg-pink-900/50 text-pink-300",
"AnimalOutcome": "bg-red-900/50 text-red-300",
"AnimalPromoted": "bg-cyan-900/50 text-cyan-300",
"AnimalMerged": "bg-slate-600 text-slate-300",
"AnimalStatusCorrected": "bg-orange-900/50 text-orange-300",
}
return type_colors.get(event_type, "bg-gray-700 text-gray-300")
def format_timeline_summary(event_type: str, summary: dict[str, Any]) -> str:
"""Format summary text based on event type."""
if event_type == "AnimalCohortCreated":
origin = summary.get("origin", "unknown")
return f"Created as part of a cohort ({origin})"
if event_type == "AnimalMoved":
to_loc = summary.get("to_location_name", "unknown")
from_loc = summary.get("from_location_name", "unknown")
return f"Moved from {from_loc} to {to_loc}"
if event_type == "AnimalAttributesUpdated":
changes = summary.get("changes", {})
if changes:
parts = [f"{k}: {v}" for k, v in changes.items()]
return f"Updated: {', '.join(parts)}"
return "Attributes updated"
if event_type == "AnimalTagged":
tag = summary.get("tag", "")
return f"Tagged as '{tag}'"
if event_type == "AnimalTagEnded":
tag = summary.get("tag", "")
return f"Tag '{tag}' removed"
if event_type == "ProductCollected":
product = summary.get("product_code", "product")
qty = summary.get("quantity", 0)
return f"Involved in collecting {qty} {product}"
if event_type == "HatchRecorded":
return "Hatched"
if event_type == "AnimalOutcome":
outcome = summary.get("outcome", "unknown")
return f"Outcome: {outcome}"
if event_type == "AnimalPromoted":
nickname = summary.get("nickname", "")
return "Promoted to identified animal" + (f" as '{nickname}'" if nickname else "")
if event_type == "AnimalMerged":
survivor = summary.get("survivor_nickname") or summary.get("survivor_id", "")[:8]
return f"Merged into {survivor}"
if event_type == "AnimalStatusCorrected":
new_status = summary.get("new_status", "")
reason = summary.get("reason", "")
return f"Status corrected to {new_status}" + (f": {reason}" if reason else "")
return event_type

View File

@@ -0,0 +1,303 @@
# ABOUTME: Tests for AnimalTimelineRepository - get_animal, get_timeline, get_merge_info.
# ABOUTME: Covers fetching animal details with joins and building event timelines.
import time
import pytest
from animaltrack.events import types as event_types
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.repositories.animal_timeline import (
AnimalDetail,
AnimalTimelineRepository,
MergeInfo,
TimelineEvent,
)
from animaltrack.services.animal import AnimalService
@pytest.fixture
def event_store(seeded_db):
"""Create an EventStore for testing."""
return EventStore(seeded_db)
@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, event_store, projection_registry):
"""Create an AnimalService for testing."""
return AnimalService(seeded_db, event_store, projection_registry)
@pytest.fixture
def valid_location_id(seeded_db):
"""Get Strip 1 location ID from seeds."""
row = seeded_db.execute("SELECT id FROM locations WHERE name = 'Strip 1'").fetchone()
return row[0]
@pytest.fixture
def strip2_location_id(seeded_db):
"""Get Strip 2 location ID from seeds."""
row = seeded_db.execute("SELECT id FROM locations WHERE name = 'Strip 2'").fetchone()
return row[0]
def make_cohort_payload(
location_id: str,
count: int = 3,
species: str = "duck",
sex: str = "unknown",
life_stage: str = "adult",
) -> AnimalCohortCreatedPayload:
"""Create a cohort payload for testing."""
return AnimalCohortCreatedPayload(
species=species,
count=count,
life_stage=life_stage,
sex=sex,
location_id=location_id,
origin="purchased",
)
class TestAnimalTimelineRepositoryGetAnimal:
"""Tests for get_animal method."""
def test_returns_animal_detail_with_joined_data(
self, seeded_db, animal_service, valid_location_id
):
"""get_animal returns AnimalDetail with species and location names."""
ts_utc = int(time.time() * 1000)
payload = make_cohort_payload(valid_location_id, count=1)
event = animal_service.create_cohort(payload, ts_utc, "test_user")
animal_id = event.entity_refs["animal_ids"][0]
repo = AnimalTimelineRepository(seeded_db)
animal = repo.get_animal(animal_id)
assert animal is not None
assert isinstance(animal, AnimalDetail)
assert animal.animal_id == animal_id
assert animal.species_code == "duck"
assert animal.species_name == "Duck" # From species table
assert animal.location_id == valid_location_id
assert animal.location_name == "Strip 1" # From locations table
assert animal.sex == "unknown"
assert animal.life_stage == "adult"
assert animal.status == "alive"
assert animal.origin == "purchased"
def test_returns_none_for_invalid_id(self, seeded_db):
"""get_animal returns None for non-existent animal ID."""
repo = AnimalTimelineRepository(seeded_db)
animal = repo.get_animal("00000000000000000000000000")
assert animal is None
def test_includes_active_tags(self, seeded_db, animal_service, valid_location_id):
"""get_animal includes currently active tags."""
ts_utc = int(time.time() * 1000)
payload = make_cohort_payload(valid_location_id, count=1)
event = animal_service.create_cohort(payload, ts_utc, "test_user")
animal_id = event.entity_refs["animal_ids"][0]
# Add tags
tag_payload = AnimalTaggedPayload(tag="layer", resolved_ids=[animal_id])
animal_service.add_tag(tag_payload, ts_utc + 1000, "test_user")
repo = AnimalTimelineRepository(seeded_db)
animal = repo.get_animal(animal_id)
assert animal is not None
assert "layer" in animal.tags
def test_includes_nickname_when_identified(self, seeded_db, animal_service, valid_location_id):
"""get_animal includes nickname for identified animals."""
ts_utc = int(time.time() * 1000)
payload = make_cohort_payload(valid_location_id, count=1)
event = animal_service.create_cohort(payload, ts_utc, "test_user")
animal_id = event.entity_refs["animal_ids"][0]
# Promote with nickname
seeded_db.execute(
"UPDATE animal_registry SET identified = 1, nickname = 'Daisy' WHERE animal_id = ?",
(animal_id,),
)
repo = AnimalTimelineRepository(seeded_db)
animal = repo.get_animal(animal_id)
assert animal is not None
assert animal.identified is True
assert animal.nickname == "Daisy"
class TestAnimalTimelineRepositoryGetTimeline:
"""Tests for get_timeline method."""
def test_returns_events_for_animal(self, seeded_db, animal_service, valid_location_id):
"""get_timeline returns events affecting the animal."""
ts_utc = int(time.time() * 1000)
payload = make_cohort_payload(valid_location_id, count=1)
event = animal_service.create_cohort(payload, ts_utc, "test_user")
animal_id = event.entity_refs["animal_ids"][0]
repo = AnimalTimelineRepository(seeded_db)
timeline = repo.get_timeline(animal_id)
assert len(timeline) == 1
assert isinstance(timeline[0], TimelineEvent)
assert timeline[0].event_type == "AnimalCohortCreated"
assert timeline[0].actor == "test_user"
def test_orders_events_newest_first(
self, seeded_db, animal_service, valid_location_id, strip2_location_id
):
"""get_timeline returns events in descending timestamp order."""
ts_utc = int(time.time() * 1000)
# Create animal
payload = make_cohort_payload(valid_location_id, count=1)
event = animal_service.create_cohort(payload, ts_utc, "test_user")
animal_id = event.entity_refs["animal_ids"][0]
# Move animal (later event)
move_payload = AnimalMovedPayload(
resolved_ids=[animal_id],
to_location_id=strip2_location_id,
)
animal_service.move_animals(move_payload, ts_utc + 1000, "test_user")
repo = AnimalTimelineRepository(seeded_db)
timeline = repo.get_timeline(animal_id)
assert len(timeline) == 2
# Most recent first
assert timeline[0].event_type == "AnimalMoved"
assert timeline[1].event_type == "AnimalCohortCreated"
def test_returns_empty_for_nonexistent_animal(self, seeded_db):
"""get_timeline returns empty list for non-existent animal."""
repo = AnimalTimelineRepository(seeded_db)
timeline = repo.get_timeline("00000000000000000000000000")
assert timeline == []
def test_respects_limit_parameter(self, seeded_db, animal_service, valid_location_id):
"""get_timeline respects the limit parameter."""
ts_utc = int(time.time() * 1000)
payload = make_cohort_payload(valid_location_id, count=1)
event = animal_service.create_cohort(payload, ts_utc, "test_user")
animal_id = event.entity_refs["animal_ids"][0]
# Add multiple tags to create more events
for i in range(5):
tag_payload = AnimalTaggedPayload(tag=f"tag{i}", resolved_ids=[animal_id])
animal_service.add_tag(tag_payload, ts_utc + (i + 1) * 1000, "test_user")
repo = AnimalTimelineRepository(seeded_db)
timeline = repo.get_timeline(animal_id, limit=3)
assert len(timeline) == 3
def test_includes_summary_data(self, seeded_db, animal_service, valid_location_id):
"""get_timeline includes summary data in each event."""
ts_utc = int(time.time() * 1000)
payload = make_cohort_payload(valid_location_id, count=1)
event = animal_service.create_cohort(payload, ts_utc, "test_user")
animal_id = event.entity_refs["animal_ids"][0]
# Add a tag
tag_payload = AnimalTaggedPayload(tag="layer", resolved_ids=[animal_id])
animal_service.add_tag(tag_payload, ts_utc + 1000, "test_user")
repo = AnimalTimelineRepository(seeded_db)
timeline = repo.get_timeline(animal_id)
# Check tag event has summary data
tag_event = timeline[0]
assert tag_event.event_type == event_types.ANIMAL_TAGGED
assert "tag" in tag_event.summary
assert tag_event.summary["tag"] == "layer"
class TestAnimalTimelineRepositoryGetMergeInfo:
"""Tests for get_merge_info method."""
def test_returns_merge_info_for_merged_animal(
self, seeded_db, animal_service, valid_location_id
):
"""get_merge_info returns info about the survivor for merged animals."""
ts_utc = int(time.time() * 1000)
# Create two cohorts
payload1 = make_cohort_payload(valid_location_id, count=1)
event1 = animal_service.create_cohort(payload1, ts_utc, "test_user")
survivor_id = event1.entity_refs["animal_ids"][0]
payload2 = make_cohort_payload(valid_location_id, count=1)
event2 = animal_service.create_cohort(payload2, ts_utc + 1000, "test_user")
alias_id = event2.entity_refs["animal_ids"][0]
# Set up survivor with nickname
seeded_db.execute(
"UPDATE animal_registry SET identified = 1, nickname = 'Daisy' WHERE animal_id = ?",
(survivor_id,),
)
# 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")
repo = AnimalTimelineRepository(seeded_db)
merge_info = repo.get_merge_info(alias_id)
assert merge_info is not None
assert isinstance(merge_info, MergeInfo)
assert merge_info.survivor_animal_id == survivor_id
assert merge_info.survivor_nickname == "Daisy"
assert merge_info.merged_at_utc == ts_utc + 2000
def test_returns_none_for_unmerged_animal(self, seeded_db, animal_service, valid_location_id):
"""get_merge_info returns None for animals that are not merged."""
ts_utc = int(time.time() * 1000)
payload = make_cohort_payload(valid_location_id, count=1)
event = animal_service.create_cohort(payload, ts_utc, "test_user")
animal_id = event.entity_refs["animal_ids"][0]
repo = AnimalTimelineRepository(seeded_db)
merge_info = repo.get_merge_info(animal_id)
assert merge_info is None
def test_returns_none_for_invalid_id(self, seeded_db):
"""get_merge_info returns None for non-existent animal."""
repo = AnimalTimelineRepository(seeded_db)
merge_info = repo.get_merge_info("00000000000000000000000000")
assert merge_info is None

View File

@@ -0,0 +1,353 @@
# 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