diff --git a/PLAN.md b/PLAN.md index c3f8f3f..ad9dcd9 100644 --- a/PLAN.md +++ b/PLAN.md @@ -326,15 +326,15 @@ Check off items as completed. Each phase builds on the previous. - [x] **Commit checkpoint** ### Step 8.3: Animal Detail Drawer -- [ ] Create `web/routes/animals.py`: - - [ ] GET /animals/{animal_id} -- [ ] Create `web/templates/animal_detail.py`: - - [ ] Header summary - - [ ] Timeline (newest first) - - [ ] Quick actions -- [ ] Create `repositories/animal_timeline.py` -- [ ] Write tests: detail renders, timeline shows events, merge info shown -- [ ] **Commit checkpoint** +- [x] Create `web/routes/animals.py`: + - [x] GET /animals/{animal_id} +- [x] Create `web/templates/animal_detail.py`: + - [x] Header summary + - [x] Timeline (newest first) + - [x] Quick actions +- [x] Create `repositories/animal_timeline.py` +- [x] Write tests: detail renders, timeline shows events, merge info shown +- [x] **Commit checkpoint** --- diff --git a/src/animaltrack/repositories/animal_timeline.py b/src/animaltrack/repositories/animal_timeline.py new file mode 100644 index 0000000..8a19b24 --- /dev/null +++ b/src/animaltrack/repositories/animal_timeline.py @@ -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], + ) diff --git a/src/animaltrack/web/app.py b/src/animaltrack/web/app.py index bae7b0b..179dd8f 100644 --- a/src/animaltrack/web/app.py +++ b/src/animaltrack/web/app.py @@ -18,6 +18,7 @@ from animaltrack.web.middleware import ( request_id_before, ) from animaltrack.web.routes import ( + register_animals_routes, register_egg_routes, register_events_routes, register_feed_routes, @@ -132,6 +133,7 @@ def create_app( # Register routes register_health_routes(rt, app) + register_animals_routes(rt, app) register_egg_routes(rt, app) register_events_routes(rt, app) register_feed_routes(rt, app) diff --git a/src/animaltrack/web/routes/__init__.py b/src/animaltrack/web/routes/__init__.py index 0e4e96b..3090247 100644 --- a/src/animaltrack/web/routes/__init__.py +++ b/src/animaltrack/web/routes/__init__.py @@ -1,6 +1,7 @@ # ABOUTME: Routes package for AnimalTrack web application. # 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.events import register_events_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 __all__ = [ + "register_animals_routes", "register_egg_routes", "register_events_routes", "register_feed_routes", diff --git a/src/animaltrack/web/routes/animals.py b/src/animaltrack/web/routes/animals.py new file mode 100644 index 0000000..6349932 --- /dev/null +++ b/src/animaltrack/web/routes/animals.py @@ -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="

Animal not found

", + 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) diff --git a/src/animaltrack/web/templates/animal_detail.py b/src/animaltrack/web/templates/animal_detail.py new file mode 100644 index 0000000..932e2cd --- /dev/null +++ b/src/animaltrack/web/templates/animal_detail.py @@ -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 diff --git a/tests/test_repository_animal_timeline.py b/tests/test_repository_animal_timeline.py new file mode 100644 index 0000000..9280f32 --- /dev/null +++ b/tests/test_repository_animal_timeline.py @@ -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 diff --git a/tests/test_web_animal_detail.py b/tests/test_web_animal_detail.py new file mode 100644 index 0000000..aed22e2 --- /dev/null +++ b/tests/test_web_animal_detail.py @@ -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