diff --git a/src/animaltrack/repositories/locations.py b/src/animaltrack/repositories/locations.py index 1b8a143..fed5bbc 100644 --- a/src/animaltrack/repositories/locations.py +++ b/src/animaltrack/repositories/locations.py @@ -102,3 +102,23 @@ class LocationRepository: ) for row in rows ] + + def list_active(self) -> list[Location]: + """Get all active locations. + + Returns: + List of active locations, ordered by name. + """ + rows = self.db.execute( + "SELECT id, name, active, created_at_utc, updated_at_utc FROM locations WHERE active = 1 ORDER BY name" + ).fetchall() + return [ + Location( + id=row[0], + name=row[1], + active=bool(row[2]), + created_at_utc=row[3], + updated_at_utc=row[4], + ) + for row in rows + ] diff --git a/src/animaltrack/web/app.py b/src/animaltrack/web/app.py index a3a708f..282c854 100644 --- a/src/animaltrack/web/app.py +++ b/src/animaltrack/web/app.py @@ -5,7 +5,7 @@ from __future__ import annotations from pathlib import Path -from fasthtml.common import H1, Beforeware, P, fast_app +from fasthtml.common import Beforeware, fast_app from monsterui.all import Theme from starlette.middleware import Middleware from starlette.requests import Request @@ -17,8 +17,7 @@ from animaltrack.web.middleware import ( csrf_before, request_id_before, ) -from animaltrack.web.routes import register_health_routes -from animaltrack.web.templates import page +from animaltrack.web.routes import register_egg_routes, register_health_routes # Default static directory relative to this module DEFAULT_STATIC_DIR = Path(__file__).parent.parent / "static" @@ -124,20 +123,8 @@ def create_app( app.state.settings = settings app.state.db = db - # Register health routes (healthz, metrics) + # Register routes register_health_routes(rt, app) - - # Placeholder index route (will be replaced with Egg Quick Capture later) - @rt("/") - def index(): - """Placeholder index route - shows egg capture page.""" - return page( - ( - H1("Egg Quick Capture", cls="text-2xl font-bold mb-4"), - P("Coming soon...", cls="text-stone-400"), - ), - title="Egg - AnimalTrack", - active_nav="egg", - ) + register_egg_routes(rt, app) return app, rt diff --git a/src/animaltrack/web/middleware.py b/src/animaltrack/web/middleware.py index 49660c4..055ffc9 100644 --- a/src/animaltrack/web/middleware.py +++ b/src/animaltrack/web/middleware.py @@ -221,6 +221,8 @@ def csrf_before(req: Request, settings: Settings) -> Response | None: 1. CSRF cookie present and matches header 2. Origin or Referer matches expected host + In dev_mode, bypasses CSRF validation entirely. + Args: req: The Starlette request object. settings: Application settings. @@ -228,6 +230,10 @@ def csrf_before(req: Request, settings: Settings) -> Response | None: Returns: None to continue processing, or Response to short-circuit. """ + # Dev mode: bypass CSRF entirely + if settings.dev_mode: + return None + # Skip CSRF check for safe methods if is_safe_method(req.method): return None diff --git a/src/animaltrack/web/routes/__init__.py b/src/animaltrack/web/routes/__init__.py index 6e27ec8..197e091 100644 --- a/src/animaltrack/web/routes/__init__.py +++ b/src/animaltrack/web/routes/__init__.py @@ -1,6 +1,7 @@ # ABOUTME: Routes package for AnimalTrack web application. # ABOUTME: Contains modular route handlers for different features. +from animaltrack.web.routes.eggs import register_egg_routes from animaltrack.web.routes.health import register_health_routes -__all__ = ["register_health_routes"] +__all__ = ["register_egg_routes", "register_health_routes"] diff --git a/src/animaltrack/web/routes/eggs.py b/src/animaltrack/web/routes/eggs.py new file mode 100644 index 0000000..3e85af7 --- /dev/null +++ b/src/animaltrack/web/routes/eggs.py @@ -0,0 +1,190 @@ +# ABOUTME: Routes for Egg Quick Capture functionality. +# ABOUTME: Handles GET / form and POST /actions/product-collected. + +from __future__ import annotations + +import json +import time +from typing import Any + +from starlette.requests import Request +from starlette.responses import HTMLResponse + +from animaltrack.events.payloads import ProductCollectedPayload +from animaltrack.events.store import EventStore +from animaltrack.projections import ProjectionRegistry +from animaltrack.projections.animal_registry import AnimalRegistryProjection +from animaltrack.projections.event_animals import EventAnimalsProjection +from animaltrack.projections.intervals import IntervalProjection +from animaltrack.projections.products import ProductsProjection +from animaltrack.repositories.locations import LocationRepository +from animaltrack.services.products import ProductService, ValidationError +from animaltrack.web.templates import page +from animaltrack.web.templates.eggs import egg_form + + +def resolve_ducks_at_location(db: Any, location_id: str, ts_utc: int) -> list[str]: + """Resolve all duck animal IDs at a location at given timestamp. + + Args: + db: Database connection. + location_id: Location ID (ULID). + ts_utc: Timestamp in ms since Unix epoch. + + Returns: + List of animal IDs (ducks at the location, alive at ts_utc). + """ + query = """ + SELECT DISTINCT ali.animal_id + FROM animal_location_intervals ali + JOIN animal_registry ar ON ali.animal_id = ar.animal_id + WHERE ali.location_id = ? + AND ali.start_utc <= ? + AND (ali.end_utc IS NULL OR ali.end_utc > ?) + AND ar.species_code = 'duck' + AND ar.status = 'alive' + ORDER BY ali.animal_id + """ + rows = db.execute(query, (location_id, ts_utc, ts_utc)).fetchall() + return [row[0] for row in rows] + + +def register_egg_routes(rt, app): + """Register egg capture routes. + + Args: + rt: FastHTML route decorator. + app: FastHTML application instance. + """ + + @rt("/") + def index(request: Request): + """GET / - Egg Quick Capture form.""" + db = app.state.db + locations = LocationRepository(db).list_active() + + # Check for pre-selected location from query params + selected_location_id = request.query_params.get("location_id") + + return page( + egg_form(locations, selected_location_id=selected_location_id), + title="Egg - AnimalTrack", + active_nav="egg", + ) + + @rt("/actions/product-collected", methods=["POST"]) + async def product_collected(request: Request): + """POST /actions/product-collected - Record egg collection.""" + db = app.state.db + form = await request.form() + + # Extract form data + location_id = form.get("location_id", "") + quantity_str = form.get("quantity", "0") + notes = form.get("notes") or None + nonce = form.get("nonce") + + # Get locations for potential re-render + locations = LocationRepository(db).list_active() + + # Validate location_id + if not location_id: + return _render_error_form(locations, None, "Please select a location") + + # Validate quantity + try: + quantity = int(quantity_str) + except ValueError: + return _render_error_form(locations, location_id, "Quantity must be a number") + + if quantity < 1: + return _render_error_form(locations, location_id, "Quantity must be at least 1") + + # Get current timestamp + ts_utc = int(time.time() * 1000) + + # Resolve ducks at location + resolved_ids = resolve_ducks_at_location(db, location_id, ts_utc) + + if not resolved_ids: + return _render_error_form(locations, location_id, "No ducks at this location") + + # Create product service + event_store = EventStore(db) + registry = ProjectionRegistry() + registry.register(AnimalRegistryProjection(db)) + registry.register(EventAnimalsProjection(db)) + registry.register(IntervalProjection(db)) + registry.register(ProductsProjection(db)) + + product_service = ProductService(db, event_store, registry) + + # Create payload + payload = ProductCollectedPayload( + location_id=location_id, + product_code="egg.duck", + quantity=quantity, + resolved_ids=resolved_ids, + notes=notes, + ) + + # Get actor from auth + auth = request.scope.get("auth") + actor = auth.username if auth else "unknown" + + # Collect product + try: + product_service.collect_product( + payload=payload, + ts_utc=ts_utc, + actor=actor, + nonce=nonce, + route="/actions/product-collected", + ) + except ValidationError as e: + return _render_error_form(locations, location_id, str(e)) + + # Success: re-render form with location sticking, qty cleared + response = HTMLResponse( + content=str( + page( + egg_form(locations, selected_location_id=location_id), + title="Egg - AnimalTrack", + active_nav="egg", + ) + ), + ) + + # Add toast trigger header + response.headers["HX-Trigger"] = json.dumps( + {"showToast": {"message": f"Recorded {quantity} eggs", "type": "success"}} + ) + + return response + + +def _render_error_form(locations, selected_location_id, error_message): + """Render form with error message. + + Args: + locations: List of active locations. + selected_location_id: Currently selected location. + error_message: Error message to display. + + Returns: + HTMLResponse with 422 status. + """ + return HTMLResponse( + content=str( + page( + egg_form( + locations, + selected_location_id=selected_location_id, + error=error_message, + ), + title="Egg - AnimalTrack", + active_nav="egg", + ) + ), + status_code=422, + ) diff --git a/src/animaltrack/web/templates/eggs.py b/src/animaltrack/web/templates/eggs.py new file mode 100644 index 0000000..52f6495 --- /dev/null +++ b/src/animaltrack/web/templates/eggs.py @@ -0,0 +1,89 @@ +# ABOUTME: Templates for Egg Quick Capture form. +# ABOUTME: Provides form components for recording egg collections. + +from fasthtml.common import H2, Form, Hidden, Option +from monsterui.all import Button, ButtonT, LabelInput, LabelSelect, LabelTextArea +from ulid import ULID + +from animaltrack.models.reference import Location + + +def egg_form( + locations: list[Location], + selected_location_id: str | None = None, + error: str | None = None, +) -> Form: + """Create the Egg Quick Capture form. + + Args: + locations: List of active locations for the dropdown. + selected_location_id: Pre-selected location ID (sticks after submission). + error: Optional error message to display. + + Returns: + Form component for egg collection. + """ + # Build location options + location_options = [ + Option( + loc.name, + value=loc.id, + selected=(loc.id == selected_location_id), + ) + for loc in locations + ] + + # Add placeholder option if no location is selected + if selected_location_id is None: + location_options.insert( + 0, Option("Select a location...", value="", disabled=True, selected=True) + ) + + # Error display component + error_component = None + if error: + from fasthtml.common import Div, P + + error_component = Div( + P(error, cls="text-red-500 text-sm"), + cls="mb-4", + ) + + return Form( + H2("Record Eggs", cls="text-xl font-bold mb-4"), + # Error message if present + error_component, + # Location dropdown + LabelSelect( + *location_options, + label="Location", + id="location_id", + name="location_id", + ), + # Quantity input (integer only, min=1) + LabelInput( + "Quantity", + id="quantity", + name="quantity", + type="number", + min="1", + step="1", + placeholder="Number of eggs", + required=True, + ), + # Optional notes + LabelTextArea( + "Notes", + id="notes", + placeholder="Optional notes", + ), + # Hidden nonce for idempotency + Hidden(name="nonce", value=str(ULID())), + # Submit button + Button("Record Eggs", type="submit", cls=ButtonT.primary), + # Form submission via HTMX + hx_post="/actions/product-collected", + hx_target="body", + hx_swap="innerHTML", + cls="space-y-4", + ) diff --git a/tests/test_web_eggs.py b/tests/test_web_eggs.py new file mode 100644 index 0000000..c7249b0 --- /dev/null +++ b/tests/test_web_eggs.py @@ -0,0 +1,213 @@ +# ABOUTME: Tests for Egg Quick Capture web routes. +# ABOUTME: Covers GET / form rendering and POST /actions/product-collected. + +import os +import time + +import pytest +from starlette.testclient import TestClient + +from animaltrack.events.payloads import AnimalCohortCreatedPayload +from animaltrack.events.store import EventStore +from animaltrack.projections import ProjectionRegistry +from animaltrack.projections.animal_registry import AnimalRegistryProjection +from animaltrack.projections.event_animals import EventAnimalsProjection +from animaltrack.projections.intervals import IntervalProjection +from animaltrack.projections.products import ProductsProjection +from animaltrack.services.animal import AnimalService + + +def make_test_settings( + csrf_secret: str = "test-secret", + trusted_proxy_ips: str = "127.0.0.1", + dev_mode: bool = True, +): + """Create Settings for testing by setting env vars temporarily.""" + from animaltrack.config import Settings + + old_env = os.environ.copy() + try: + os.environ["CSRF_SECRET"] = csrf_secret + os.environ["TRUSTED_PROXY_IPS"] = trusted_proxy_ips + os.environ["DEV_MODE"] = str(dev_mode).lower() + return Settings() + finally: + os.environ.clear() + os.environ.update(old_env) + + +@pytest.fixture +def client(seeded_db): + """Create a test client for the app.""" + from animaltrack.web.app import create_app + + settings = make_test_settings(trusted_proxy_ips="testclient") + app, rt = create_app(settings=settings, db=seeded_db) + # Use raise_server_exceptions=True to see actual errors + return TestClient(app, raise_server_exceptions=True) + + +@pytest.fixture +def location_strip1_id(seeded_db): + """Get Strip 1 location ID from seeded data.""" + row = seeded_db.execute("SELECT id FROM locations WHERE name = 'Strip 1'").fetchone() + return row[0] + + +@pytest.fixture +def location_nursery1_id(seeded_db): + """Get Nursery 1 location ID from seeded data (no ducks here).""" + row = seeded_db.execute("SELECT id FROM locations WHERE name = 'Nursery 1'").fetchone() + return row[0] + + +@pytest.fixture +def ducks_at_strip1(seeded_db, location_strip1_id): + """Create ducks at Strip 1 for testing egg collection.""" + event_store = EventStore(seeded_db) + registry = ProjectionRegistry() + registry.register(AnimalRegistryProjection(seeded_db)) + registry.register(EventAnimalsProjection(seeded_db)) + registry.register(IntervalProjection(seeded_db)) + registry.register(ProductsProjection(seeded_db)) + + animal_service = AnimalService(seeded_db, event_store, registry) + + payload = AnimalCohortCreatedPayload( + species="duck", + count=5, + life_stage="adult", + sex="female", + location_id=location_strip1_id, + origin="purchased", + ) + ts_utc = int(time.time() * 1000) + event = animal_service.create_cohort(payload, ts_utc, "test_user") + return event.entity_refs["animal_ids"] + + +class TestEggFormRendering: + """Tests for GET / egg capture form.""" + + def test_egg_form_renders(self, client): + """GET / returns 200 with form elements.""" + resp = client.get("/") + assert resp.status_code == 200 + assert "Record Eggs" in resp.text or "Egg" in resp.text + + def test_egg_form_shows_locations(self, client): + """Form has location dropdown with seeded locations.""" + resp = client.get("/") + assert resp.status_code == 200 + # Check for seeded location names in the response + assert "Strip 1" in resp.text + assert "Strip 2" in resp.text + + def test_egg_form_has_quantity_field(self, client): + """Form has quantity input field.""" + resp = client.get("/") + assert resp.status_code == 200 + assert 'name="quantity"' in resp.text or 'id="quantity"' in resp.text + + +class TestEggCollection: + """Tests for POST /actions/product-collected.""" + + def test_egg_collection_creates_event( + self, client, seeded_db, location_strip1_id, ducks_at_strip1 + ): + """POST creates ProductCollected event when ducks exist at location.""" + resp = client.post( + "/actions/product-collected", + data={ + "location_id": location_strip1_id, + "quantity": "12", + "notes": "Morning collection", + "nonce": "test-nonce-123", + }, + ) + + # Should succeed (200 or redirect) + assert resp.status_code in [200, 302, 303] + + # Verify event was created in database + event_row = seeded_db.execute( + "SELECT type, payload FROM events WHERE type = 'ProductCollected' ORDER BY id DESC LIMIT 1" + ).fetchone() + assert event_row is not None + assert event_row[0] == "ProductCollected" + + def test_egg_collection_validation_quantity_zero( + self, client, location_strip1_id, ducks_at_strip1 + ): + """quantity=0 returns 422.""" + resp = client.post( + "/actions/product-collected", + data={ + "location_id": location_strip1_id, + "quantity": "0", + "nonce": "test-nonce-456", + }, + ) + + assert resp.status_code == 422 + + def test_egg_collection_validation_quantity_negative( + self, client, location_strip1_id, ducks_at_strip1 + ): + """quantity=-1 returns 422.""" + resp = client.post( + "/actions/product-collected", + data={ + "location_id": location_strip1_id, + "quantity": "-1", + "nonce": "test-nonce-789", + }, + ) + + assert resp.status_code == 422 + + def test_egg_collection_validation_location_missing(self, client, ducks_at_strip1): + """Missing location returns 422.""" + resp = client.post( + "/actions/product-collected", + data={ + "quantity": "12", + "nonce": "test-nonce-abc", + }, + ) + + assert resp.status_code == 422 + + def test_egg_collection_no_ducks_at_location(self, client, location_nursery1_id): + """POST to location with no ducks returns 422.""" + resp = client.post( + "/actions/product-collected", + data={ + "location_id": location_nursery1_id, + "quantity": "12", + "nonce": "test-nonce-def", + }, + ) + + assert resp.status_code == 422 + # Error message should indicate no ducks + assert "duck" in resp.text.lower() or "animal" in resp.text.lower() + + def test_egg_collection_location_sticks( + self, client, seeded_db, location_strip1_id, ducks_at_strip1 + ): + """After successful POST, returned form shows same location selected.""" + resp = client.post( + "/actions/product-collected", + data={ + "location_id": location_strip1_id, + "quantity": "6", + "nonce": "test-nonce-ghi", + }, + ) + + assert resp.status_code == 200 + # The response should contain the form with the location pre-selected + # Check for "selected" attribute on the option with our location_id + assert "selected" in resp.text and location_strip1_id in resp.text