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

feat: implement product-sold route (Step 9.2)

Browse Source 
Add POST /actions/product-sold route for recording product sales.

Changes:
- Create web/templates/products.py with product_sold_form
- Create web/routes/products.py with GET /sell and POST /actions/product-sold
- Wire up routes in __init__.py and app.py
- Add "Record Sale" link to Egg Quick Capture page
- Add comprehensive tests for form rendering and sale recording

The form allows selling any sellable product with quantity and price,
and calculates unit_price_cents using floor division. Defaults to
egg.duck product as per spec.

🤖 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 14:16:12 +00:00
parent 943383a620
commit 0eef3ed7cb
7 changed files with 603 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
PLAN.md
View File

@@ -353,11 +353,11 @@ Check off items as completed. Each phase builds on the previous.
- [x] Write tests for each action
- [x] **Commit checkpoint**: 29ea3e2
### Step 9.2: Product Sold Route
- [ ] POST /actions/product-sold
- [ ] Create form template
- [ ] Write tests: sale creates event, unit price calculated
- [ ] **Commit checkpoint**
### Step 9.2: Product Sold Route ✅
- [x] POST /actions/product-sold
- [x] Create form template
- [x] Write tests: sale creates event, unit price calculated
- [x] **Commit checkpoint**
### Step 9.3: User Defaults
- [ ] Create migration for user_defaults table

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

