feat: add Promote Animal action route (Step 9.1 continued)

- Add /actions/promote/{animal_id} GET and /actions/animal-promote POST routes
- Add promote_form() template with nickname, sex, repro_status fields
- Add AnimalRepository.get() method for single-animal lookup
- 10 new tests for promote form rendering and submission

🤖 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-31 09:41:17 +00:00
parent f9e89fe5d6
commit 99f2fbb964
4 changed files with 460 additions and 3 deletions

View File

@@ -1,13 +1,18 @@
# ABOUTME: Repository for animal registry queries. # ABOUTME: Repository for animal registry queries.
# ABOUTME: Provides list_animals with filtering/pagination and facet counts. # ABOUTME: Provides list_animals with filtering/pagination and facet counts.
from __future__ import annotations
import base64 import base64
import json import json
from dataclasses import dataclass, field from dataclasses import dataclass, field
from typing import Any from typing import TYPE_CHECKING, Any
from animaltrack.selection.parser import parse_filter from animaltrack.selection.parser import parse_filter
if TYPE_CHECKING:
from animaltrack.models.animals import Animal
@dataclass @dataclass
class AnimalListItem: class AnimalListItem:
@@ -59,6 +64,49 @@ class AnimalRepository:
""" """
self.db = db self.db = db
def get(self, animal_id: str) -> Animal | None:
"""Get an animal by ID.
Args:
animal_id: The animal ID to look up.
Returns:
Animal model if found, None otherwise.
"""
from animaltrack.models.animals import Animal
row = self.db.execute(
"""
SELECT
animal_id, species_code, sex, life_stage, status,
location_id, repro_status, identified, nickname,
origin, born_or_hatched_at, acquired_at, first_seen_utc, last_event_utc
FROM animal_registry
WHERE animal_id = ?
""",
(animal_id,),
).fetchone()
if not row:
return None
return Animal(
animal_id=row[0],
species_code=row[1],
sex=row[2],
life_stage=row[3],
status=row[4],
location_id=row[5],
repro_status=row[6],
identified=bool(row[7]),
nickname=row[8],
origin=row[9],
born_or_hatched_at=row[10],
acquired_at=row[11],
first_seen_utc=row[12],
last_event_utc=row[13],
)
def list_animals( def list_animals(
self, self,
filter_str: str = "", filter_str: str = "",

View File

@@ -13,6 +13,7 @@ from starlette.responses import HTMLResponse
from animaltrack.events.payloads import ( from animaltrack.events.payloads import (
AnimalCohortCreatedPayload, AnimalCohortCreatedPayload,
AnimalPromotedPayload,
HatchRecordedPayload, HatchRecordedPayload,
) )
from animaltrack.events.store import EventStore from animaltrack.events.store import EventStore
@@ -20,11 +21,12 @@ from animaltrack.projections import EventLogProjection, ProjectionRegistry
from animaltrack.projections.animal_registry import AnimalRegistryProjection from animaltrack.projections.animal_registry import AnimalRegistryProjection
from animaltrack.projections.event_animals import EventAnimalsProjection from animaltrack.projections.event_animals import EventAnimalsProjection
from animaltrack.projections.intervals import IntervalProjection from animaltrack.projections.intervals import IntervalProjection
from animaltrack.repositories.animals import AnimalRepository
from animaltrack.repositories.locations import LocationRepository from animaltrack.repositories.locations import LocationRepository
from animaltrack.repositories.species import SpeciesRepository from animaltrack.repositories.species import SpeciesRepository
from animaltrack.services.animal import AnimalService, ValidationError from animaltrack.services.animal import AnimalService, ValidationError
from animaltrack.web.templates import page from animaltrack.web.templates import page
from animaltrack.web.templates.actions import cohort_form, hatch_form from animaltrack.web.templates.actions import cohort_form, hatch_form, promote_form
def _create_animal_service(db: Any) -> AnimalService: def _create_animal_service(db: Any) -> AnimalService:
@@ -317,6 +319,128 @@ def _render_hatch_error(
) )
# =============================================================================
# Animal Promotion
# =============================================================================
def promote_index(request: Request, animal_id: str):
"""GET /actions/promote/{animal_id} - Promote Animal form."""
db = request.app.state.db
# Look up the animal
animal_repo = AnimalRepository(db)
animal = animal_repo.get(animal_id)
if not animal:
return HTMLResponse(content="Animal not found", status_code=404)
if animal.status != "alive":
return HTMLResponse(content="Only alive animals can be promoted", status_code=400)
if animal.identified:
return HTMLResponse(content="Animal is already identified", status_code=400)
return page(
promote_form(animal),
title="Promote Animal - AnimalTrack",
active_nav=None,
)
async def animal_promote(request: Request):
"""POST /actions/animal-promote - Promote an animal to identified."""
db = request.app.state.db
form = await request.form()
# Extract form data
animal_id = form.get("animal_id", "")
nickname = form.get("nickname", "") or None
sex = form.get("sex", "") or None
repro_status = form.get("repro_status", "") or None
distinguishing_traits = form.get("distinguishing_traits", "") or None
notes = form.get("notes", "") or None
nonce = form.get("nonce")
# Validate animal_id
if not animal_id:
return HTMLResponse(content="Animal ID is required", status_code=422)
# Look up the animal
animal_repo = AnimalRepository(db)
animal = animal_repo.get(animal_id)
if not animal:
return HTMLResponse(content="Animal not found", status_code=404)
if animal.status != "alive":
return _render_promote_error(animal, "Only alive animals can be promoted", form)
if animal.identified:
return _render_promote_error(animal, "Animal is already identified", form)
# Create payload
try:
payload = AnimalPromotedPayload(
animal_id=animal_id,
nickname=nickname,
sex=sex,
repro_status=repro_status,
distinguishing_traits=distinguishing_traits,
notes=notes,
)
except Exception as e:
return _render_promote_error(animal, str(e), form)
# Get actor from auth
auth = request.scope.get("auth")
actor = auth.username if auth else "unknown"
# Promote animal
ts_utc = int(time.time() * 1000)
service = _create_animal_service(db)
try:
service.promote_animal(payload, ts_utc, actor, nonce=nonce, route="/actions/animal-promote")
except ValidationError as e:
return _render_promote_error(animal, str(e), form)
# Success: redirect to animal detail page
from starlette.responses import RedirectResponse
response = RedirectResponse(
url=f"/animals/{animal_id}",
status_code=303,
)
return response
def _render_promote_error(
animal: Any,
error_message: str,
form_data: Any = None,
) -> HTMLResponse:
"""Render promote form with error message."""
return HTMLResponse(
content=to_xml(
page(
promote_form(
animal,
error=error_message,
nickname_value=form_data.get("nickname", "") if form_data else "",
selected_sex=form_data.get("sex", "") if form_data else "",
selected_repro_status=form_data.get("repro_status", "") if form_data else "",
traits_value=form_data.get("distinguishing_traits", "") if form_data else "",
),
title="Promote Animal - AnimalTrack",
active_nav=None,
)
),
status_code=422,
)
# ============================================================================= # =============================================================================
# Route Registration # Route Registration
# ============================================================================= # =============================================================================
@@ -334,3 +458,7 @@ def register_action_routes(rt, app):
rt("/actions/animal-cohort", methods=["POST"])(animal_cohort) rt("/actions/animal-cohort", methods=["POST"])(animal_cohort)
rt("/actions/hatch")(hatch_index) rt("/actions/hatch")(hatch_index)
rt("/actions/hatch-recorded", methods=["POST"])(hatch_recorded) rt("/actions/hatch-recorded", methods=["POST"])(hatch_recorded)
# Single animal actions
rt("/actions/promote/{animal_id}")(promote_index)
rt("/actions/animal-promote", methods=["POST"])(animal_promote)

View File

@@ -5,9 +5,18 @@ from collections.abc import Callable
from typing import Any from typing import Any
from fasthtml.common import H2, Div, Form, Hidden, Option, P from fasthtml.common import H2, Div, Form, Hidden, Option, P
from monsterui.all import Alert, AlertT, Button, ButtonT, LabelInput, LabelSelect from monsterui.all import (
Alert,
AlertT,
Button,
ButtonT,
LabelInput,
LabelSelect,
LabelTextArea,
)
from ulid import ULID from ulid import ULID
from animaltrack.models.animals import Animal
from animaltrack.models.reference import Location, Species from animaltrack.models.reference import Location, Species
# ============================================================================= # =============================================================================
@@ -275,3 +284,114 @@ def hatch_form(
method="post", method="post",
cls="space-y-4", cls="space-y-4",
) )
# =============================================================================
# Animal Promotion Form
# =============================================================================
def promote_form(
animal: Animal,
error: str | None = None,
nickname_value: str = "",
selected_sex: str = "",
selected_repro_status: str = "",
traits_value: str = "",
action: Callable[..., Any] | str = "/actions/animal-promote",
) -> Form:
"""Create the Animal Promotion form.
Args:
animal: The animal to promote.
error: Optional error message to display.
nickname_value: Pre-filled nickname value.
selected_sex: Pre-selected sex (to refine).
selected_repro_status: Pre-selected repro status.
traits_value: Pre-filled distinguishing traits.
action: Route function or URL string for form submission.
Returns:
Form component for promoting an animal.
"""
display_id = f"{animal.animal_id[:8]}..."
# Build sex options (optional - can refine current value)
sexes = [
("", "Keep current"),
("female", "Female"),
("male", "Male"),
]
sex_options = []
for code, label in sexes:
sex_options.append(Option(label, value=code, selected=code == selected_sex))
# Build repro status options (optional)
repro_statuses = [
("", "Unknown"),
("breeding", "Breeding"),
("non_breeding", "Non-Breeding"),
("broody", "Broody"),
("molting", "Molting"),
]
repro_status_options = []
for code, label in repro_statuses:
repro_status_options.append(
Option(label, value=code, selected=code == selected_repro_status)
)
# Error display component
error_component = None
if error:
error_component = Alert(error, cls=AlertT.warning)
return Form(
H2("Promote Animal", cls="text-xl font-bold mb-4"),
# Animal info
Div(
P(f"Promoting: {display_id}", cls="text-sm text-stone-400"),
P(f"Species: {animal.species_code}, Sex: {animal.sex}", cls="text-sm text-stone-400"),
cls="p-3 bg-slate-800 rounded-md mb-4",
),
# Error message if present
error_component,
# Nickname input (optional)
LabelInput(
"Nickname (optional)",
id="nickname",
name="nickname",
value=nickname_value,
placeholder="Give this animal a name",
),
# Sex refinement dropdown (optional)
LabelSelect(
*sex_options,
label="Refine Sex (optional)",
id="sex",
name="sex",
),
# Repro status dropdown (optional)
LabelSelect(
*repro_status_options,
label="Reproductive Status",
id="repro_status",
name="repro_status",
),
# Distinguishing traits (optional)
LabelTextArea(
"Distinguishing Traits (optional)",
id="distinguishing_traits",
name="distinguishing_traits",
value=traits_value,
placeholder="Color markings, size, personality...",
),
# Hidden fields
Hidden(name="animal_id", value=animal.animal_id),
Hidden(name="nonce", value=str(ULID())),
# Submit button
Button("Promote to Identified", type="submit", cls=ButtonT.primary),
# Form submission via standard action/method (hx-boost handles AJAX)
action=action,
method="post",
cls="space-y-4",
)

