Alo/animaltrack
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: complete Step 9.1 with outcome, status-correct, and quick actions

Browse Source 
- Add animal-outcome route with yield items section for harvest products
- Add animal-status-correct route with @require_role(ADMIN) decorator
- Add exception handlers for AuthenticationError (401) and AuthorizationError (403)
- Enable quick action buttons in animal detail page (Add Tag, Promote, Record Outcome)
- Add comprehensive tests for outcome and status-correct routes (81 total action tests)

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
Petru
2025-12-31 13:45:06 +00:00
parent 3acb731a6c
commit 29ea3e27cb
5 changed files with 2036 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
src/animaltrack/web/app.py
View File

@@ -9,9 +9,11 @@ from fasthtml.common import Beforeware, fast_app
from monsterui.all import Theme
from starlette.middleware import Middleware
from starlette.requests import Request
from starlette.responses import PlainTextResponse
from animaltrack.config import Settings
from animaltrack.db import get_db
from animaltrack.web.exceptions import AuthenticationError, AuthorizationError
from animaltrack.web.middleware import (
auth_before,
csrf_before,
@@ -132,6 +134,16 @@ def create_app(
app.state.settings = settings
app.state.db = db
# Register exception handlers for auth errors
async def authentication_error_handler(request, exc):
return PlainTextResponse(str(exc) or "Authentication required", status_code=401)
async def authorization_error_handler(request, exc):
return PlainTextResponse(str(exc) or "Forbidden", status_code=403)
app.add_exception_handler(AuthenticationError, authentication_error_handler)
app.add_exception_handler(AuthorizationError, authorization_error_handler)
# Register routes
register_health_routes(rt, app)
register_action_routes(rt, app)

725
src/animaltrack/web/routes/actions.py
View File

@@ -11,12 +11,18 @@ from fasthtml.common import to_xml
from starlette.requests import Request
from starlette.responses import HTMLResponse
from animaltrack.events.enums import AnimalStatus, Outcome
from animaltrack.events.payloads import (
AnimalAttributesUpdatedPayload,
AnimalCohortCreatedPayload,
AnimalOutcomePayload,
AnimalPromotedPayload,
AnimalStatusCorrectedPayload,
AnimalTagEndedPayload,
AnimalTaggedPayload,
AttributeSet,
HatchRecordedPayload,
YieldItem,
)
from animaltrack.events.store import EventStore
from animaltrack.projections import EventLogProjection, ProjectionRegistry
@@ -26,15 +32,23 @@ from animaltrack.projections.intervals import IntervalProjection
from animaltrack.projections.tags import TagProjection
from animaltrack.repositories.animals import AnimalRepository
from animaltrack.repositories.locations import LocationRepository
from animaltrack.repositories.products import ProductRepository
from animaltrack.repositories.species import SpeciesRepository
from animaltrack.selection import compute_roster_hash, parse_filter, resolve_filter
from animaltrack.selection.validation import SelectionContext, validate_selection
from animaltrack.services.animal import AnimalService, ValidationError
from animaltrack.web.auth import UserRole, require_role
from animaltrack.web.templates import page
from animaltrack.web.templates.actions import (
attrs_diff_panel,
attrs_form,
cohort_form,
hatch_form,
outcome_diff_panel,
outcome_form,
promote_form,
status_correct_diff_panel,
status_correct_form,
tag_add_diff_panel,
tag_add_form,
tag_end_diff_panel,
@@ -885,6 +899,709 @@ def _render_tag_end_error_form(db, filter_str, error_message):
)
# =============================================================================
# Update Attributes
# =============================================================================
def attrs_index(request: Request):
"""GET /actions/attrs - Update Attributes form."""
db = request.app.state.db
# Get filter from query params
filter_str = request.query_params.get("filter", "")
# Resolve selection if filter provided
ts_utc = int(time.time() * 1000)
resolved_ids: list[str] = []
roster_hash = ""
if filter_str:
filter_ast = parse_filter(filter_str)
resolution = resolve_filter(db, filter_ast, ts_utc)
resolved_ids = resolution.animal_ids
if resolved_ids:
roster_hash = compute_roster_hash(resolved_ids, None)
return page(
attrs_form(
filter_str=filter_str,
resolved_ids=resolved_ids,
roster_hash=roster_hash,
ts_utc=ts_utc,
resolved_count=len(resolved_ids),
),
title="Update Attributes - AnimalTrack",
active_nav=None,
)
async def animal_attrs(request: Request):
"""POST /actions/animal-attrs - Update attributes on animals."""
db = request.app.state.db
form = await request.form()
# Extract form data
filter_str = form.get("filter", "")
sex = form.get("sex", "").strip() or None
life_stage = form.get("life_stage", "").strip() or None
repro_status = form.get("repro_status", "").strip() or None
roster_hash = form.get("roster_hash", "")
confirmed = form.get("confirmed", "") == "true"
nonce = form.get("nonce")
# Get timestamp - use provided or current
ts_utc_str = form.get("ts_utc", "0")
try:
ts_utc = int(ts_utc_str)
if ts_utc == 0:
ts_utc = int(time.time() * 1000)
except ValueError:
ts_utc = int(time.time() * 1000)
# resolved_ids can be multiple values
resolved_ids = form.getlist("resolved_ids")
# Validation: at least one attribute required
if not sex and not life_stage and not repro_status:
return _render_attrs_error_form(
db, filter_str, "Please select at least one attribute to update"
)
# Validation: must have animals
if not resolved_ids:
return _render_attrs_error_form(db, filter_str, "No animals selected")
# Build selection context for validation
context = SelectionContext(
filter=filter_str,
resolved_ids=list(resolved_ids),
roster_hash=roster_hash,
ts_utc=ts_utc,
from_location_id=None,
confirmed=confirmed,
)
# Validate selection (check for concurrent changes)
result = validate_selection(db, context)
if not result.valid:
# Mismatch detected - return 409 with diff panel
return HTMLResponse(
content=to_xml(
page(
attrs_diff_panel(
diff=result.diff,
filter_str=filter_str,
resolved_ids=result.resolved_ids,
roster_hash=result.roster_hash,
sex=sex,
life_stage=life_stage,
repro_status=repro_status,
ts_utc=ts_utc,
),
title="Update Attributes - AnimalTrack",
active_nav=None,
)
),
status_code=409,
)
# When confirmed, re-resolve to get current server IDs
if confirmed:
current_ts = max(int(time.time() * 1000), ts_utc)
filter_ast = parse_filter(filter_str)
current_resolution = resolve_filter(db, filter_ast, current_ts)
ids_to_update = current_resolution.animal_ids
else:
ids_to_update = resolved_ids
# Check we still have animals
if not ids_to_update:
return _render_attrs_error_form(db, filter_str, "No animals remaining")
# Create payload
try:
attr_set = AttributeSet(
sex=sex,
life_stage=life_stage,
repro_status=repro_status,
)
payload = AnimalAttributesUpdatedPayload(
resolved_ids=list(ids_to_update),
set=attr_set,
)
except Exception as e:
return _render_attrs_error_form(db, filter_str, str(e))
# Get actor from auth
auth = request.scope.get("auth")
actor = auth.username if auth else "unknown"
# Update attributes
service = _create_animal_service(db)
try:
event = service.update_attributes(
payload, ts_utc, actor, nonce=nonce, route="/actions/animal-attrs"
)
except ValidationError as e:
return _render_attrs_error_form(db, filter_str, str(e))
# Success: re-render fresh form
response = HTMLResponse(
content=to_xml(
page(
attrs_form(),
title="Update Attributes - AnimalTrack",
active_nav=None,
)
),
)
# Add toast trigger header
updated_count = len(event.entity_refs.get("animal_ids", []))
response.headers["HX-Trigger"] = json.dumps(
{
"showToast": {
"message": f"Updated attributes on {updated_count} animal(s)",
"type": "success",
}
}
)
return response
def _render_attrs_error_form(db, filter_str, error_message):
"""Render attributes form with error message."""
# Re-resolve to show current selection info
ts_utc = int(time.time() * 1000)
resolved_ids: list[str] = []
roster_hash = ""
if filter_str:
filter_ast = parse_filter(filter_str)
resolution = resolve_filter(db, filter_ast, ts_utc)
resolved_ids = resolution.animal_ids
if resolved_ids:
roster_hash = compute_roster_hash(resolved_ids, None)
return HTMLResponse(
content=to_xml(
page(
attrs_form(
filter_str=filter_str,
resolved_ids=resolved_ids,
roster_hash=roster_hash,
ts_utc=ts_utc,
resolved_count=len(resolved_ids),
error=error_message,
),
title="Update Attributes - AnimalTrack",
active_nav=None,
)
),
status_code=422,
)
# =============================================================================
# Record Outcome
# =============================================================================
def outcome_index(request: Request):
"""GET /actions/outcome - Record Outcome form."""
db = request.app.state.db
# Get filter from query params
filter_str = request.query_params.get("filter", "")
# Resolve selection if filter provided
ts_utc = int(time.time() * 1000)
resolved_ids: list[str] = []
roster_hash = ""
if filter_str:
filter_ast = parse_filter(filter_str)
resolution = resolve_filter(db, filter_ast, ts_utc)
resolved_ids = resolution.animal_ids
if resolved_ids:
roster_hash = compute_roster_hash(resolved_ids, None)
# Get active products for yield items dropdown
product_repo = ProductRepository(db)
products = [(p.code, p.name) for p in product_repo.list_all() if p.active]
return page(
outcome_form(
filter_str=filter_str,
resolved_ids=resolved_ids,
roster_hash=roster_hash,
ts_utc=ts_utc,
resolved_count=len(resolved_ids),
products=products,
),
title="Record Outcome - AnimalTrack",
active_nav=None,
)
async def animal_outcome(request: Request):
"""POST /actions/animal-outcome - Record outcome for animals."""
db = request.app.state.db
form = await request.form()
# Extract form data
filter_str = form.get("filter", "")
outcome_str = form.get("outcome", "").strip()
reason = form.get("reason", "").strip() or None
notes = form.get("notes", "").strip() or None
roster_hash = form.get("roster_hash", "")
confirmed = form.get("confirmed", "") == "true"
nonce = form.get("nonce")
# Yield item fields
yield_product_code = form.get("yield_product_code", "").strip() or None
yield_unit = form.get("yield_unit", "").strip() or None
yield_quantity_str = form.get("yield_quantity", "").strip()
yield_weight_str = form.get("yield_weight_kg", "").strip()
yield_quantity: int | None = None
yield_weight_kg: float | None = None
if yield_quantity_str:
try:
yield_quantity = int(yield_quantity_str)
except ValueError:
pass
if yield_weight_str:
try:
yield_weight_kg = float(yield_weight_str)
except ValueError:
pass
# Get timestamp - use provided or current
ts_utc_str = form.get("ts_utc", "0")
try:
ts_utc = int(ts_utc_str)
if ts_utc == 0:
ts_utc = int(time.time() * 1000)
except ValueError:
ts_utc = int(time.time() * 1000)
# resolved_ids can be multiple values
resolved_ids = form.getlist("resolved_ids")
# Validation: outcome required
if not outcome_str:
return _render_outcome_error_form(db, filter_str, "Please select an outcome")
# Validate outcome is valid enum value
try:
outcome_enum = Outcome(outcome_str)
except ValueError:
return _render_outcome_error_form(db, filter_str, f"Invalid outcome: {outcome_str}")
# Validation: must have animals
if not resolved_ids:
return _render_outcome_error_form(db, filter_str, "No animals selected")
# Build selection context for validation
context = SelectionContext(
filter=filter_str,
resolved_ids=list(resolved_ids),
roster_hash=roster_hash,
ts_utc=ts_utc,
from_location_id=None,
confirmed=confirmed,
)
# Validate selection (check for concurrent changes)
result = validate_selection(db, context)
if not result.valid:
# Mismatch detected - return 409 with diff panel
return HTMLResponse(
content=to_xml(
page(
outcome_diff_panel(
diff=result.diff,
filter_str=filter_str,
resolved_ids=result.resolved_ids,
roster_hash=result.roster_hash,
outcome=outcome_str,
reason=reason,
yield_product_code=yield_product_code,
yield_unit=yield_unit,
yield_quantity=yield_quantity,
yield_weight_kg=yield_weight_kg,
ts_utc=ts_utc,
),
title="Record Outcome - AnimalTrack",
active_nav=None,
)
),
status_code=409,
)
# When confirmed, re-resolve to get current server IDs
if confirmed:
current_ts = max(int(time.time() * 1000), ts_utc)
filter_ast = parse_filter(filter_str)
current_resolution = resolve_filter(db, filter_ast, current_ts)
ids_to_update = current_resolution.animal_ids
else:
ids_to_update = resolved_ids
# Check we still have animals
if not ids_to_update:
return _render_outcome_error_form(db, filter_str, "No animals remaining")
# Build yield items if provided
yield_items: list[YieldItem] | None = None
if yield_product_code and yield_quantity and yield_quantity > 0:
try:
yield_items = [
YieldItem(
product_code=yield_product_code,
unit=yield_unit or "piece",
quantity=yield_quantity,
weight_kg=yield_weight_kg,
)
]
except Exception:
# Invalid yield item - ignore
yield_items = None
# Create payload
try:
payload = AnimalOutcomePayload(
resolved_ids=list(ids_to_update),
outcome=outcome_enum,
reason=reason,
yield_items=yield_items,
notes=notes,
)
except Exception as e:
return _render_outcome_error_form(db, filter_str, str(e))
# Get actor from auth
auth = request.scope.get("auth")
actor = auth.username if auth else "unknown"
# Record outcome
service = _create_animal_service(db)
try:
event = service.record_outcome(
payload, ts_utc, actor, nonce=nonce, route="/actions/animal-outcome"
)
except ValidationError as e:
return _render_outcome_error_form(db, filter_str, str(e))
# Success: re-render fresh form
product_repo = ProductRepository(db)
products = [(p.code, p.name) for p in product_repo.list_all() if p.active]
response = HTMLResponse(
content=to_xml(
page(
outcome_form(
filter_str="",
resolved_ids=[],
roster_hash="",
ts_utc=int(time.time() * 1000),
resolved_count=0,
products=products,
),
title="Record Outcome - AnimalTrack",
active_nav=None,
)
),
)
# Add toast trigger header
outcome_count = len(event.entity_refs.get("animal_ids", []))
response.headers["HX-Trigger"] = json.dumps(
{
"showToast": {
"message": f"Recorded {outcome_str} for {outcome_count} animal(s)",
"type": "success",
}
}
)
return response
def _render_outcome_error_form(db, filter_str, error_message):
"""Render outcome form with error message."""
# Re-resolve to show current selection info
ts_utc = int(time.time() * 1000)
resolved_ids: list[str] = []
roster_hash = ""
if filter_str:
filter_ast = parse_filter(filter_str)
resolution = resolve_filter(db, filter_ast, ts_utc)
resolved_ids = resolution.animal_ids
if resolved_ids:
roster_hash = compute_roster_hash(resolved_ids, None)
# Get active products for yield items dropdown
product_repo = ProductRepository(db)
products = [(p.code, p.name) for p in product_repo.list_all() if p.active]
return HTMLResponse(
content=to_xml(
page(
outcome_form(
filter_str=filter_str,
resolved_ids=resolved_ids,
roster_hash=roster_hash,
ts_utc=ts_utc,
resolved_count=len(resolved_ids),
products=products,
error=error_message,
),
title="Record Outcome - AnimalTrack",
active_nav=None,
)
),
status_code=422,
)
# =============================================================================
# Correct Status (Admin-Only)
# =============================================================================
@require_role(UserRole.ADMIN)
async def status_correct_index(req: Request):
"""GET /actions/status-correct - Correct Status form (admin-only)."""
db = req.app.state.db
# Get filter from query params
filter_str = req.query_params.get("filter", "")
# Resolve selection if filter provided
ts_utc = int(time.time() * 1000)
resolved_ids: list[str] = []
roster_hash = ""
if filter_str:
filter_ast = parse_filter(filter_str)
resolution = resolve_filter(db, filter_ast, ts_utc)
resolved_ids = resolution.animal_ids
if resolved_ids:
roster_hash = compute_roster_hash(resolved_ids, None)
return page(
status_correct_form(
filter_str=filter_str,
resolved_ids=resolved_ids,
roster_hash=roster_hash,
ts_utc=ts_utc,
resolved_count=len(resolved_ids),
),
title="Correct Status - AnimalTrack",
active_nav=None,
)
@require_role(UserRole.ADMIN)
async def animal_status_correct(req: Request):
"""POST /actions/animal-status-correct - Correct status of animals (admin-only)."""
db = req.app.state.db
form = await req.form()
# Extract form data
filter_str = form.get("filter", "")
new_status_str = form.get("new_status", "").strip()
reason = form.get("reason", "").strip()
notes = form.get("notes", "").strip() or None
roster_hash = form.get("roster_hash", "")
confirmed = form.get("confirmed", "") == "true"
nonce = form.get("nonce")
# Get timestamp - use provided or current
ts_utc_str = form.get("ts_utc", "0")
try:
ts_utc = int(ts_utc_str)
if ts_utc == 0:
ts_utc = int(time.time() * 1000)
except ValueError:
ts_utc = int(time.time() * 1000)
# resolved_ids can be multiple values
resolved_ids = form.getlist("resolved_ids")
# Validation: new_status required
if not new_status_str:
return _render_status_correct_error_form(db, filter_str, "Please select a new status")
# Validate status is valid enum value
try:
new_status_enum = AnimalStatus(new_status_str)
except ValueError:
return _render_status_correct_error_form(
db, filter_str, f"Invalid status: {new_status_str}"
)
# Validation: reason required for admin actions
if not reason:
return _render_status_correct_error_form(db, filter_str, "Reason is required")
# Validation: must have animals
if not resolved_ids:
return _render_status_correct_error_form(db, filter_str, "No animals selected")
# Build selection context for validation
context = SelectionContext(
filter=filter_str,
resolved_ids=list(resolved_ids),
roster_hash=roster_hash,
ts_utc=ts_utc,
from_location_id=None,
confirmed=confirmed,
)
# Validate selection (check for concurrent changes)
result = validate_selection(db, context)
if not result.valid:
# Mismatch detected - return 409 with diff panel
return HTMLResponse(
content=to_xml(
page(
status_correct_diff_panel(
diff=result.diff,
filter_str=filter_str,
resolved_ids=result.resolved_ids,
roster_hash=result.roster_hash,
new_status=new_status_str,
reason=reason,
ts_utc=ts_utc,
),
title="Correct Status - AnimalTrack",
active_nav=None,
)
),
status_code=409,
)
# When confirmed, re-resolve to get current server IDs
if confirmed:
current_ts = max(int(time.time() * 1000), ts_utc)
filter_ast = parse_filter(filter_str)
current_resolution = resolve_filter(db, filter_ast, current_ts)
ids_to_update = current_resolution.animal_ids
else:
ids_to_update = resolved_ids
# Check we still have animals
if not ids_to_update:
return _render_status_correct_error_form(db, filter_str, "No animals remaining")
# Create payload
try:
payload = AnimalStatusCorrectedPayload(
resolved_ids=list(ids_to_update),
new_status=new_status_enum,
reason=reason,
notes=notes,
)
except Exception as e:
return _render_status_correct_error_form(db, filter_str, str(e))
# Get actor from auth
auth = req.scope.get("auth")
actor = auth.username if auth else "unknown"
# Correct status
service = _create_animal_service(db)
try:
event = service.correct_status(
payload, ts_utc, actor, nonce=nonce, route="/actions/animal-status-correct"
)
except ValidationError as e:
return _render_status_correct_error_form(db, filter_str, str(e))
# Success: re-render fresh form
response = HTMLResponse(
content=to_xml(
page(
status_correct_form(
filter_str="",
resolved_ids=[],
roster_hash="",
ts_utc=int(time.time() * 1000),
resolved_count=0,
),
title="Correct Status - AnimalTrack",
active_nav=None,
)
),
)
# Add toast trigger header
corrected_count = len(event.entity_refs.get("animal_ids", []))
response.headers["HX-Trigger"] = json.dumps(
{
"showToast": {
"message": f"Corrected status to {new_status_str} for {corrected_count} animal(s)",
"type": "success",
}
}
)
return response
def _render_status_correct_error_form(db, filter_str, error_message):
"""Render status correct form with error message."""
# Re-resolve to show current selection info
ts_utc = int(time.time() * 1000)
resolved_ids: list[str] = []
roster_hash = ""
if filter_str:
filter_ast = parse_filter(filter_str)
resolution = resolve_filter(db, filter_ast, ts_utc)
resolved_ids = resolution.animal_ids
if resolved_ids:
roster_hash = compute_roster_hash(resolved_ids, None)
return HTMLResponse(
content=to_xml(
page(
status_correct_form(
filter_str=filter_str,
resolved_ids=resolved_ids,
roster_hash=roster_hash,
ts_utc=ts_utc,
resolved_count=len(resolved_ids),
error=error_message,
),
title="Correct Status - AnimalTrack",
active_nav=None,
)
),
status_code=422,
)
# =============================================================================
# Route Registration
# =============================================================================
@@ -912,3 +1629,11 @@ def register_action_routes(rt, app):
rt("/actions/animal-tag-add", methods=["POST"])(animal_tag_add)
rt("/actions/tag-end")(tag_end_index)
rt("/actions/animal-tag-end", methods=["POST"])(animal_tag_end)
rt("/actions/attrs")(attrs_index)
rt("/actions/animal-attrs", methods=["POST"])(animal_attrs)
rt("/actions/outcome")(outcome_index)
rt("/actions/animal-outcome", methods=["POST"])(animal_outcome)
# Admin-only actions
rt("/actions/status-correct")(status_correct_index)
rt("/actions/animal-status-correct", methods=["POST"])(animal_status_correct)

704
src/animaltrack/web/templates/actions.py
View File

@@ -4,7 +4,7 @@
from collections.abc import Callable
from typing import Any
from fasthtml.common import H2, Div, Form, Hidden, Option, P, Span
from fasthtml.common import H2, H3, Div, Form, Hidden, Option, P, Span
from monsterui.all import (
Alert,
AlertT,
@@ -758,3 +758,705 @@ def tag_end_diff_panel(
confirm_form,
cls="space-y-4",
)
# =============================================================================
# Update Attributes Form
# =============================================================================
def attrs_form(
filter_str: str = "",
resolved_ids: list[str] | None = None,
roster_hash: str = "",
ts_utc: int | None = None,
resolved_count: int = 0,
error: str | None = None,
action: Callable[..., Any] | str = "/actions/animal-attrs",
) -> Form:
"""Create the Update Attributes form.
Args:
filter_str: Current filter string (DSL).
resolved_ids: Resolved animal IDs from filter.
roster_hash: Hash of resolved selection.
ts_utc: Timestamp of resolution.
resolved_count: Number of resolved animals.
error: Optional error message to display.
action: Route function or URL string for form submission.
Returns:
Form component for updating animal attributes.
"""
if resolved_ids is None:
resolved_ids = []
# Error display component
error_component = None
if error:
error_component = Alert(error, cls=AlertT.warning)
# Selection preview component
selection_preview = None
if resolved_count > 0:
selection_preview = Div(
P(
Span(f"{resolved_count}", cls="font-bold text-lg"),
" animals selected",
cls="text-sm",
),
cls="p-3 bg-slate-100 dark:bg-slate-800 rounded-md mb-4",
)
elif filter_str:
selection_preview = Div(
P("No animals match this filter", cls="text-sm text-amber-600"),
cls="p-3 bg-amber-50 dark:bg-amber-900/20 rounded-md mb-4",
)
# Build sex options
sex_options = [
Option("No change", value="", selected=True),
Option("Female", value="female"),
Option("Male", value="male"),
Option("Unknown", value="unknown"),
]
# Build life stage options
life_stage_options = [
Option("No change", value="", selected=True),
Option("Hatchling", value="hatchling"),
Option("Juvenile", value="juvenile"),
Option("Subadult", value="subadult"),
Option("Adult", value="adult"),
]
# Build repro status options (intact, wether, spayed, unknown)
repro_status_options = [
Option("No change", value="", selected=True),
Option("Intact", value="intact"),
Option("Wether (castrated male)", value="wether"),
Option("Spayed (female)", value="spayed"),
Option("Unknown", value="unknown"),
]
# Hidden fields for resolved_ids (as multiple values)
resolved_id_fields = [
Hidden(name="resolved_ids", value=animal_id) for animal_id in resolved_ids
]
return Form(
H2("Update Attributes", cls="text-xl font-bold mb-4"),
# Error message if present
error_component,
# Filter input
LabelInput(
"Filter",
id="filter",
name="filter",
value=filter_str,
placeholder="e.g., species:duck life_stage:juvenile",
),
# Selection preview
selection_preview,
# Attribute dropdowns
LabelSelect(
*sex_options,
label="Sex",
id="sex",
name="sex",
),
LabelSelect(
*life_stage_options,
label="Life Stage",
id="life_stage",
name="life_stage",
),
LabelSelect(
*repro_status_options,
label="Reproductive Status",
id="repro_status",
name="repro_status",
),
# Hidden fields for selection context
*resolved_id_fields,
Hidden(name="roster_hash", value=roster_hash),
Hidden(name="ts_utc", value=str(ts_utc or 0)),
Hidden(name="confirmed", value=""),
Hidden(name="nonce", value=str(ULID())),
# Submit button
Button("Update Attributes", type="submit", cls=ButtonT.primary),
# Form submission via standard action/method (hx-boost handles AJAX)
action=action,
method="post",
cls="space-y-4",
)
def attrs_diff_panel(
diff: SelectionDiff,
filter_str: str,
resolved_ids: list[str],
roster_hash: str,
sex: str | None,
life_stage: str | None,
repro_status: str | None,
ts_utc: int,
action: Callable[..., Any] | str = "/actions/animal-attrs",
) -> Div:
"""Create the mismatch confirmation panel for attributes update.
Args:
diff: SelectionDiff with added/removed counts.
filter_str: Original filter string.
resolved_ids: Server's resolved IDs (current).
roster_hash: Server's roster hash (current).
sex: Sex to set (or None).
life_stage: Life stage to set (or None).
repro_status: Repro status to set (or None).
ts_utc: Timestamp for resolution.
action: Route function or URL for confirmation submit.
Returns:
Div containing the diff panel with confirm button.
"""
# Build description of changes
changes = []
if diff.removed:
changes.append(f"{len(diff.removed)} animals were removed since you loaded this page")
if diff.added:
changes.append(f"{len(diff.added)} animals were added")
changes_text = ". ".join(changes) + "." if changes else "The selection has changed."
# Build confirmation form with hidden fields
resolved_id_fields = [
Hidden(name="resolved_ids", value=animal_id) for animal_id in resolved_ids
]
confirm_form = Form(
*resolved_id_fields,
Hidden(name="filter", value=filter_str),
Hidden(name="roster_hash", value=roster_hash),
Hidden(name="sex", value=sex or ""),
Hidden(name="life_stage", value=life_stage or ""),
Hidden(name="repro_status", value=repro_status or ""),
Hidden(name="ts_utc", value=str(ts_utc)),
Hidden(name="confirmed", value="true"),
Hidden(name="nonce", value=str(ULID())),
Div(
Button(
"Cancel",
type="button",
cls=ButtonT.default,
onclick="window.location.href='/actions/attrs'",
),
Button(
f"Confirm Update ({diff.server_count} animals)",
type="submit",
cls=ButtonT.primary,
),
cls="flex gap-3 mt-4",
),
action=action,
method="post",
)
return Div(
Alert(
Div(
P("Selection Changed", cls="font-bold text-lg mb-2"),
P(changes_text, cls="mb-2"),
P(
f"Would you like to proceed with updating {diff.server_count} animals?",
cls="text-sm",
),
),
cls=AlertT.warning,
),
confirm_form,
cls="space-y-4",
)
# =============================================================================
# Record Outcome Form
# =============================================================================
def outcome_form(
filter_str: str = "",
resolved_ids: list[str] | None = None,
roster_hash: str = "",
ts_utc: int | None = None,
resolved_count: int = 0,
products: list[tuple[str, str]] | None = None,
error: str | None = None,
action: Callable[..., Any] | str = "/actions/animal-outcome",
) -> Form:
"""Create the Record Outcome form.
Args:
filter_str: Current filter string (DSL).
resolved_ids: Resolved animal IDs from filter.
roster_hash: Hash of resolved selection.
ts_utc: Timestamp of resolution.
resolved_count: Number of resolved animals.
products: List of (code, name) tuples for product dropdown.
error: Optional error message to display.
action: Route function or URL string for form submission.
Returns:
Form component for recording animal outcomes.
"""
if resolved_ids is None:
resolved_ids = []
if products is None:
products = []
# Error display component
error_component = None
if error:
error_component = Alert(error, cls=AlertT.warning)
# Selection preview component
selection_preview = None
if resolved_count > 0:
selection_preview = Div(
P(
Span(f"{resolved_count}", cls="font-bold text-lg"),
" animals selected",
cls="text-sm",
),
cls="p-3 bg-slate-100 dark:bg-slate-800 rounded-md mb-4",
)
elif filter_str:
selection_preview = Div(
P("No animals match this filter", cls="text-sm text-amber-600"),
cls="p-3 bg-amber-50 dark:bg-amber-900/20 rounded-md mb-4",
)
# Build outcome options
outcome_options = [
Option("Select outcome...", value="", selected=True, disabled=True),
Option("Death (natural)", value="death"),
Option("Harvest", value="harvest"),
Option("Sold", value="sold"),
Option("Predator Loss", value="predator_loss"),
Option("Unknown", value="unknown"),
]
# Build product options for yield items
product_options = [Option("Select product...", value="", selected=True)]
for code, name in products:
product_options.append(Option(f"{name} ({code})", value=code))
# Unit options for yield items
unit_options = [
Option("Piece", value="piece", selected=True),
Option("Kg", value="kg"),
]
# Hidden fields for resolved_ids (as multiple values)
resolved_id_fields = [
Hidden(name="resolved_ids", value=animal_id) for animal_id in resolved_ids
]
# Yield items section (for harvest - visible via CSS/JS, but we include it always)
yield_section = Div(
H3("Yield Items", cls="text-lg font-semibold mt-4 mb-2"),
P("Optional: record products collected from harvest", cls="text-sm text-stone-500 mb-3"),
Div(
LabelSelect(
*product_options,
label="Product",
id="yield_product_code",
name="yield_product_code",
cls="flex-1",
),
LabelSelect(
*unit_options,
label="Unit",
id="yield_unit",
name="yield_unit",
cls="w-32",
),
cls="flex gap-3",
),
Div(
LabelInput(
label="Quantity",
id="yield_quantity",
name="yield_quantity",
type="number",
min="1",
placeholder="1",
cls="flex-1",
),
LabelInput(
label="Weight (kg)",
id="yield_weight_kg",
name="yield_weight_kg",
type="number",
step="0.001",
placeholder="Optional",
cls="flex-1",
),
cls="flex gap-3",
),
id="yield-section",
cls="p-3 bg-slate-100 dark:bg-slate-800 rounded-md space-y-3",
)
return Form(
H2("Record Outcome", cls="text-xl font-bold mb-4"),
error_component,
selection_preview,
# Filter field
LabelInput(
label="Filter (DSL)",
id="filter",
name="filter",
value=filter_str,
placeholder="e.g., species:duck location:Coop1",
),
# Outcome selection
LabelSelect(
*outcome_options,
label="Outcome",
id="outcome",
name="outcome",
required=True,
),
# Reason field
LabelInput(
label="Reason (optional)",
id="reason",
name="reason",
placeholder="e.g., old age, processing day",
),
# Yield items section
yield_section,
# Notes field
LabelTextArea(
label="Notes (optional)",
id="notes",
name="notes",
rows=2,
placeholder="Any additional notes...",
),
# Hidden fields for selection context
*resolved_id_fields,
Hidden(name="roster_hash", value=roster_hash),
Hidden(name="ts_utc", value=str(ts_utc or 0)),
Hidden(name="confirmed", value=""),
Hidden(name="nonce", value=str(ULID())),
# Submit button
Button("Record Outcome", type="submit", cls=ButtonT.destructive),
# Form submission via standard action/method (hx-boost handles AJAX)
action=action,
method="post",
cls="space-y-4",
)
def outcome_diff_panel(
diff: SelectionDiff,
filter_str: str,
resolved_ids: list[str],
roster_hash: str,
outcome: str,
reason: str | None,
yield_product_code: str | None,
yield_unit: str | None,
yield_quantity: int | None,
yield_weight_kg: float | None,
ts_utc: int,
action: Callable[..., Any] | str = "/actions/animal-outcome",
) -> Div:
"""Create the mismatch confirmation panel for outcome recording.
Args:
diff: SelectionDiff with added/removed counts.
filter_str: Original filter string.
resolved_ids: Server's resolved IDs (current).
roster_hash: Server's roster hash (current).
outcome: Outcome to record.
reason: Reason for outcome (or None).
yield_product_code: Product code for yield item (or None).
yield_unit: Unit for yield item (or None).
yield_quantity: Quantity for yield item (or None).
yield_weight_kg: Weight in kg for yield item (or None).
ts_utc: Timestamp for resolution.
action: Route function or URL for confirmation submit.
Returns:
Div containing the diff panel with confirm button.
"""
# Build description of changes
changes = []
if diff.removed:
changes.append(f"{len(diff.removed)} animals were removed since you loaded this page")
if diff.added:
changes.append(f"{len(diff.added)} animals were added")
changes_text = ". ".join(changes) + "." if changes else "The selection has changed."
# Build confirmation form with hidden fields
resolved_id_fields = [
Hidden(name="resolved_ids", value=animal_id) for animal_id in resolved_ids
]
confirm_form = Form(
*resolved_id_fields,
Hidden(name="filter", value=filter_str),
Hidden(name="roster_hash", value=roster_hash),
Hidden(name="outcome", value=outcome),
Hidden(name="reason", value=reason or ""),
Hidden(name="yield_product_code", value=yield_product_code or ""),
Hidden(name="yield_unit", value=yield_unit or ""),
Hidden(name="yield_quantity", value=str(yield_quantity) if yield_quantity else ""),
Hidden(name="yield_weight_kg", value=str(yield_weight_kg) if yield_weight_kg else ""),
Hidden(name="ts_utc", value=str(ts_utc)),
Hidden(name="confirmed", value="true"),
Hidden(name="nonce", value=str(ULID())),
Div(
Button(
"Cancel",
type="button",
cls=ButtonT.default,
onclick="window.location.href='/actions/outcome'",
),
Button(
f"Confirm Outcome ({diff.server_count} animals)",
type="submit",
cls=ButtonT.destructive,
),
cls="flex gap-3 mt-4",
),
action=action,
method="post",
)
return Div(
Alert(
Div(
P("Selection Changed", cls="font-bold text-lg mb-2"),
P(changes_text, cls="mb-2"),
P(
f"Would you like to proceed with recording {outcome} for {diff.server_count} animals?",
cls="text-sm",
),
),
cls=AlertT.warning,
),
confirm_form,
cls="space-y-4",
)
# =============================================================================
# Correct Status Form (Admin-Only)
# =============================================================================
def status_correct_form(
filter_str: str = "",
resolved_ids: list[str] | None = None,
roster_hash: str = "",
ts_utc: int | None = None,
resolved_count: int = 0,
error: str | None = None,
action: Callable[..., Any] | str = "/actions/animal-status-correct",
) -> Form:
"""Create the Correct Status form (admin-only).
Args:
filter_str: Current filter string (DSL).
resolved_ids: Resolved animal IDs from filter.
roster_hash: Hash of resolved selection.
ts_utc: Timestamp of resolution.
resolved_count: Number of resolved animals.
error: Optional error message to display.
action: Route function or URL string for form submission.
Returns:
Form component for correcting animal status.
"""
if resolved_ids is None:
resolved_ids = []
# Error display component
error_component = None
if error:
error_component = Alert(error, cls=AlertT.warning)
# Admin warning
admin_warning = Alert(
"This is an administrative action. Changes will be logged with your reason.",
cls=AlertT.warning,
)
# Selection preview component
selection_preview = None
if resolved_count > 0:
selection_preview = Div(
P(
Span(f"{resolved_count}", cls="font-bold text-lg"),
" animals selected",
cls="text-sm",
),
cls="p-3 bg-slate-100 dark:bg-slate-800 rounded-md mb-4",
)
elif filter_str:
selection_preview = Div(
P("No animals match this filter", cls="text-sm text-amber-600"),
cls="p-3 bg-amber-50 dark:bg-amber-900/20 rounded-md mb-4",
)
# Build status options
status_options = [
Option("Select new status...", value="", selected=True, disabled=True),
Option("Alive", value="alive"),
Option("Dead", value="dead"),
Option("Harvested", value="harvested"),
Option("Sold", value="sold"),
]
# Hidden fields for resolved_ids (as multiple values)
resolved_id_fields = [
Hidden(name="resolved_ids", value=animal_id) for animal_id in resolved_ids
]
return Form(
H2("Correct Animal Status", cls="text-xl font-bold mb-4"),
admin_warning,
error_component,
selection_preview,
# Filter field
LabelInput(
label="Filter (DSL)",
id="filter",
name="filter",
value=filter_str,
placeholder="e.g., species:duck location:Coop1",
),
# New status selection
LabelSelect(
*status_options,
label="New Status",
id="new_status",
name="new_status",
required=True,
),
# Reason field (required for admin actions)
LabelInput(
label="Reason (required)",
id="reason",
name="reason",
placeholder="e.g., Data entry error, mis-identified animal",
required=True,
),
# Notes field
LabelTextArea(
label="Notes (optional)",
id="notes",
name="notes",
rows=2,
placeholder="Any additional notes...",
),
# Hidden fields for selection context
*resolved_id_fields,
Hidden(name="roster_hash", value=roster_hash),
Hidden(name="ts_utc", value=str(ts_utc or 0)),
Hidden(name="confirmed", value=""),
Hidden(name="nonce", value=str(ULID())),
# Submit button
Button("Correct Status", type="submit", cls=ButtonT.destructive),
# Form submission via standard action/method (hx-boost handles AJAX)
action=action,
method="post",
cls="space-y-4",
)
def status_correct_diff_panel(
diff: SelectionDiff,
filter_str: str,
resolved_ids: list[str],
roster_hash: str,
new_status: str,
reason: str,
ts_utc: int,
action: Callable[..., Any] | str = "/actions/animal-status-correct",
) -> Div:
"""Create the mismatch confirmation panel for status correction.
Args:
diff: SelectionDiff with added/removed counts.
filter_str: Original filter string.
resolved_ids: Server's resolved IDs (current).
roster_hash: Server's roster hash (current).
new_status: New status to set.
reason: Reason for correction.
ts_utc: Timestamp for resolution.
action: Route function or URL for confirmation submit.
Returns:
Div containing the diff panel with confirm button.
"""
# Build description of changes
changes = []
if diff.removed:
changes.append(f"{len(diff.removed)} animals were removed since you loaded this page")
if diff.added:
changes.append(f"{len(diff.added)} animals were added")
changes_text = ". ".join(changes) + "." if changes else "The selection has changed."
# Build confirmation form with hidden fields
resolved_id_fields = [
Hidden(name="resolved_ids", value=animal_id) for animal_id in resolved_ids
]
confirm_form = Form(
*resolved_id_fields,
Hidden(name="filter", value=filter_str),
Hidden(name="roster_hash", value=roster_hash),
Hidden(name="new_status", value=new_status),
Hidden(name="reason", value=reason),
Hidden(name="ts_utc", value=str(ts_utc)),
Hidden(name="confirmed", value="true"),
Hidden(name="nonce", value=str(ULID())),
Div(
Button(
"Cancel",
type="button",
cls=ButtonT.default,
onclick="window.location.href='/actions/status-correct'",
),
Button(
f"Confirm Correction ({diff.server_count} animals)",
type="submit",
cls=ButtonT.destructive,
),
cls="flex gap-3 mt-4",
),
action=action,
method="post",
)
return Div(
Alert(
Div(
P("Selection Changed", cls="font-bold text-lg mb-2"),
P(changes_text, cls="mb-2"),
P(
f"Would you like to proceed with correcting status to {new_status} for {diff.server_count} animals?",
cls="text-sm",
),
),
cls=AlertT.warning,
),
confirm_form,
cls="space-y-4",
)

15
src/animaltrack/web/templates/animal_detail.py
View File

@@ -155,14 +155,23 @@ def quick_actions_card(animal: AnimalDetail) -> Card:
)
)
actions.append(
Button("Add Tag", cls=ButtonT.default + " w-full", disabled=True),
A(
Button("Add Tag", cls=ButtonT.default + " w-full"),
href=f"/actions/tag-add?filter=animal_id:{animal.animal_id}",
)
)
if not animal.identified:
actions.append(
Button("Promote", cls=ButtonT.default + " w-full", disabled=True),
A(
Button("Promote", cls=ButtonT.default + " w-full"),
href=f"/actions/promote/{animal.animal_id}",
)
)
actions.append(
Button("Record Outcome", cls=ButtonT.destructive + " w-full", disabled=True),
A(
Button("Record Outcome", cls=ButtonT.destructive + " w-full"),
href=f"/actions/outcome?filter=animal_id:{animal.animal_id}",
)
)
return Card(

584
tests/test_web_actions.py
View File

@@ -968,3 +968,587 @@ class TestTagEndValidation:
)
assert resp.status_code == 422
# =============================================================================
# Update Attributes Tests
# =============================================================================
class TestAttrsFormRendering:
"""Tests for GET /actions/attrs form rendering."""
def test_attrs_form_renders(self, client):
"""GET /actions/attrs returns 200 with form elements."""
resp = client.get("/actions/attrs")
assert resp.status_code == 200
assert "Update Attributes" in resp.text
def test_attrs_form_has_filter_field(self, client):
"""Form has filter input field."""
resp = client.get("/actions/attrs")
assert resp.status_code == 200
assert 'name="filter"' in resp.text
def test_attrs_form_has_sex_dropdown(self, client):
"""Form has sex dropdown."""
resp = client.get("/actions/attrs")
assert resp.status_code == 200
assert 'name="sex"' in resp.text
def test_attrs_form_has_life_stage_dropdown(self, client):
"""Form has life_stage dropdown."""
resp = client.get("/actions/attrs")
assert resp.status_code == 200
assert 'name="life_stage"' in resp.text
class TestAttrsSuccess:
"""Tests for successful POST /actions/animal-attrs."""
def test_attrs_creates_event(self, client, seeded_db, animals_for_tagging):
"""POST creates AnimalAttributesUpdated event when valid."""
import time
from animaltrack.selection import compute_roster_hash
roster_hash = compute_roster_hash(animals_for_tagging, None)
ts_utc = int(time.time() * 1000)
resp = client.post(
"/actions/animal-attrs",
data={
"filter": "species:duck",
"sex": "male",
"resolved_ids": animals_for_tagging,
"roster_hash": roster_hash,
"ts_utc": str(ts_utc),
"nonce": "test-attrs-nonce-1",
},
)
assert resp.status_code == 200
# Verify event was created
event_row = seeded_db.execute(
"SELECT type FROM events WHERE type = 'AnimalAttributesUpdated' ORDER BY id DESC LIMIT 1"
).fetchone()
assert event_row is not None
assert event_row[0] == "AnimalAttributesUpdated"
def test_attrs_updates_registry(self, client, seeded_db, animals_for_tagging):
"""POST updates animal_registry with new attributes."""
import time
from animaltrack.selection import compute_roster_hash
roster_hash = compute_roster_hash(animals_for_tagging, None)
ts_utc = int(time.time() * 1000)
resp = client.post(
"/actions/animal-attrs",
data={
"filter": "species:duck",
"life_stage": "adult",
"resolved_ids": animals_for_tagging,
"roster_hash": roster_hash,
"ts_utc": str(ts_utc),
"nonce": "test-attrs-nonce-2",
},
)
assert resp.status_code == 200
# Verify attribute was updated
adult_count = seeded_db.execute(
"SELECT COUNT(*) FROM animal_registry WHERE animal_id IN ({}) AND life_stage = 'adult'".format(
",".join("?" * len(animals_for_tagging))
),
animals_for_tagging,
).fetchone()[0]
assert adult_count == len(animals_for_tagging)
def test_attrs_success_returns_toast(self, client, seeded_db, animals_for_tagging):
"""Successful attrs update returns HX-Trigger with toast."""
import time
from animaltrack.selection import compute_roster_hash
roster_hash = compute_roster_hash(animals_for_tagging, None)
ts_utc = int(time.time() * 1000)
resp = client.post(
"/actions/animal-attrs",
data={
"filter": "species:duck",
"repro_status": "intact",
"resolved_ids": animals_for_tagging,
"roster_hash": roster_hash,
"ts_utc": str(ts_utc),
"nonce": "test-attrs-nonce-3",
},
)
assert resp.status_code == 200
assert "HX-Trigger" in resp.headers
assert "showToast" in resp.headers["HX-Trigger"]
class TestAttrsValidation:
"""Tests for validation errors in POST /actions/animal-attrs."""
def test_attrs_no_attribute_returns_422(self, client, animals_for_tagging):
"""No attribute selected returns 422."""
import time
from animaltrack.selection import compute_roster_hash
roster_hash = compute_roster_hash(animals_for_tagging, None)
ts_utc = int(time.time() * 1000)
resp = client.post(
"/actions/animal-attrs",
data={
"filter": "species:duck",
# No sex, life_stage, or repro_status
"resolved_ids": animals_for_tagging,
"roster_hash": roster_hash,
"ts_utc": str(ts_utc),
"nonce": "test-attrs-nonce-4",
},
)
assert resp.status_code == 422
def test_attrs_no_animals_returns_422(self, client):
"""No animals selected returns 422."""
import time
ts_utc = int(time.time() * 1000)
resp = client.post(
"/actions/animal-attrs",
data={
"filter": "",
"sex": "male",
# No resolved_ids
"roster_hash": "",
"ts_utc": str(ts_utc),
"nonce": "test-attrs-nonce-5",
},
)
assert resp.status_code == 422
# =============================================================================
# Record Outcome Form Tests
# =============================================================================
class TestOutcomeFormRendering:
"""Tests for GET /actions/outcome form rendering."""
def test_outcome_form_renders(self, client):
"""Outcome form page renders with 200."""
resp = client.get("/actions/outcome")
assert resp.status_code == 200
assert "Record Outcome" in resp.text
def test_outcome_form_has_filter_field(self, client):
"""Outcome form has filter input field."""
resp = client.get("/actions/outcome")
assert 'name="filter"' in resp.text
def test_outcome_form_has_outcome_dropdown(self, client):
"""Outcome form has outcome dropdown."""
resp = client.get("/actions/outcome")
assert 'name="outcome"' in resp.text
assert "death" in resp.text.lower()
assert "harvest" in resp.text.lower()
def test_outcome_form_has_yield_section(self, client):
"""Outcome form has yield items section."""
resp = client.get("/actions/outcome")
assert "Yield Items" in resp.text
class TestOutcomeSuccess:
"""Tests for successful POST /actions/animal-outcome."""
def test_outcome_creates_event(self, client, seeded_db, animals_for_tagging):
"""Recording outcome creates AnimalOutcome event."""
import time
from animaltrack.selection import compute_roster_hash
roster_hash = compute_roster_hash(animals_for_tagging, None)
ts_utc = int(time.time() * 1000)
resp = client.post(
"/actions/animal-outcome",
data={
"filter": "species:duck",
"outcome": "death",
"reason": "old age",
"resolved_ids": animals_for_tagging,
"roster_hash": roster_hash,
"ts_utc": str(ts_utc),
"nonce": "test-outcome-nonce-1",
},
)
assert resp.status_code == 200
# Verify event was created
event_row = seeded_db.execute(
"SELECT type FROM events WHERE type = 'AnimalOutcome'"
).fetchone()
assert event_row is not None
def test_outcome_updates_status(self, client, seeded_db, animals_for_tagging):
"""Recording outcome updates animal status in registry."""
import time
from animaltrack.selection import compute_roster_hash
roster_hash = compute_roster_hash(animals_for_tagging, None)
ts_utc = int(time.time() * 1000)
resp = client.post(
"/actions/animal-outcome",
data={
"filter": "species:duck",
"outcome": "harvest",
"resolved_ids": animals_for_tagging,
"roster_hash": roster_hash,
"ts_utc": str(ts_utc),
"nonce": "test-outcome-nonce-2",
},
)
assert resp.status_code == 200
# Verify status was updated
harvested_count = seeded_db.execute(
"SELECT COUNT(*) FROM animal_registry WHERE animal_id IN ({}) AND status = 'harvested'".format(
",".join("?" * len(animals_for_tagging))
),
animals_for_tagging,
).fetchone()[0]
assert harvested_count == len(animals_for_tagging)
def test_outcome_success_returns_toast(self, client, seeded_db, animals_for_tagging):
"""Successful outcome recording returns HX-Trigger with toast."""
import time
from animaltrack.selection import compute_roster_hash
roster_hash = compute_roster_hash(animals_for_tagging, None)
ts_utc = int(time.time() * 1000)
resp = client.post(
"/actions/animal-outcome",
data={
"filter": "species:duck",
"outcome": "sold",
"resolved_ids": animals_for_tagging,
"roster_hash": roster_hash,
"ts_utc": str(ts_utc),
"nonce": "test-outcome-nonce-3",
},
)
assert resp.status_code == 200
assert "HX-Trigger" in resp.headers
assert "showToast" in resp.headers["HX-Trigger"]
class TestOutcomeValidation:
"""Tests for validation errors in POST /actions/animal-outcome."""
def test_outcome_no_outcome_returns_422(self, client, animals_for_tagging):
"""No outcome selected returns 422."""
import time
from animaltrack.selection import compute_roster_hash
roster_hash = compute_roster_hash(animals_for_tagging, None)
ts_utc = int(time.time() * 1000)
resp = client.post(
"/actions/animal-outcome",
data={
"filter": "species:duck",
# No outcome
"resolved_ids": animals_for_tagging,
"roster_hash": roster_hash,
"ts_utc": str(ts_utc),
"nonce": "test-outcome-nonce-4",
},
)
assert resp.status_code == 422
def test_outcome_no_animals_returns_422(self, client):
"""No animals selected returns 422."""
import time
ts_utc = int(time.time() * 1000)
resp = client.post(
"/actions/animal-outcome",
data={
"filter": "",
"outcome": "death",
# No resolved_ids
"roster_hash": "",
"ts_utc": str(ts_utc),
"nonce": "test-outcome-nonce-5",
},
)
assert resp.status_code == 422
# =============================================================================
# Status Correct Form Tests (Admin-Only)
# =============================================================================
@pytest.fixture
def admin_client(seeded_db):
"""Test client with admin role (dev mode - bypasses CSRF, auto-admin auth)."""
from animaltrack.web.app import create_app
# Use dev_mode=True so CSRF is bypassed and auth is auto-admin
settings = make_test_settings(trusted_proxy_ips="testclient", dev_mode=True)
app, rt = create_app(settings=settings, db=seeded_db)
return TestClient(app, raise_server_exceptions=True)
@pytest.fixture
def user_client(seeded_db):
"""Test client with regular (recorder) role."""
import time
from animaltrack.models.reference import User, UserRole
from animaltrack.repositories.users import UserRepository
from animaltrack.web.app import create_app
# Create recorder user in database
user_repo = UserRepository(seeded_db)
now = int(time.time() * 1000)
recorder_user = User(
username="recorder",
role=UserRole.RECORDER,
active=True,
created_at_utc=now,
updated_at_utc=now,
)
user_repo.upsert(recorder_user)
# Use dev_mode=False so auth_before checks the database
settings = make_test_settings(trusted_proxy_ips="testclient", dev_mode=False)
app, rt = create_app(settings=settings, db=seeded_db)
# Create client that sends the auth header
# raise_server_exceptions=False to get 403 instead of exception
client = TestClient(app, raise_server_exceptions=False)
client.headers["X-Oidc-Username"] = "recorder"
return client
class TestStatusCorrectFormRendering:
"""Tests for GET /actions/status-correct form rendering."""
def test_status_correct_form_renders_for_admin(self, admin_client):
"""Status correct form page renders for admin."""
resp = admin_client.get("/actions/status-correct")
assert resp.status_code == 200
assert "Correct" in resp.text
def test_status_correct_form_has_filter_field(self, admin_client):
"""Status correct form has filter input field."""
resp = admin_client.get("/actions/status-correct")
assert 'name="filter"' in resp.text
def test_status_correct_form_has_status_dropdown(self, admin_client):
"""Status correct form has status dropdown."""
resp = admin_client.get("/actions/status-correct")
assert 'name="new_status"' in resp.text
assert "alive" in resp.text.lower()
def test_status_correct_form_requires_reason(self, admin_client):
"""Status correct form has required reason field."""
resp = admin_client.get("/actions/status-correct")
assert 'name="reason"' in resp.text
def test_status_correct_returns_403_for_user(self, user_client):
"""Status correct returns 403 for non-admin user."""
resp = user_client.get("/actions/status-correct")
assert resp.status_code == 403
class TestStatusCorrectSuccess:
"""Tests for successful POST /actions/animal-status-correct."""
def test_status_correct_creates_event(self, admin_client, seeded_db, animals_for_tagging):
"""Correcting status creates AnimalStatusCorrected event."""
import time
from animaltrack.selection import compute_roster_hash
roster_hash = compute_roster_hash(animals_for_tagging, None)
ts_utc = int(time.time() * 1000)
resp = admin_client.post(
"/actions/animal-status-correct",
data={
"filter": "species:duck",
"new_status": "dead",
"reason": "Data entry error",
"resolved_ids": animals_for_tagging,
"roster_hash": roster_hash,
"ts_utc": str(ts_utc),
"nonce": "test-status-correct-nonce-1",
},
)
assert resp.status_code == 200
# Verify event was created
event_row = seeded_db.execute(
"SELECT type FROM events WHERE type = 'AnimalStatusCorrected'"
).fetchone()
assert event_row is not None
def test_status_correct_updates_status(self, admin_client, seeded_db, animals_for_tagging):
"""Correcting status updates animal status in registry."""
import time
from animaltrack.selection import compute_roster_hash
roster_hash = compute_roster_hash(animals_for_tagging, None)
ts_utc = int(time.time() * 1000)
resp = admin_client.post(
"/actions/animal-status-correct",
data={
"filter": "species:duck",
"new_status": "dead",
"reason": "Mis-identified animal",
"resolved_ids": animals_for_tagging,
"roster_hash": roster_hash,
"ts_utc": str(ts_utc),
"nonce": "test-status-correct-nonce-2",
},
)
assert resp.status_code == 200
# Verify status was updated
dead_count = seeded_db.execute(
"SELECT COUNT(*) FROM animal_registry WHERE animal_id IN ({}) AND status = 'dead'".format(
",".join("?" * len(animals_for_tagging))
),
animals_for_tagging,
).fetchone()[0]
assert dead_count == len(animals_for_tagging)
class TestStatusCorrectValidation:
"""Tests for validation errors in POST /actions/animal-status-correct."""
def test_status_correct_no_status_returns_422(self, admin_client, animals_for_tagging):
"""No status selected returns 422."""
import time
from animaltrack.selection import compute_roster_hash
roster_hash = compute_roster_hash(animals_for_tagging, None)
ts_utc = int(time.time() * 1000)
resp = admin_client.post(
"/actions/animal-status-correct",
data={
"filter": "species:duck",
# No new_status
"reason": "Some reason",
"resolved_ids": animals_for_tagging,
"roster_hash": roster_hash,
"ts_utc": str(ts_utc),
"nonce": "test-status-correct-nonce-3",
},
)
assert resp.status_code == 422
def test_status_correct_no_reason_returns_422(self, admin_client, animals_for_tagging):
"""No reason provided returns 422."""
import time
from animaltrack.selection import compute_roster_hash
roster_hash = compute_roster_hash(animals_for_tagging, None)
ts_utc = int(time.time() * 1000)
resp = admin_client.post(
"/actions/animal-status-correct",
data={
"filter": "species:duck",
"new_status": "dead",
# No reason
"resolved_ids": animals_for_tagging,
"roster_hash": roster_hash,
"ts_utc": str(ts_utc),
"nonce": "test-status-correct-nonce-4",
},
)
assert resp.status_code == 422
def test_status_correct_no_animals_returns_422(self, admin_client):
"""No animals selected returns 422."""
import time
ts_utc = int(time.time() * 1000)
resp = admin_client.post(
"/actions/animal-status-correct",
data={
"filter": "",
"new_status": "dead",
"reason": "Data entry error",
# No resolved_ids
"roster_hash": "",
"ts_utc": str(ts_utc),
"nonce": "test-status-correct-nonce-5",
},
)
assert resp.status_code == 422
def test_status_correct_returns_403_for_user(self, user_client, animals_for_tagging):
"""Status correct returns 403 for non-admin user."""
import time
from animaltrack.selection import compute_roster_hash
roster_hash = compute_roster_hash(animals_for_tagging, None)
ts_utc = int(time.time() * 1000)
resp = user_client.post(
"/actions/animal-status-correct",
data={
"filter": "species:duck",
"new_status": "dead",
"reason": "Data entry error",
"resolved_ids": animals_for_tagging,
"roster_hash": roster_hash,
"ts_utc": str(ts_utc),
"nonce": "test-status-correct-nonce-6",
},
)
assert resp.status_code == 403