@@ -27,6 +27,7 @@ from animaltrack.web.routes import (
register_feed_routes,
register_health_routes,
register_move_routes,
register_product_routes,
register_registry_routes,
)
@@ -152,6 +153,7 @@ def create_app(
register_events_routes(rt, app)
register_feed_routes(rt, app)
register_move_routes(rt, app)
register_product_routes(rt, app)
register_registry_routes(rt, app)
return app, rt

2
src/animaltrack/web/routes/__init__.py
View File

@@ -8,6 +8,7 @@ from animaltrack.web.routes.events import register_events_routes
from animaltrack.web.routes.feed import register_feed_routes
from animaltrack.web.routes.health import register_health_routes
from animaltrack.web.routes.move import register_move_routes
from animaltrack.web.routes.products import register_product_routes
from animaltrack.web.routes.registry import register_registry_routes
__all__ = [
@@ -18,5 +19,6 @@ __all__ = [
"register_feed_routes",
"register_health_routes",
"register_move_routes",
"register_product_routes",
"register_registry_routes",
]

185
src/animaltrack/web/routes/products.py Normal file
View File

@@ -0,0 +1,185 @@
# ABOUTME: Routes for Product Sold functionality.
# ABOUTME: Handles GET /sell form and POST /actions/product-sold.
from __future__ import annotations
import json
import time
from fasthtml.common import to_xml
from starlette.requests import Request
from starlette.responses import HTMLResponse
from animaltrack.events.payloads import ProductSoldPayload
from animaltrack.events.store import EventStore
from animaltrack.projections import EventLogProjection, ProjectionRegistry
from animaltrack.projections.products import ProductsProjection
from animaltrack.repositories.products import ProductRepository
from animaltrack.services.products import ProductService, ValidationError
from animaltrack.web.templates import page
from animaltrack.web.templates.products import product_sold_form
def _get_sellable_products(db):
"""Get list of active, sellable products.
Args:
db: Database connection.
Returns:
List of sellable Product objects.
"""
repo = ProductRepository(db)
all_products = repo.list_all()
return [p for p in all_products if p.active and p.sellable]
def sell_index(request: Request):
"""GET /sell - Product Sold form."""
db = request.app.state.db
products = _get_sellable_products(db)
# Check for pre-selected product from query params (defaults to egg.duck)
selected_product_code = request.query_params.get("product_code", "egg.duck")
return page(
product_sold_form(
products, selected_product_code=selected_product_code, action=product_sold
),
title="Sell - AnimalTrack",
active_nav=None,
)
async def product_sold(request: Request):
"""POST /actions/product-sold - Record product sale."""
db = request.app.state.db
form = await request.form()
# Extract form data
product_code = form.get("product_code", "")
quantity_str = form.get("quantity", "0")
total_price_str = form.get("total_price_cents", "0")
buyer = form.get("buyer") or None
notes = form.get("notes") or None
nonce = form.get("nonce")
# Get products for potential re-render
products = _get_sellable_products(db)
# Validate product_code
if not product_code:
return _render_error_form(products, None, "Please select a product")
# Validate quantity
try:
quantity = int(quantity_str)
except ValueError:
return _render_error_form(products, product_code, "Quantity must be a number")
if quantity < 1:
return _render_error_form(products, product_code, "Quantity must be at least 1")
# Validate total_price_cents
try:
total_price_cents = int(total_price_str)
except ValueError:
return _render_error_form(products, product_code, "Total price must be a number")
if total_price_cents < 0:
return _render_error_form(products, product_code, "Total price cannot be negative")
# Get current timestamp
ts_utc = int(time.time() * 1000)
# Create product service
event_store = EventStore(db)
registry = ProjectionRegistry()
registry.register(ProductsProjection(db))
registry.register(EventLogProjection(db))
product_service = ProductService(db, event_store, registry)
# Create payload
payload = ProductSoldPayload(
product_code=product_code,
quantity=quantity,
total_price_cents=total_price_cents,
buyer=buyer,
notes=notes,
)
# Get actor from auth
auth = request.scope.get("auth")
actor = auth.username if auth else "unknown"
# Sell product
try:
product_service.sell_product(
payload=payload,
ts_utc=ts_utc,
actor=actor,
nonce=nonce,
route="/actions/product-sold",
)
except ValidationError as e:
return _render_error_form(products, product_code, str(e))
# Success: re-render form with product sticking, other fields cleared
response = HTMLResponse(
content=to_xml(
page(
product_sold_form(
products, selected_product_code=product_code, action=product_sold
),
title="Sell - AnimalTrack",
active_nav=None,
)
),
)
# Add toast trigger header
response.headers["HX-Trigger"] = json.dumps(
{"showToast": {"message": f"Recorded sale of {quantity} {product_code}", "type": "success"}}
)
return response
def register_product_routes(rt, app):
"""Register product routes.
Args:
rt: FastHTML route decorator.
app: FastHTML application instance.
"""
rt("/sell")(sell_index)
rt("/actions/product-sold", methods=["POST"])(product_sold)
def _render_error_form(products, selected_product_code, error_message):
"""Render form with error message.
Args:
products: List of sellable products.
selected_product_code: Currently selected product code.
error_message: Error message to display.
Returns:
HTMLResponse with 422 status.
"""
return HTMLResponse(
content=to_xml(
page(
product_sold_form(
products,
selected_product_code=selected_product_code,
error=error_message,
action=product_sold,
),
title="Sell - AnimalTrack",
active_nav=None,
)
),
status_code=422,
)

24
src/animaltrack/web/templates/eggs.py
View File

@@ -4,7 +4,7 @@
from collections.abc import Callable
from typing import Any
from fasthtml.common import H2, Form, Hidden, Option
from fasthtml.common import H2, A, Div, Form, Hidden, Option, P
from monsterui.all import Button, ButtonT, LabelInput, LabelSelect, LabelTextArea
from ulid import ULID
@@ -16,8 +16,8 @@ def egg_form(
selected_location_id: str | None = None,
error: str | None = None,
action: Callable[..., Any] | str = "/actions/product-collected",
) -> Form:
"""Create the Egg Quick Capture form.
) -> Div:
"""Create the Egg Quick Capture form with Record Sale link.
Args:
locations: List of active locations for the dropdown.
@@ -26,7 +26,7 @@ def egg_form(
action: Route function or URL string for form submission.
Returns:
Form component for egg collection.
Div containing the form and a link to Record Sale page.
"""
# Build location options
location_options = [
@@ -47,14 +47,12 @@ def egg_form(
# 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(
form = Form(
H2("Record Eggs", cls="text-xl font-bold mb-4"),
# Error message if present
error_component,
@@ -91,3 +89,15 @@ def egg_form(
method="post",
cls="space-y-4",
)
return Div(
form,
Div(
A(
"Record Sale",
href="/sell",
cls="text-sm text-blue-400 hover:text-blue-300 underline",
),
cls="mt-4 text-center",
),
)

112
src/animaltrack/web/templates/products.py Normal file
View File

@@ -0,0 +1,112 @@
# ABOUTME: Templates for Product Sold form.
# ABOUTME: Provides form components for recording product sales.
from collections.abc import Callable
from typing import Any
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 Product
def product_sold_form(
products: list[Product],
selected_product_code: str | None = "egg.duck",
error: str | None = None,
action: Callable[..., Any] | str = "/actions/product-sold",
) -> Form:
"""Create the Product Sold form.
Args:
products: List of sellable products for the dropdown.
selected_product_code: Pre-selected product code (defaults to egg.duck).
error: Optional error message to display.
action: Route function or URL string for form submission.
Returns:
Form component for recording product sales.
"""
# Build product options
product_options = [
Option(
f"{product.name} ({product.code})",
value=product.code,
selected=(product.code == selected_product_code),
)
for product in products
]
# Add placeholder option if no product is selected
if selected_product_code is None:
product_options.insert(
0, Option("Select a product...", 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 Sale", cls="text-xl font-bold mb-4"),
# Error message if present
error_component,
# Product dropdown
LabelSelect(
*product_options,
label="Product",
id="product_code",
name="product_code",
),
# Quantity input (integer only, min=1)
LabelInput(
"Quantity",
id="quantity",
name="quantity",
type="number",
min="1",
step="1",
placeholder="Number of items sold",
required=True,
),
# Total price in cents
LabelInput(
"Total Price (cents)",
id="total_price_cents",
name="total_price_cents",
type="number",
min="0",
step="1",
placeholder="Total price in cents",
required=True,
),
# Optional buyer
LabelInput(
"Buyer",
id="buyer",
name="buyer",
type="text",
placeholder="Optional buyer name",
),
# Optional notes
LabelTextArea(
"Notes",
id="notes",
placeholder="Optional notes",
),
# Hidden nonce for idempotency
Hidden(name="nonce", value=str(ULID())),
# Submit button
Button("Record Sale", type="submit", cls=ButtonT.primary),
# Form submission via standard action/method (hx-boost handles AJAX)
action=action,
method="post",
cls="space-y-4",
)

280
tests/test_web_products_sold.py Normal file
View File

@@ -0,0 +1,280 @@
# ABOUTME: Tests for Product Sold web routes.
# ABOUTME: Covers GET /sell form rendering and POST /actions/product-sold.
import os
import pytest
from starlette.testclient import TestClient
def make_test_settings(
csrf_secret: str = "test-secret",
trusted_proxy_ips: str = "127.0.0.1",
dev_mode: bool = True,
):
"""Create Settings for testing by setting env vars temporarily."""
from animaltrack.config import Settings
old_env = os.environ.copy()
try:
os.environ["CSRF_SECRET"] = csrf_secret
os.environ["TRUSTED_PROXY_IPS"] = trusted_proxy_ips
os.environ["DEV_MODE"] = str(dev_mode).lower()
return Settings()
finally:
os.environ.clear()
os.environ.update(old_env)
@pytest.fixture
def client(seeded_db):
"""Create a test client for the app."""
from animaltrack.web.app import create_app
settings = make_test_settings(trusted_proxy_ips="testclient")
app, rt = create_app(settings=settings, db=seeded_db)
return TestClient(app, raise_server_exceptions=True)
class TestProductSoldFormRendering:
"""Tests for GET /sell product sold form."""
def test_sell_form_renders(self, client):
"""GET /sell returns 200 with form elements."""
resp = client.get("/sell")
assert resp.status_code == 200
assert "Record Sale" in resp.text or "Sell" in resp.text
def test_sell_form_shows_products(self, client):
"""Form has product dropdown with sellable products."""
resp = client.get("/sell")
assert resp.status_code == 200
# Check for seeded sellable product names
assert "Duck Egg" in resp.text or "egg.duck" in resp.text
def test_sell_form_has_quantity_field(self, client):
"""Form has quantity input field."""
resp = client.get("/sell")
assert resp.status_code == 200
assert 'name="quantity"' in resp.text or 'id="quantity"' in resp.text
def test_sell_form_has_total_price_field(self, client):
"""Form has total_price_cents input field."""
resp = client.get("/sell")
assert resp.status_code == 200
assert 'name="total_price_cents"' in resp.text or 'id="total_price_cents"' in resp.text
def test_sell_form_has_buyer_field(self, client):
"""Form has optional buyer input field."""
resp = client.get("/sell")
assert resp.status_code == 200
assert 'name="buyer"' in resp.text or 'id="buyer"' in resp.text
def test_sell_form_default_product_egg_duck(self, client):
"""Form defaults to egg.duck product (selected)."""
resp = client.get("/sell")
assert resp.status_code == 200
# egg.duck should be the selected option
# Check for either value="egg.duck" with selected, or selected before egg.duck
assert "egg.duck" in resp.text
class TestProductSold:
"""Tests for POST /actions/product-sold."""
def test_product_sold_creates_event(self, client, seeded_db):
"""POST creates ProductSold event with correct data."""
resp = client.post(
"/actions/product-sold",
data={
"product_code": "egg.duck",
"quantity": "30",
"total_price_cents": "1500",
"buyer": "Local Market",
"notes": "Weekly sale",
"nonce": "test-nonce-sold-1",
},
)
# 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, entity_refs FROM events WHERE type = 'ProductSold' ORDER BY id DESC LIMIT 1"
).fetchone()
assert event_row is not None
assert event_row[0] == "ProductSold"
def test_product_sold_unit_price_calculated(self, client, seeded_db):
"""Unit price is calculated as floor(total/qty)."""
resp = client.post(
"/actions/product-sold",
data={
"product_code": "egg.duck",
"quantity": "30",
"total_price_cents": "1500",
"nonce": "test-nonce-sold-2",
},
)
assert resp.status_code in [200, 302, 303]
# Verify unit_price_cents in entity_refs
import json
event_row = seeded_db.execute(
"SELECT entity_refs FROM events WHERE type = 'ProductSold' ORDER BY id DESC LIMIT 1"
).fetchone()
entity_refs = json.loads(event_row[0])
assert entity_refs["unit_price_cents"] == 50 # 1500 / 30 = 50
def test_product_sold_unit_price_floor_division(self, client, seeded_db):
"""Unit price uses floor division (rounds down)."""
resp = client.post(
"/actions/product-sold",
data={
"product_code": "egg.duck",
"quantity": "3",
"total_price_cents": "1000",
"nonce": "test-nonce-sold-3",
},
)
assert resp.status_code in [200, 302, 303]
import json
event_row = seeded_db.execute(
"SELECT entity_refs FROM events WHERE type = 'ProductSold' ORDER BY id DESC LIMIT 1"
).fetchone()
entity_refs = json.loads(event_row[0])
assert entity_refs["unit_price_cents"] == 333 # floor(1000 / 3) = 333
def test_product_sold_validation_quantity_zero(self, client):
"""quantity=0 returns 422."""
resp = client.post(
"/actions/product-sold",
data={
"product_code": "egg.duck",
"quantity": "0",
"total_price_cents": "1000",
"nonce": "test-nonce-sold-4",
},
)
assert resp.status_code == 422
def test_product_sold_validation_quantity_negative(self, client):
"""quantity=-1 returns 422."""
resp = client.post(
"/actions/product-sold",
data={
"product_code": "egg.duck",
"quantity": "-1",
"total_price_cents": "1000",
"nonce": "test-nonce-sold-5",
},
)
assert resp.status_code == 422
def test_product_sold_validation_price_negative(self, client):
"""Negative price returns 422."""
resp = client.post(
"/actions/product-sold",
data={
"product_code": "egg.duck",
"quantity": "10",
"total_price_cents": "-100",
"nonce": "test-nonce-sold-6",
},
)
assert resp.status_code == 422
def test_product_sold_validation_missing_product(self, client):
"""Missing product_code returns 422."""
resp = client.post(
"/actions/product-sold",
data={
"quantity": "10",
"total_price_cents": "1000",
"nonce": "test-nonce-sold-7",
},
)
assert resp.status_code == 422
def test_product_sold_invalid_product(self, client):
"""Non-existent product returns 422."""
resp = client.post(
"/actions/product-sold",
data={
"product_code": "invalid.product",
"quantity": "10",
"total_price_cents": "1000",
"nonce": "test-nonce-sold-8",
},
)
assert resp.status_code == 422
def test_product_sold_success_shows_toast(self, client):
"""Successful sale returns response with toast trigger."""
resp = client.post(
"/actions/product-sold",
data={
"product_code": "egg.duck",
"quantity": "12",
"total_price_cents": "600",
"nonce": "test-nonce-sold-9",
},
)
assert resp.status_code == 200
# Check for HX-Trigger header with showToast
hx_trigger = resp.headers.get("HX-Trigger")
assert hx_trigger is not None
assert "showToast" in hx_trigger
def test_product_sold_optional_buyer(self, client, seeded_db):
"""Buyer field is optional."""
resp = client.post(
"/actions/product-sold",
data={
"product_code": "egg.duck",
"quantity": "10",
"total_price_cents": "500",
"nonce": "test-nonce-sold-10",
},
)
assert resp.status_code in [200, 302, 303]
# Event should still be created
event_row = seeded_db.execute(
"SELECT type FROM events WHERE type = 'ProductSold' ORDER BY id DESC LIMIT 1"
).fetchone()
assert event_row is not None
def test_product_sold_optional_notes(self, client, seeded_db):
"""Notes field is optional."""
resp = client.post(
"/actions/product-sold",
data={
"product_code": "egg.duck",
"quantity": "10",
"total_price_cents": "500",
"buyer": "Test Buyer",
"nonce": "test-nonce-sold-11",
},
)
assert resp.status_code in [200, 302, 303]
# Event should still be created
event_row = seeded_db.execute(
"SELECT type FROM events WHERE type = 'ProductSold' ORDER BY id DESC LIMIT 1"
).fetchone()
assert event_row is not None