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

feat: implement animal-tag-add and animal-tag-end routes (Step 9.1)

Browse Source 
Add selection-based tag actions with optimistic locking:
- GET /actions/tag-add and POST /actions/animal-tag-add
- GET /actions/tag-end and POST /actions/animal-tag-end
- Form templates with selection preview and tag input/dropdown
- Diff panel for handling selection mismatches (409 response)
- Add TagProjection to the action service registry
- 16 tests covering form rendering, success, validation

🤖 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 12:50:38 +00:00
parent 99f2fbb964
commit 3acb731a6c
3 changed files with 1199 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -14,6 +14,8 @@ from starlette.responses import HTMLResponse
from animaltrack.events.payloads import (
AnimalCohortCreatedPayload,
AnimalPromotedPayload,
AnimalTagEndedPayload,
AnimalTaggedPayload,
HatchRecordedPayload,
)
from animaltrack.events.store import EventStore
@@ -21,12 +23,23 @@ from animaltrack.projections import EventLogProjection, 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.animals import AnimalRepository
from animaltrack.repositories.locations import LocationRepository
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.templates import page
from animaltrack.web.templates.actions import cohort_form, hatch_form, promote_form
from animaltrack.web.templates.actions import (
cohort_form,
hatch_form,
promote_form,
tag_add_diff_panel,
tag_add_form,
tag_end_diff_panel,
tag_end_form,
)
def _create_animal_service(db: Any) -> AnimalService:
@@ -44,6 +57,7 @@ def _create_animal_service(db: Any) -> AnimalService:
registry.register(EventAnimalsProjection(db))
registry.register(IntervalProjection(db))
registry.register(EventLogProjection(db))
registry.register(TagProjection(db))
return AnimalService(db, event_store, registry)
@@ -441,6 +455,436 @@ def _render_promote_error(
)
# =============================================================================
# Add Tag
# =============================================================================
def tag_add_index(request: Request):
"""GET /actions/tag-add - Add Tag 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(
tag_add_form(
filter_str=filter_str,
resolved_ids=resolved_ids,
roster_hash=roster_hash,
ts_utc=ts_utc,
resolved_count=len(resolved_ids),
),
title="Add Tag - AnimalTrack",
active_nav=None,
)
async def animal_tag_add(request: Request):
"""POST /actions/animal-tag-add - Add tag to animals."""
db = request.app.state.db
form = await request.form()
# Extract form data
filter_str = form.get("filter", "")
tag = form.get("tag", "").strip()
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: tag required
if not tag:
return _render_tag_add_error_form(db, filter_str, "Please enter a tag")
# Validation: must have animals
if not resolved_ids:
return _render_tag_add_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(
tag_add_diff_panel(
diff=result.diff,
filter_str=filter_str,
resolved_ids=result.resolved_ids,
roster_hash=result.roster_hash,
tag=tag,
ts_utc=ts_utc,
),
title="Add Tag - 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_tag = current_resolution.animal_ids
else:
ids_to_tag = resolved_ids
# Check we still have animals
if not ids_to_tag:
return _render_tag_add_error_form(db, filter_str, "No animals remaining to tag")
# Create payload
try:
payload = AnimalTaggedPayload(
resolved_ids=list(ids_to_tag),
tag=tag,
)
except Exception as e:
return _render_tag_add_error_form(db, filter_str, str(e))
# Get actor from auth
auth = request.scope.get("auth")
actor = auth.username if auth else "unknown"
# Add tag
service = _create_animal_service(db)
try:
event = service.add_tag(
payload, ts_utc, actor, nonce=nonce, route="/actions/animal-tag-add"
)
except ValidationError as e:
return _render_tag_add_error_form(db, filter_str, str(e))
# Success: re-render fresh form
response = HTMLResponse(
content=to_xml(
page(
tag_add_form(),
title="Add Tag - AnimalTrack",
active_nav=None,
)
),
)
# Add toast trigger header
actually_tagged = event.entity_refs.get("actually_tagged", [])
response.headers["HX-Trigger"] = json.dumps(
{
"showToast": {
"message": f"Tagged {len(actually_tagged)} animal(s) as '{tag}'",
"type": "success",
}
}
)
return response
def _render_tag_add_error_form(db, filter_str, error_message):
"""Render tag add 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(
tag_add_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="Add Tag - AnimalTrack",
active_nav=None,
)
),
status_code=422,
)
# =============================================================================
# End Tag
# =============================================================================
def _get_active_tags_for_animals(db: Any, animal_ids: list[str]) -> list[str]:
"""Get tags that are active on at least one of the given animals.
Args:
db: Database connection.
animal_ids: List of animal IDs to check.
Returns:
Sorted list of unique active tag names.
"""
if not animal_ids:
return []
placeholders = ",".join("?" * len(animal_ids))
rows = db.execute(
f"""
SELECT DISTINCT tag
FROM animal_tag_intervals
WHERE animal_id IN ({placeholders})
AND end_utc IS NULL
ORDER BY tag
""",
animal_ids,
).fetchall()
return [row[0] for row in rows]
def tag_end_index(request: Request):
"""GET /actions/tag-end - End Tag 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 = ""
active_tags: list[str] = []
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)
active_tags = _get_active_tags_for_animals(db, resolved_ids)
return page(
tag_end_form(
filter_str=filter_str,
resolved_ids=resolved_ids,
roster_hash=roster_hash,
ts_utc=ts_utc,
resolved_count=len(resolved_ids),
active_tags=active_tags,
),
title="End Tag - AnimalTrack",
active_nav=None,
)
async def animal_tag_end(request: Request):
"""POST /actions/animal-tag-end - End tag on animals."""
db = request.app.state.db
form = await request.form()
# Extract form data
filter_str = form.get("filter", "")
tag = form.get("tag", "").strip()
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: tag required
if not tag:
return _render_tag_end_error_form(db, filter_str, "Please select a tag to end")
# Validation: must have animals
if not resolved_ids:
return _render_tag_end_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(
tag_end_diff_panel(
diff=result.diff,
filter_str=filter_str,
resolved_ids=result.resolved_ids,
roster_hash=result.roster_hash,
tag=tag,
ts_utc=ts_utc,
),
title="End Tag - 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_untag = current_resolution.animal_ids
else:
ids_to_untag = resolved_ids
# Check we still have animals
if not ids_to_untag:
return _render_tag_end_error_form(db, filter_str, "No animals remaining")
# Create payload
try:
payload = AnimalTagEndedPayload(
resolved_ids=list(ids_to_untag),
tag=tag,
)
except Exception as e:
return _render_tag_end_error_form(db, filter_str, str(e))
# Get actor from auth
auth = request.scope.get("auth")
actor = auth.username if auth else "unknown"
# End tag
service = _create_animal_service(db)
try:
event = service.end_tag(
payload, ts_utc, actor, nonce=nonce, route="/actions/animal-tag-end"
)
except ValidationError as e:
return _render_tag_end_error_form(db, filter_str, str(e))
# Success: re-render fresh form
response = HTMLResponse(
content=to_xml(
page(
tag_end_form(),
title="End Tag - AnimalTrack",
active_nav=None,
)
),
)
# Add toast trigger header
actually_ended = event.entity_refs.get("actually_ended", [])
response.headers["HX-Trigger"] = json.dumps(
{
"showToast": {
"message": f"Ended tag '{tag}' on {len(actually_ended)} animal(s)",
"type": "success",
}
}
)
return response
def _render_tag_end_error_form(db, filter_str, error_message):
"""Render tag end form with error message."""
# Re-resolve to show current selection info
ts_utc = int(time.time() * 1000)
resolved_ids: list[str] = []
roster_hash = ""
active_tags: list[str] = []
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)
active_tags = _get_active_tags_for_animals(db, resolved_ids)
return HTMLResponse(
content=to_xml(
page(
tag_end_form(
filter_str=filter_str,
resolved_ids=resolved_ids,
roster_hash=roster_hash,
ts_utc=ts_utc,
resolved_count=len(resolved_ids),
active_tags=active_tags,
error=error_message,
),
title="End Tag - AnimalTrack",
active_nav=None,
)
),
status_code=422,
)
# =============================================================================
# Route Registration
# =============================================================================
@@ -462,3 +906,9 @@ def register_action_routes(rt, app):
# Single animal actions
rt("/actions/promote/{animal_id}")(promote_index)
rt("/actions/animal-promote", methods=["POST"])(animal_promote)
# Selection-based actions
rt("/actions/tag-add")(tag_add_index)
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)

365
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
from fasthtml.common import H2, Div, Form, Hidden, Option, P, Span
from monsterui.all import (
Alert,
AlertT,
@@ -18,6 +18,7 @@ from ulid import ULID
from animaltrack.models.animals import Animal
from animaltrack.models.reference import Location, Species
from animaltrack.selection.validation import SelectionDiff
# =============================================================================
# Cohort Creation Form
@@ -395,3 +396,365 @@ def promote_form(
method="post",
cls="space-y-4",
)
# =============================================================================
# Add Tag Form
# =============================================================================
def tag_add_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-tag-add",
) -> Form:
"""Create the Add Tag 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 adding tags to animals.
"""
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",
)
# 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("Add Tag", 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., location:"Strip 1" species:duck',
),
# Selection preview
selection_preview,
# Tag input
LabelInput(
"Tag",
id="tag",
name="tag",
placeholder="Enter tag name",
),
# 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("Add Tag", type="submit", cls=ButtonT.primary),
# Form submission via standard action/method (hx-boost handles AJAX)
action=action,
method="post",
cls="space-y-4",
)
def tag_add_diff_panel(
diff: SelectionDiff,
filter_str: str,
resolved_ids: list[str],
roster_hash: str,
tag: str,
ts_utc: int,
action: Callable[..., Any] | str = "/actions/animal-tag-add",
) -> Div:
"""Create the mismatch confirmation panel for tag add.
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).
tag: Tag to add.
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="tag", value=tag),
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/tag-add'",
),
Button(
f"Confirm Tag ({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 tagging {diff.server_count} animals as '{tag}'?",
cls="text-sm",
),
),
cls=AlertT.warning,
),
confirm_form,
cls="space-y-4",
)
# =============================================================================
# End Tag Form
# =============================================================================
def tag_end_form(
filter_str: str = "",
resolved_ids: list[str] | None = None,
roster_hash: str = "",
ts_utc: int | None = None,
resolved_count: int = 0,
active_tags: list[str] | None = None,
error: str | None = None,
action: Callable[..., Any] | str = "/actions/animal-tag-end",
) -> Form:
"""Create the End Tag 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.
active_tags: List of tags active on selected animals.
error: Optional error message to display.
action: Route function or URL string for form submission.
Returns:
Form component for ending tags on animals.
"""
if resolved_ids is None:
resolved_ids = []
if active_tags is None:
active_tags = []
# 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 tag options from active_tags
tag_options = [Option("Select tag to end...", value="", disabled=True, selected=True)]
for tag in active_tags:
tag_options.append(Option(tag, value=tag))
# 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("End Tag", 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., tag:layer-birds species:duck",
),
# Selection preview
selection_preview,
# Tag dropdown
LabelSelect(
*tag_options,
label="Tag to End",
id="tag",
name="tag",
)
if active_tags
else Div(
P("No active tags on selected animals", cls="text-sm text-stone-400"),
cls="p-3 bg-slate-800 rounded-md",
),
# 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("End Tag", type="submit", cls=ButtonT.primary, disabled=not active_tags),
# Form submission via standard action/method (hx-boost handles AJAX)
action=action,
method="post",
cls="space-y-4",
)
def tag_end_diff_panel(
diff: SelectionDiff,
filter_str: str,
resolved_ids: list[str],
roster_hash: str,
tag: str,
ts_utc: int,
action: Callable[..., Any] | str = "/actions/animal-tag-end",
) -> Div:
"""Create the mismatch confirmation panel for tag end.
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).
tag: Tag to end.
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="tag", value=tag),
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/tag-end'",
),
Button(
f"Confirm End Tag ({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 ending tag '{tag}' on {diff.server_count} animals?",
cls="text-sm",
),
),
cls=AlertT.warning,
),
confirm_form,
cls="space-y-4",
)

384
tests/test_web_actions.py
View File

@@ -584,3 +584,387 @@ class TestPromoteValidation:
)
assert resp.status_code == 404
# =============================================================================
# Add Tag Tests
# =============================================================================
@pytest.fixture
def animals_for_tagging(seeded_db, client, location_strip1_id):
"""Create animals for tag testing and return their IDs."""
# Create a cohort with 3 animals
resp = client.post(
"/actions/animal-cohort",
data={
"species": "duck",
"location_id": location_strip1_id,
"count": "3",
"life_stage": "adult",
"sex": "female",
"origin": "purchased",
"nonce": "test-tag-fixture-cohort-1",
},
)
assert resp.status_code == 200
# Get the animal IDs
rows = seeded_db.execute(
"SELECT animal_id FROM animal_registry WHERE origin = 'purchased' AND life_stage = 'adult' ORDER BY animal_id DESC LIMIT 3"
).fetchall()
return [row[0] for row in rows]
class TestTagAddFormRendering:
"""Tests for GET /actions/tag-add form rendering."""
def test_tag_add_form_renders(self, client):
"""GET /actions/tag-add returns 200 with form elements."""
resp = client.get("/actions/tag-add")
assert resp.status_code == 200
assert "Add Tag" in resp.text
def test_tag_add_form_has_filter_field(self, client):
"""Form has filter input field."""
resp = client.get("/actions/tag-add")
assert resp.status_code == 200
assert 'name="filter"' in resp.text
def test_tag_add_form_has_tag_field(self, client):
"""Form has tag input field."""
resp = client.get("/actions/tag-add")
assert resp.status_code == 200
assert 'name="tag"' in resp.text
def test_tag_add_form_with_filter_shows_selection(self, client, animals_for_tagging):
"""Form with filter shows selection preview."""
# Use species filter which is a valid filter field
resp = client.get("/actions/tag-add?filter=species:duck")
assert resp.status_code == 200
# Should show selection count
assert "selected" in resp.text.lower()
class TestTagAddSuccess:
"""Tests for successful POST /actions/animal-tag-add."""
def test_tag_add_creates_event(self, client, seeded_db, animals_for_tagging):
"""POST creates AnimalTagged 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-tag-add",
data={
"filter": "species:duck",
"tag": "test-tag",
"resolved_ids": animals_for_tagging,
"roster_hash": roster_hash,
"ts_utc": str(ts_utc),
"nonce": "test-tag-nonce-1",
},
)
assert resp.status_code == 200
# Verify event was created
event_row = seeded_db.execute(
"SELECT type FROM events WHERE type = 'AnimalTagged' ORDER BY id DESC LIMIT 1"
).fetchone()
assert event_row is not None
assert event_row[0] == "AnimalTagged"
def test_tag_add_creates_tag_intervals(self, client, seeded_db, animals_for_tagging):
"""POST creates tag intervals for animals."""
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-tag-add",
data={
"filter": "species:duck",
"tag": "layer-birds",
"resolved_ids": animals_for_tagging,
"roster_hash": roster_hash,
"ts_utc": str(ts_utc),
"nonce": "test-tag-nonce-2",
},
)
assert resp.status_code == 200
# Verify tag intervals were created
tag_count = seeded_db.execute(
"SELECT COUNT(*) FROM animal_tag_intervals WHERE tag = 'layer-birds' AND end_utc IS NULL"
).fetchone()[0]
assert tag_count >= len(animals_for_tagging)
def test_tag_add_success_returns_toast(self, client, seeded_db, animals_for_tagging):
"""Successful tag add 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-tag-add",
data={
"filter": "species:duck",
"tag": "test-tag-toast",
"resolved_ids": animals_for_tagging,
"roster_hash": roster_hash,
"ts_utc": str(ts_utc),
"nonce": "test-tag-nonce-3",
},
)
assert resp.status_code == 200
assert "HX-Trigger" in resp.headers
assert "showToast" in resp.headers["HX-Trigger"]
class TestTagAddValidation:
"""Tests for validation errors in POST /actions/animal-tag-add."""
def test_tag_add_missing_tag_returns_422(self, client, animals_for_tagging):
"""Missing tag 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-tag-add",
data={
"filter": "species:duck",
# Missing tag
"resolved_ids": animals_for_tagging,
"roster_hash": roster_hash,
"ts_utc": str(ts_utc),
"nonce": "test-tag-nonce-4",
},
)
assert resp.status_code == 422
def test_tag_add_no_animals_returns_422(self, client):
"""No animals selected returns 422."""
import time
ts_utc = int(time.time() * 1000)
resp = client.post(
"/actions/animal-tag-add",
data={
"filter": "",
"tag": "test-tag",
# No resolved_ids
"roster_hash": "",
"ts_utc": str(ts_utc),
"nonce": "test-tag-nonce-5",
},
)
assert resp.status_code == 422
# =============================================================================
# End Tag Tests
# =============================================================================
@pytest.fixture
def tagged_animals(seeded_db, client, animals_for_tagging):
"""Tag animals and return their IDs."""
import time
from animaltrack.selection import compute_roster_hash
roster_hash = compute_roster_hash(animals_for_tagging, None)
ts_utc = int(time.time() * 1000)
# First tag the animals
resp = client.post(
"/actions/animal-tag-add",
data={
"filter": "species:duck",
"tag": "test-end-tag",
"resolved_ids": animals_for_tagging,
"roster_hash": roster_hash,
"ts_utc": str(ts_utc),
"nonce": "test-fixture-tag-1",
},
)
assert resp.status_code == 200
return animals_for_tagging
class TestTagEndFormRendering:
"""Tests for GET /actions/tag-end form rendering."""
def test_tag_end_form_renders(self, client):
"""GET /actions/tag-end returns 200 with form elements."""
resp = client.get("/actions/tag-end")
assert resp.status_code == 200
assert "End Tag" in resp.text
def test_tag_end_form_has_filter_field(self, client):
"""Form has filter input field."""
resp = client.get("/actions/tag-end")
assert resp.status_code == 200
assert 'name="filter"' in resp.text
class TestTagEndSuccess:
"""Tests for successful POST /actions/animal-tag-end."""
def test_tag_end_creates_event(self, client, seeded_db, tagged_animals):
"""POST creates AnimalTagEnded event when valid."""
import time
from animaltrack.selection import compute_roster_hash
roster_hash = compute_roster_hash(tagged_animals, None)
ts_utc = int(time.time() * 1000)
resp = client.post(
"/actions/animal-tag-end",
data={
"filter": "species:duck",
"tag": "test-end-tag",
"resolved_ids": tagged_animals,
"roster_hash": roster_hash,
"ts_utc": str(ts_utc),
"nonce": "test-tag-end-nonce-1",
},
)
assert resp.status_code == 200
# Verify event was created
event_row = seeded_db.execute(
"SELECT type FROM events WHERE type = 'AnimalTagEnded' ORDER BY id DESC LIMIT 1"
).fetchone()
assert event_row is not None
assert event_row[0] == "AnimalTagEnded"
def test_tag_end_closes_intervals(self, client, seeded_db, tagged_animals):
"""POST closes tag intervals for animals."""
import time
from animaltrack.selection import compute_roster_hash
roster_hash = compute_roster_hash(tagged_animals, None)
ts_utc = int(time.time() * 1000)
# Verify intervals are open before
open_before = seeded_db.execute(
"SELECT COUNT(*) FROM animal_tag_intervals WHERE tag = 'test-end-tag' AND end_utc IS NULL"
).fetchone()[0]
assert open_before >= len(tagged_animals)
resp = client.post(
"/actions/animal-tag-end",
data={
"filter": "species:duck",
"tag": "test-end-tag",
"resolved_ids": tagged_animals,
"roster_hash": roster_hash,
"ts_utc": str(ts_utc),
"nonce": "test-tag-end-nonce-2",
},
)
assert resp.status_code == 200
# Verify intervals are closed after
open_after = seeded_db.execute(
"SELECT COUNT(*) FROM animal_tag_intervals WHERE tag = 'test-end-tag' AND end_utc IS NULL"
).fetchone()[0]
assert open_after == 0
def test_tag_end_success_returns_toast(self, client, seeded_db, tagged_animals):
"""Successful tag end returns HX-Trigger with toast."""
import time
from animaltrack.selection import compute_roster_hash
roster_hash = compute_roster_hash(tagged_animals, None)
ts_utc = int(time.time() * 1000)
resp = client.post(
"/actions/animal-tag-end",
data={
"filter": "species:duck",
"tag": "test-end-tag",
"resolved_ids": tagged_animals,
"roster_hash": roster_hash,
"ts_utc": str(ts_utc),
"nonce": "test-tag-end-nonce-3",
},
)
assert resp.status_code == 200
assert "HX-Trigger" in resp.headers
assert "showToast" in resp.headers["HX-Trigger"]
class TestTagEndValidation:
"""Tests for validation errors in POST /actions/animal-tag-end."""
def test_tag_end_missing_tag_returns_422(self, client, tagged_animals):
"""Missing tag returns 422."""
import time
from animaltrack.selection import compute_roster_hash
roster_hash = compute_roster_hash(tagged_animals, None)
ts_utc = int(time.time() * 1000)
resp = client.post(
"/actions/animal-tag-end",
data={
"filter": "species:duck",
# Missing tag
"resolved_ids": tagged_animals,
"roster_hash": roster_hash,
"ts_utc": str(ts_utc),
"nonce": "test-tag-end-nonce-4",
},
)
assert resp.status_code == 422
def test_tag_end_no_animals_returns_422(self, client):
"""No animals selected returns 422."""
import time
ts_utc = int(time.time() * 1000)
resp = client.post(
"/actions/animal-tag-end",
data={
"filter": "",
"tag": "some-tag",
# No resolved_ids
"roster_hash": "",
"ts_utc": str(ts_utc),
"nonce": "test-tag-end-nonce-5",
},
)
assert resp.status_code == 422