View File

@@ -423,3 +423,164 @@ class TestHatchRecordingValidation:
) )
assert resp.status_code == 422 assert resp.status_code == 422
# =============================================================================
# Animal Promotion Tests
# =============================================================================
@pytest.fixture
def unpromoted_animal_id(seeded_db, client, location_strip1_id):
"""Create an unpromoted animal for testing."""
# Create a cohort with one animal
resp = client.post(
"/actions/animal-cohort",
data={
"species": "duck",
"location_id": location_strip1_id,
"count": "1",
"life_stage": "adult",
"sex": "female",
"origin": "purchased",
"nonce": "test-fixture-cohort-1",
},
)
assert resp.status_code == 200
# Get the animal ID
row = seeded_db.execute(
"SELECT animal_id FROM animal_registry WHERE identified = 0 ORDER BY animal_id DESC LIMIT 1"
).fetchone()
return row[0]
class TestPromoteFormRendering:
"""Tests for GET /actions/promote/{animal_id} form rendering."""
def test_promote_form_renders(self, client, unpromoted_animal_id):
"""GET /actions/promote/{animal_id} returns 200 with form elements."""
resp = client.get(f"/actions/promote/{unpromoted_animal_id}")
assert resp.status_code == 200
assert "Promote" in resp.text
def test_promote_form_shows_animal_info(self, client, unpromoted_animal_id):
"""Form shows animal information."""
resp = client.get(f"/actions/promote/{unpromoted_animal_id}")
assert resp.status_code == 200
# Should show animal ID (truncated)
assert unpromoted_animal_id[:8] in resp.text
def test_promote_form_has_nickname_field(self, client, unpromoted_animal_id):
"""Form has nickname input field."""
resp = client.get(f"/actions/promote/{unpromoted_animal_id}")
assert resp.status_code == 200
assert 'name="nickname"' in resp.text
def test_promote_form_has_sex_dropdown(self, client, unpromoted_animal_id):
"""Form has sex dropdown."""
resp = client.get(f"/actions/promote/{unpromoted_animal_id}")
assert resp.status_code == 200
assert 'name="sex"' in resp.text
def test_promote_form_not_found(self, client):
"""Non-existent animal returns 404."""
resp = client.get("/actions/promote/01ABCDEFGHIJKLMNOPQRSTUV00")
assert resp.status_code == 404
class TestPromoteSuccess:
"""Tests for successful POST /actions/animal-promote."""
def test_promote_creates_event(self, client, seeded_db, unpromoted_animal_id):
"""POST creates AnimalPromoted event when valid."""
resp = client.post(
"/actions/animal-promote",
data={
"animal_id": unpromoted_animal_id,
"nickname": "Daffy",
"nonce": "test-promote-nonce-1",
},
follow_redirects=False,
)
# Should redirect
assert resp.status_code == 303
# Verify event was created
event_row = seeded_db.execute(
"SELECT type FROM events WHERE type = 'AnimalPromoted' ORDER BY id DESC LIMIT 1"
).fetchone()
assert event_row is not None
assert event_row[0] == "AnimalPromoted"
def test_promote_sets_identified(self, client, seeded_db, unpromoted_animal_id):
"""POST sets identified flag on animal."""
resp = client.post(
"/actions/animal-promote",
data={
"animal_id": unpromoted_animal_id,
"nickname": "Huey",
"nonce": "test-promote-nonce-2",
},
follow_redirects=False,
)
assert resp.status_code == 303
# Verify identified is now true
row = seeded_db.execute(
"SELECT identified FROM animal_registry WHERE animal_id = ?",
(unpromoted_animal_id,),
).fetchone()
assert row[0] == 1 # identified = true
def test_promote_sets_nickname(self, client, seeded_db, unpromoted_animal_id):
"""POST sets nickname on animal."""
resp = client.post(
"/actions/animal-promote",
data={
"animal_id": unpromoted_animal_id,
"nickname": "Dewey",
"nonce": "test-promote-nonce-3",
},
follow_redirects=False,
)
assert resp.status_code == 303
# Verify nickname is set
row = seeded_db.execute(
"SELECT nickname FROM animal_registry WHERE animal_id = ?",
(unpromoted_animal_id,),
).fetchone()
assert row[0] == "Dewey"
class TestPromoteValidation:
"""Tests for validation errors in POST /actions/animal-promote."""
def test_promote_missing_animal_id_returns_422(self, client):
"""Missing animal_id returns 422."""
resp = client.post(
"/actions/animal-promote",
data={
"nickname": "Test",
"nonce": "test-promote-nonce-4",
},
)
assert resp.status_code == 422
def test_promote_not_found_returns_404(self, client):
"""Non-existent animal returns 404."""
resp = client.post(
"/actions/animal-promote",
data={
"animal_id": "01ABCDEFGHIJKLMNOPQRSTUV00",
"nickname": "Test",
"nonce": "test-promote-nonce-5",
},
)
assert resp.status_code == 404