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

Add recent events and stats to eggs, feed, and move forms
All checks were successful
Deploy / deploy (push) Successful in 2m40s
Details

Browse Source 
- Create recent_events.py helper for rendering event lists with humanized
  timestamps and deleted event styling (line-through + opacity)
- Query events with ORDER BY ts_utc DESC to show newest first
- Join event_tombstones to detect deleted events
- Fix move form to read animal_ids (not resolved_ids) from entity_refs
- Fix feed purchase format to use total_kg from entity_refs
- Use hx_get with #event-panel-content target for slide-over panel
- Add days-since-last stats for move and feed forms

🤖 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
2026-01-08 21:10:09 +00:00
parent 62cc6c07d1
commit e42eede010
11 changed files with 1102 additions and 33 deletions
Download Patch File Download Diff File Expand all files Collapse all files

176
src/animaltrack/web/routes/eggs.py
View File

@@ -10,6 +10,7 @@ from fasthtml.common import APIRouter, add_toast, to_xml
from starlette.requests import Request
from starlette.responses import HTMLResponse
from animaltrack.events import PRODUCT_COLLECTED, PRODUCT_SOLD
from animaltrack.events.payloads import ProductCollectedPayload, ProductSoldPayload
from animaltrack.events.store import EventStore
from animaltrack.models.reference import UserDefault
@@ -26,6 +27,9 @@ from animaltrack.services.products import ProductService, ValidationError
from animaltrack.web.templates import render_page
from animaltrack.web.templates.eggs import eggs_page
# 30 days in milliseconds
THIRTY_DAYS_MS = 30 * 24 * 60 * 60 * 1000
def _parse_ts_utc(form_value: str | None) -> int:
"""Parse ts_utc from form, defaulting to current time if empty or zero.
@@ -90,6 +94,132 @@ def _get_sellable_products(db):
return [p for p in all_products if p.active and p.sellable]
def _get_recent_events(db: Any, event_type: str, limit: int = 10):
"""Get recent events of a type, most recent first.
Args:
db: Database connection.
event_type: Event type string (e.g., PRODUCT_COLLECTED).
limit: Maximum number of events to return.
Returns:
List of (Event, is_deleted) tuples, most recent first.
"""
import json
from animaltrack.models.events import Event
# Query newest events first with tombstone status
query = """
SELECT e.id, e.type, e.ts_utc, e.actor, e.entity_refs, e.payload, e.version,
CASE WHEN t.target_event_id IS NOT NULL THEN 1 ELSE 0 END as is_deleted
FROM events e
LEFT JOIN event_tombstones t ON e.id = t.target_event_id
WHERE e.type = ?
ORDER BY e.ts_utc DESC
LIMIT ?
"""
rows = db.execute(query, (event_type, limit)).fetchall()
return [
(
Event(
id=row[0],
type=row[1],
ts_utc=row[2],
actor=row[3],
entity_refs=json.loads(row[4]),
payload=json.loads(row[5]),
version=row[6],
),
bool(row[7]),
)
for row in rows
]
def _get_eggs_per_day(db: Any, now_ms: int) -> float | None:
"""Calculate eggs per day over 30-day window.
Args:
db: Database connection.
now_ms: Current timestamp in milliseconds.
Returns:
Eggs per day average, or None if no data.
"""
window_start = now_ms - THIRTY_DAYS_MS
event_store = EventStore(db)
events = event_store.list_events(
event_type=PRODUCT_COLLECTED,
since_utc=window_start,
until_utc=now_ms,
limit=10000,
)
total_eggs = 0
for event in events:
product_code = event.entity_refs.get("product_code", "")
if product_code.startswith("egg."):
total_eggs += event.entity_refs.get("quantity", 0)
if total_eggs == 0:
return None
return total_eggs / 30.0
def _get_sales_stats(db: Any, now_ms: int) -> dict | None:
"""Calculate sales statistics over 30-day window.
Args:
db: Database connection.
now_ms: Current timestamp in milliseconds.
Returns:
Dict with 'total_qty' and 'total_cents', or None if no data.
"""
window_start = now_ms - THIRTY_DAYS_MS
event_store = EventStore(db)
events = event_store.list_events(
event_type=PRODUCT_SOLD,
since_utc=window_start,
until_utc=now_ms,
limit=10000,
)
if not events:
return None
total_qty = 0
total_cents = 0
for event in events:
total_qty += event.entity_refs.get("quantity", 0)
total_cents += event.entity_refs.get("total_price_cents", 0)
return {"total_qty": total_qty, "total_cents": total_cents}
def _get_eggs_display_data(db: Any, locations: list) -> dict:
"""Get all display data for eggs page (events and stats).
Args:
db: Database connection.
locations: List of Location objects for name lookup.
Returns:
Dict with harvest_events, sell_events, eggs_per_day, sales_stats, location_names.
"""
now_ms = int(time.time() * 1000)
return {
"harvest_events": _get_recent_events(db, PRODUCT_COLLECTED, limit=10),
"sell_events": _get_recent_events(db, PRODUCT_SOLD, limit=10),
"eggs_per_day": _get_eggs_per_day(db, now_ms),
"sales_stats": _get_sales_stats(db, now_ms),
"location_names": {loc.id: loc.name for loc in locations},
}
@ar("/")
def egg_index(request: Request):
"""GET / - Eggs page with Harvest/Sell tabs."""
@@ -115,6 +245,9 @@ def egg_index(request: Request):
if defaults:
selected_location_id = defaults.location_id
# Get recent events and stats
display_data = _get_eggs_display_data(db, locations)
return render_page(
request,
eggs_page(
@@ -124,6 +257,7 @@ def egg_index(request: Request):
selected_location_id=selected_location_id,
harvest_action=product_collected,
sell_action=product_sold,
**display_data,
),
title="Eggs - AnimalTrack",
active_nav="eggs",
@@ -152,19 +286,21 @@ async def product_collected(request: Request, session):
# Validate location_id
if not location_id:
return _render_harvest_error(request, locations, products, None, "Please select a location")
return _render_harvest_error(
request, db, locations, products, None, "Please select a location"
)
# Validate quantity
try:
quantity = int(quantity_str)
except ValueError:
return _render_harvest_error(
request, locations, products, location_id, "Quantity must be a number"
request, db, locations, products, location_id, "Quantity must be a number"
)
if quantity < 1:
return _render_harvest_error(
request, locations, products, location_id, "Quantity must be at least 1"
request, db, locations, products, location_id, "Quantity must be at least 1"
)
# Get timestamp - use provided or current (supports backdating)
@@ -175,7 +311,7 @@ async def product_collected(request: Request, session):
if not resolved_ids:
return _render_harvest_error(
request, locations, products, location_id, "No ducks at this location"
request, db, locations, products, location_id, "No ducks at this location"
)
# Create product service
@@ -208,7 +344,7 @@ async def product_collected(request: Request, session):
route="/actions/product-collected",
)
except ValidationError as e:
return _render_harvest_error(request, locations, products, location_id, str(e))
return _render_harvest_error(request, db, locations, products, location_id, str(e))
# Save user defaults (only if user exists in database)
if UserRepository(db).get(actor):
@@ -228,6 +364,9 @@ async def product_collected(request: Request, session):
"success",
)
# Get display data (includes newly created event)
display_data = _get_eggs_display_data(db, locations)
# Success: re-render form with location sticking, qty cleared
return render_page(
request,
@@ -238,6 +377,7 @@ async def product_collected(request: Request, session):
selected_location_id=location_id,
harvest_action=product_collected,
sell_action=product_sold,
**display_data,
),
title="Eggs - AnimalTrack",
active_nav="eggs",
@@ -268,19 +408,19 @@ async def product_sold(request: Request, session):
# Validate product_code
if not product_code:
return _render_sell_error(request, locations, products, None, "Please select a product")
return _render_sell_error(request, db, locations, products, None, "Please select a product")
# Validate quantity
try:
quantity = int(quantity_str)
except ValueError:
return _render_sell_error(
request, locations, products, product_code, "Quantity must be a number"
request, db, locations, products, product_code, "Quantity must be a number"
)
if quantity < 1:
return _render_sell_error(
request, locations, products, product_code, "Quantity must be at least 1"
request, db, locations, products, product_code, "Quantity must be at least 1"
)
# Validate total_price_cents
@@ -288,12 +428,12 @@ async def product_sold(request: Request, session):
total_price_cents = int(total_price_str)
except ValueError:
return _render_sell_error(
request, locations, products, product_code, "Total price must be a number"
request, db, locations, products, product_code, "Total price must be a number"
)
if total_price_cents < 0:
return _render_sell_error(
request, locations, products, product_code, "Total price cannot be negative"
request, db, locations, products, product_code, "Total price cannot be negative"
)
# Get timestamp - use provided or current (supports backdating)
@@ -326,7 +466,7 @@ async def product_sold(request: Request, session):
route="/actions/product-sold",
)
except ValidationError as e:
return _render_sell_error(request, locations, products, product_code, str(e))
return _render_sell_error(request, db, locations, products, product_code, str(e))
# Add success toast with link to event
add_toast(
@@ -335,6 +475,9 @@ async def product_sold(request: Request, session):
"success",
)
# Get display data (includes newly created event)
display_data = _get_eggs_display_data(db, locations)
# Success: re-render form with product sticking
return render_page(
request,
@@ -345,17 +488,19 @@ async def product_sold(request: Request, session):
selected_product_code=product_code,
harvest_action=product_collected,
sell_action=product_sold,
**display_data,
),
title="Eggs - AnimalTrack",
active_nav="eggs",
)
def _render_harvest_error(request, locations, products, selected_location_id, error_message):
def _render_harvest_error(request, db, locations, products, selected_location_id, error_message):
"""Render harvest form with error message.
Args:
request: The HTTP request.
db: Database connection.
locations: List of active locations.
products: List of sellable products.
selected_location_id: Currently selected location.
@@ -364,6 +509,7 @@ def _render_harvest_error(request, locations, products, selected_location_id, er
Returns:
HTMLResponse with 422 status.
"""
display_data = _get_eggs_display_data(db, locations)
return HTMLResponse(
content=to_xml(
render_page(
@@ -376,6 +522,7 @@ def _render_harvest_error(request, locations, products, selected_location_id, er
harvest_error=error_message,
harvest_action=product_collected,
sell_action=product_sold,
**display_data,
),
title="Eggs - AnimalTrack",
active_nav="eggs",
@@ -385,11 +532,12 @@ def _render_harvest_error(request, locations, products, selected_location_id, er
)
def _render_sell_error(request, locations, products, selected_product_code, error_message):
def _render_sell_error(request, db, locations, products, selected_product_code, error_message):
"""Render sell form with error message.
Args:
request: The HTTP request.
db: Database connection.
locations: List of active locations.
products: List of sellable products.
selected_product_code: Currently selected product code.
@@ -398,6 +546,7 @@ def _render_sell_error(request, locations, products, selected_product_code, erro
Returns:
HTMLResponse with 422 status.
"""
display_data = _get_eggs_display_data(db, locations)
return HTMLResponse(
content=to_xml(
render_page(
@@ -410,6 +559,7 @@ def _render_sell_error(request, locations, products, selected_product_code, erro
sell_error=error_message,
harvest_action=product_collected,
sell_action=product_sold,
**display_data,
),
title="Eggs - AnimalTrack",
active_nav="eggs",

197
src/animaltrack/web/routes/feed.py
View File

@@ -10,8 +10,10 @@ from fasthtml.common import APIRouter, add_toast, to_xml
from starlette.requests import Request
from starlette.responses import HTMLResponse
from animaltrack.events import FEED_GIVEN, FEED_PURCHASED
from animaltrack.events.payloads import FeedGivenPayload, FeedPurchasedPayload
from animaltrack.events.store import EventStore
from animaltrack.models.events import Event
from animaltrack.models.reference import UserDefault
from animaltrack.projections import EventLogProjection, ProjectionRegistry
from animaltrack.projections.feed import FeedInventoryProjection
@@ -23,6 +25,9 @@ from animaltrack.services.feed import FeedService, ValidationError
from animaltrack.web.templates import render_page
from animaltrack.web.templates.feed import feed_page
# 30 days in milliseconds
THIRTY_DAYS_MS = 30 * 24 * 60 * 60 * 1000
def _parse_ts_utc(form_value: str | None) -> int:
"""Parse ts_utc from form, defaulting to current time if empty or zero.
@@ -64,6 +69,164 @@ def get_feed_balance(db: Any, feed_type_code: str) -> int | None:
return row[0] if row else None
def _get_recent_events(db: Any, event_type: str, limit: int = 10):
"""Get recent events of a type, most recent first.
Args:
db: Database connection.
event_type: Event type string.
limit: Maximum number of events to return.
Returns:
List of (Event, is_deleted) tuples, most recent first.
"""
import json
# Query newest events first with tombstone status
query = """
SELECT e.id, e.type, e.ts_utc, e.actor, e.entity_refs, e.payload, e.version,
CASE WHEN t.target_event_id IS NOT NULL THEN 1 ELSE 0 END as is_deleted
FROM events e
LEFT JOIN event_tombstones t ON e.id = t.target_event_id
WHERE e.type = ?
ORDER BY e.ts_utc DESC
LIMIT ?
"""
rows = db.execute(query, (event_type, limit)).fetchall()
return [
(
Event(
id=row[0],
type=row[1],
ts_utc=row[2],
actor=row[3],
entity_refs=json.loads(row[4]),
payload=json.loads(row[5]),
version=row[6],
),
bool(row[7]),
)
for row in rows
]
def _get_feed_per_bird_per_day(db: Any, now_ms: int) -> float | None:
"""Calculate feed consumption per bird per day over 30-day window.
Uses global bird-days across all locations.
Args:
db: Database connection.
now_ms: Current timestamp in milliseconds.
Returns:
Feed consumption in grams per bird per day, or None if no data.
"""
window_start = now_ms - THIRTY_DAYS_MS
# Get total feed given in window (all locations)
event_store = EventStore(db)
events = event_store.list_events(
event_type=FEED_GIVEN,
since_utc=window_start,
until_utc=now_ms,
limit=10000,
)
total_kg = sum(e.entity_refs.get("amount_kg", 0) for e in events)
if total_kg == 0:
return None
total_g = total_kg * 1000
# Get total bird-days across all locations
row = db.execute(
"""
SELECT COALESCE(SUM(
MIN(COALESCE(ali.end_utc, :window_end), :window_end) -
MAX(ali.start_utc, :window_start)
), 0) as total_ms
FROM animal_location_intervals ali
JOIN animal_registry ar ON ali.animal_id = ar.animal_id
WHERE ali.start_utc < :window_end
AND (ali.end_utc IS NULL OR ali.end_utc > :window_start)
AND ar.status = 'alive'
""",
{"window_start": window_start, "window_end": now_ms},
).fetchone()
total_ms = row[0] if row else 0
ms_per_day = 24 * 60 * 60 * 1000
bird_days = total_ms // ms_per_day if total_ms else 0
if bird_days == 0:
return None
return total_g / bird_days
def _get_purchase_stats(db: Any, now_ms: int) -> dict | None:
"""Calculate purchase statistics over 30-day window.
Args:
db: Database connection.
now_ms: Current timestamp in milliseconds.
Returns:
Dict with 'total_kg' and 'avg_price_per_kg_cents', or None if no data.
"""
window_start = now_ms - THIRTY_DAYS_MS
event_store = EventStore(db)
events = event_store.list_events(
event_type=FEED_PURCHASED,
since_utc=window_start,
until_utc=now_ms,
limit=10000,
)
if not events:
return None
total_kg = 0
total_cost_cents = 0
for event in events:
bag_size = event.entity_refs.get("bag_size_kg", 0)
bags_count = event.entity_refs.get("bags_count", 0)
bag_price_cents = event.entity_refs.get("bag_price_cents", 0)
total_kg += bag_size * bags_count
total_cost_cents += bag_price_cents * bags_count
if total_kg == 0:
return None
avg_price_per_kg_cents = total_cost_cents / total_kg
return {"total_kg": total_kg, "avg_price_per_kg_cents": avg_price_per_kg_cents}
def _get_feed_display_data(db: Any, locations: list, feed_types: list) -> dict:
"""Get all display data for feed page (events and stats).
Args:
db: Database connection.
locations: List of Location objects for name lookup.
feed_types: List of FeedType objects for name lookup.
Returns:
Dict with display data for feed page.
"""
now_ms = int(time.time() * 1000)
return {
"give_events": _get_recent_events(db, FEED_GIVEN, limit=10),
"purchase_events": _get_recent_events(db, FEED_PURCHASED, limit=10),
"feed_per_bird_per_day_g": _get_feed_per_bird_per_day(db, now_ms),
"purchase_stats": _get_purchase_stats(db, now_ms),
"location_names": {loc.id: loc.name for loc in locations},
"feed_type_names": {ft.code: ft.name for ft in feed_types},
}
@ar("/feed")
def feed_index(request: Request):
"""GET /feed - Feed Quick Capture page."""
@@ -90,6 +253,9 @@ def feed_index(request: Request):
selected_feed_type_code = defaults.feed_type_code
default_amount_kg = defaults.amount_kg
# Get recent events and stats
display_data = _get_feed_display_data(db, locations, feed_types)
return render_page(
request,
feed_page(
@@ -101,6 +267,7 @@ def feed_index(request: Request):
default_amount_kg=default_amount_kg,
give_action=feed_given,
purchase_action=feed_purchased,
**display_data,
),
title="Feed - AnimalTrack",
active_nav="feed",
@@ -135,6 +302,7 @@ async def feed_given(request: Request, session):
if not location_id:
return _render_give_error(
request,
db,
locations,
feed_types,
"Please select a location",
@@ -146,6 +314,7 @@ async def feed_given(request: Request, session):
if not feed_type_code:
return _render_give_error(
request,
db,
locations,
feed_types,
"Please select a feed type",
@@ -159,6 +328,7 @@ async def feed_given(request: Request, session):
except ValueError:
return _render_give_error(
request,
db,
locations,
feed_types,
"Amount must be a number",
@@ -169,6 +339,7 @@ async def feed_given(request: Request, session):
if amount_kg < 1:
return _render_give_error(
request,
db,
locations,
feed_types,
"Amount must be at least 1 kg",
@@ -211,6 +382,7 @@ async def feed_given(request: Request, session):
except ValidationError as e:
return _render_give_error(
request,
db,
locations,
feed_types,
str(e),
@@ -244,6 +416,9 @@ async def feed_given(request: Request, session):
"success",
)
# Get display data (includes newly created event)
display_data = _get_feed_display_data(db, locations, feed_types)
# Success: re-render form with location/type sticking, amount reset
return render_page(
request,
@@ -257,6 +432,7 @@ async def feed_given(request: Request, session):
balance_warning=balance_warning,
give_action=feed_given,
purchase_action=feed_purchased,
**display_data,
),
title="Feed - AnimalTrack",
active_nav="feed",
@@ -286,6 +462,7 @@ async def feed_purchased(request: Request, session):
if not feed_type_code:
return _render_purchase_error(
request,
db,
locations,
feed_types,
"Please select a feed type",
@@ -297,6 +474,7 @@ async def feed_purchased(request: Request, session):
except ValueError:
return _render_purchase_error(
request,
db,
locations,
feed_types,
"Bag size must be a number",
@@ -305,6 +483,7 @@ async def feed_purchased(request: Request, session):
if bag_size_kg < 1:
return _render_purchase_error(
request,
db,
locations,
feed_types,
"Bag size must be at least 1 kg",
@@ -316,6 +495,7 @@ async def feed_purchased(request: Request, session):
except ValueError:
return _render_purchase_error(
request,
db,
locations,
feed_types,
"Bags count must be a number",
@@ -324,6 +504,7 @@ async def feed_purchased(request: Request, session):
if bags_count < 1:
return _render_purchase_error(
request,
db,
locations,
feed_types,
"Bags count must be at least 1",
@@ -336,6 +517,7 @@ async def feed_purchased(request: Request, session):
except ValueError:
return _render_purchase_error(
request,
db,
locations,
feed_types,
"Price must be a number",
@@ -344,6 +526,7 @@ async def feed_purchased(request: Request, session):
if bag_price_cents < 0:
return _render_purchase_error(
request,
db,
locations,
feed_types,
"Price cannot be negative",
@@ -385,6 +568,7 @@ async def feed_purchased(request: Request, session):
except ValidationError as e:
return _render_purchase_error(
request,
db,
locations,
feed_types,
str(e),
@@ -400,6 +584,9 @@ async def feed_purchased(request: Request, session):
"success",
)
# Get display data (includes newly created event)
display_data = _get_feed_display_data(db, locations, feed_types)
# Success: re-render form with fields cleared
return render_page(
request,
@@ -409,6 +596,7 @@ async def feed_purchased(request: Request, session):
active_tab="purchase",
give_action=feed_given,
purchase_action=feed_purchased,
**display_data,
),
title="Feed - AnimalTrack",
active_nav="feed",
@@ -417,6 +605,7 @@ async def feed_purchased(request: Request, session):
def _render_give_error(
request,
db,
locations,
feed_types,
error_message,
@@ -427,6 +616,7 @@ def _render_give_error(
Args:
request: The Starlette request object.
db: Database connection.
locations: List of active locations.
feed_types: List of active feed types.
error_message: Error message to display.
@@ -436,6 +626,7 @@ def _render_give_error(
Returns:
HTMLResponse with 422 status.
"""
display_data = _get_feed_display_data(db, locations, feed_types)
return HTMLResponse(
content=to_xml(
render_page(
@@ -449,6 +640,7 @@ def _render_give_error(
give_error=error_message,
give_action=feed_given,
purchase_action=feed_purchased,
**display_data,
),
title="Feed - AnimalTrack",
active_nav="feed",
@@ -458,11 +650,12 @@ def _render_give_error(
)
def _render_purchase_error(request, locations, feed_types, error_message):
def _render_purchase_error(request, db, locations, feed_types, error_message):
"""Render purchase form with error message.
Args:
request: The Starlette request object.
db: Database connection.
locations: List of active locations.
feed_types: List of active feed types.
error_message: Error message to display.
@@ -470,6 +663,7 @@ def _render_purchase_error(request, locations, feed_types, error_message):
Returns:
HTMLResponse with 422 status.
"""
display_data = _get_feed_display_data(db, locations, feed_types)
return HTMLResponse(
content=to_xml(
render_page(
@@ -481,6 +675,7 @@ def _render_purchase_error(request, locations, feed_types, error_message):
purchase_error=error_message,
give_action=feed_given,
purchase_action=feed_purchased,
**display_data,
),
title="Feed - AnimalTrack",
active_nav="feed",

102
src/animaltrack/web/routes/move.py
View File

@@ -10,8 +10,10 @@ from fasthtml.common import APIRouter, add_toast, to_xml
from starlette.requests import Request
from starlette.responses import HTMLResponse
from animaltrack.events import ANIMAL_MOVED
from animaltrack.events.payloads import AnimalMovedPayload
from animaltrack.events.store import EventStore
from animaltrack.models.events import Event
from animaltrack.projections import EventLogProjection, ProjectionRegistry
from animaltrack.projections.animal_registry import AnimalRegistryProjection
from animaltrack.projections.event_animals import EventAnimalsProjection
@@ -24,6 +26,9 @@ from animaltrack.services.animal import AnimalService, ValidationError
from animaltrack.web.templates import render_page
from animaltrack.web.templates.move import diff_panel, move_form
# Milliseconds per day
MS_PER_DAY = 24 * 60 * 60 * 1000
def _parse_ts_utc(form_value: str | None) -> int:
"""Parse ts_utc from form, defaulting to current time if empty or zero.
@@ -44,6 +49,91 @@ def _parse_ts_utc(form_value: str | None) -> int:
return int(time.time() * 1000)
def _get_recent_move_events(db: Any, limit: int = 10):
"""Get recent AnimalMoved events, most recent first.
Args:
db: Database connection.
limit: Maximum number of events to return.
Returns:
List of (Event, is_deleted) tuples, most recent first.
"""
import json
# Query newest events first with tombstone status
query = """
SELECT e.id, e.type, e.ts_utc, e.actor, e.entity_refs, e.payload, e.version,
CASE WHEN t.target_event_id IS NOT NULL THEN 1 ELSE 0 END as is_deleted
FROM events e
LEFT JOIN event_tombstones t ON e.id = t.target_event_id
WHERE e.type = ?
ORDER BY e.ts_utc DESC
LIMIT ?
"""
rows = db.execute(query, (ANIMAL_MOVED, limit)).fetchall()
return [
(
Event(
id=row[0],
type=row[1],
ts_utc=row[2],
actor=row[3],
entity_refs=json.loads(row[4]),
payload=json.loads(row[5]),
version=row[6],
),
bool(row[7]),
)
for row in rows
]
def _get_days_since_last_move(db: Any, now_ms: int) -> int | None:
"""Calculate days since the last move event.
Args:
db: Database connection.
now_ms: Current timestamp in milliseconds.
Returns:
Number of days since last move, or None if no moves exist.
"""
# Query the most recent move event (newest first)
query = """
SELECT ts_utc FROM events
WHERE type = ?
ORDER BY ts_utc DESC
LIMIT 1
"""
row = db.execute(query, (ANIMAL_MOVED,)).fetchone()
if not row:
return None
diff_ms = now_ms - row[0]
return diff_ms // MS_PER_DAY
def _get_move_display_data(db: Any, locations: list) -> dict:
"""Get all display data for move page (events and stats).
Args:
db: Database connection.
locations: List of Location objects for name lookup.
Returns:
Dict with recent_events, days_since_last_move, location_names.
"""
now_ms = int(time.time() * 1000)
return {
"recent_events": _get_recent_move_events(db, limit=10),
"days_since_last_move": _get_days_since_last_move(db, now_ms),
"location_names": {loc.id: loc.name for loc in locations},
}
# APIRouter for multi-file route organization
ar = APIRouter()
@@ -115,6 +205,9 @@ def move_index(request: Request):
animal_repo = AnimalRepository(db)
animals = animal_repo.get_by_ids(resolved_ids)
# Get recent events and stats
display_data = _get_move_display_data(db, locations)
return render_page(
request,
move_form(
@@ -128,6 +221,7 @@ def move_index(request: Request):
from_location_name=from_location_name,
action=animal_move,
animals=animals,
**display_data,
),
title="Move - AnimalTrack",
active_nav="move",
@@ -298,12 +392,16 @@ async def animal_move(request: Request, session):
"success",
)
# Get display data for fresh form
display_data = _get_move_display_data(db, locations)
# Success: re-render fresh form (nothing sticks per spec)
return render_page(
request,
move_form(
locations,
action=animal_move,
**display_data,
),
title="Move - AnimalTrack",
active_nav="move",
@@ -339,6 +437,9 @@ def _render_error_form(request, db, locations, filter_str, error_message):
from_location_id, from_location_name = _get_from_location(db, resolved_ids, ts_utc)
roster_hash = compute_roster_hash(resolved_ids, from_location_id)
# Get display data for recent events and stats
display_data = _get_move_display_data(db, locations)
return HTMLResponse(
content=to_xml(
render_page(
@@ -354,6 +455,7 @@ def _render_error_form(request, db, locations, filter_str, error_message):
from_location_name=from_location_name,
error=error_message,
action=animal_move,
**display_data,
),
title="Move - AnimalTrack",
active_nav="move",

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

@@ -15,8 +15,10 @@ from monsterui.all import (
)
from ulid import ULID
from animaltrack.models.events import Event
from animaltrack.models.reference import Location, Product
from animaltrack.web.templates.actions import event_datetime_field
from animaltrack.web.templates.recent_events import recent_events_section
def eggs_page(
@@ -29,6 +31,11 @@ def eggs_page(
sell_error: str | None = None,
harvest_action: Callable[..., Any] | str = "/actions/product-collected",
sell_action: Callable[..., Any] | str = "/actions/product-sold",
harvest_events: list[tuple[Event, bool]] | None = None,
sell_events: list[tuple[Event, bool]] | None = None,
eggs_per_day: float | None = None,
sales_stats: dict | None = None,
location_names: dict[str, str] | None = None,
):
"""Create the Eggs page with tabbed forms.
@@ -42,11 +49,18 @@ def eggs_page(
sell_error: Error message for sell form.
harvest_action: Route function or URL for harvest form.
sell_action: Route function or URL for sell form.
harvest_events: Recent ProductCollected events (most recent first).
sell_events: Recent ProductSold events (most recent first).
eggs_per_day: 30-day average eggs per day.
sales_stats: Dict with 'total_qty' and 'total_cents' for 30-day sales.
location_names: Dict mapping location_id to location name for display.
Returns:
Page content with tabbed forms.
"""
harvest_active = active_tab == "harvest"
if location_names is None:
location_names = {}
return Div(
H1("Eggs", cls="text-2xl font-bold mb-6"),
@@ -65,6 +79,9 @@ def eggs_page(
selected_location_id=selected_location_id,
error=harvest_error,
action=harvest_action,
recent_events=harvest_events,
eggs_per_day=eggs_per_day,
location_names=location_names,
),
cls="uk-active" if harvest_active else None,
),
@@ -74,6 +91,8 @@ def eggs_page(
selected_product_code=selected_product_code,
error=sell_error,
action=sell_action,
recent_events=sell_events,
sales_stats=sales_stats,
),
cls=None if harvest_active else "uk-active",
),
@@ -87,7 +106,10 @@ def harvest_form(
selected_location_id: str | None = None,
error: str | None = None,
action: Callable[..., Any] | str = "/actions/product-collected",
) -> Form:
recent_events: list[tuple[Event, bool]] | None = None,
eggs_per_day: float | None = None,
location_names: dict[str, str] | None = None,
) -> Div:
"""Create the Harvest form for egg collection.
Args:
@@ -95,10 +117,18 @@ def harvest_form(
selected_location_id: Pre-selected location ID (sticks after submission).
error: Optional error message to display.
action: Route function or URL string for form submission.
recent_events: Recent (Event, is_deleted) tuples, most recent first.
eggs_per_day: 30-day average eggs per day.
location_names: Dict mapping location_id to location name for display.
Returns:
Form component for recording egg harvests.
Div containing form and recent events section.
"""
if recent_events is None:
recent_events = []
if location_names is None:
location_names = {}
# Build location options
location_options = [
Option(
@@ -123,7 +153,19 @@ def harvest_form(
cls="mb-4",
)
return Form(
# Format function for harvest events
def format_harvest_event(event: Event) -> tuple[str, str]:
quantity = event.entity_refs.get("quantity", 0)
loc_id = event.entity_refs.get("location_id", "")
loc_name = location_names.get(loc_id, "Unknown")
return f"{quantity} eggs from {loc_name}", event.id
# Build stats text
stat_text = None
if eggs_per_day is not None:
stat_text = f"{eggs_per_day:.1f} eggs/day (30-day avg)"
form = Form(
H2("Harvest Eggs", cls="text-xl font-bold mb-4"),
# Error message if present
error_component,
@@ -164,13 +206,25 @@ def harvest_form(
cls="space-y-4",
)
return Div(
form,
recent_events_section(
title="Recent Harvests",
events=recent_events,
format_fn=format_harvest_event,
stat_text=stat_text,
),
)
def sell_form(
products: list[Product],
selected_product_code: str | None = "egg.duck",
error: str | None = None,
action: Callable[..., Any] | str = "/actions/product-sold",
) -> Form:
recent_events: list[tuple[Event, bool]] | None = None,
sales_stats: dict | None = None,
) -> Div:
"""Create the Sell form for recording sales.
Args:
@@ -178,10 +232,17 @@ def sell_form(
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.
recent_events: Recent (Event, is_deleted) tuples, most recent first.
sales_stats: Dict with 'total_qty' and 'total_cents' for 30-day sales.
Returns:
Form component for recording product sales.
Div containing form and recent events section.
"""
if recent_events is None:
recent_events = []
if sales_stats is None:
sales_stats = {}
# Build product options
product_options = [
Option(
@@ -206,7 +267,23 @@ def sell_form(
cls="mb-4",
)
return Form(
# Format function for sell events
def format_sell_event(event: Event) -> tuple[str, str]:
quantity = event.entity_refs.get("quantity", 0)
product_code = event.entity_refs.get("product_code", "")
total_cents = event.entity_refs.get("total_price_cents", 0)
total_eur = total_cents / 100
return f"{quantity} {product_code} for €{total_eur:.2f}", event.id
# Build stats text
stat_text = None
total_qty = sales_stats.get("total_qty")
total_cents = sales_stats.get("total_cents")
if total_qty is not None and total_cents is not None:
total_eur = total_cents / 100
stat_text = f"{total_qty} sold for €{total_eur:.2f} (30-day total)"
form = Form(
H2("Sell Products", cls="text-xl font-bold mb-4"),
# Error message if present
error_component,
@@ -266,6 +343,16 @@ def sell_form(
cls="space-y-4",
)
return Div(
form,
recent_events_section(
title="Recent Sales",
events=recent_events,
format_fn=format_sell_event,
stat_text=stat_text,
),
)
# Keep the old function name for backwards compatibility
def egg_form(
@@ -274,8 +361,8 @@ def egg_form(
error: str | None = None,
action: Callable[..., Any] | str = "/actions/product-collected",
) -> Div:
"""Legacy function - returns harvest form wrapped in a Div.
"""Legacy function - returns harvest form.
Deprecated: Use eggs_page() for the full tabbed interface.
"""
return Div(harvest_form(locations, selected_location_id, error, action))
return harvest_form(locations, selected_location_id, error, action)

124
src/animaltrack/web/templates/feed.py
View File

@@ -15,8 +15,10 @@ from monsterui.all import (
)
from ulid import ULID
from animaltrack.models.events import Event
from animaltrack.models.reference import FeedType, Location
from animaltrack.web.templates.actions import event_datetime_field
from animaltrack.web.templates.recent_events import recent_events_section
def feed_page(
@@ -31,6 +33,12 @@ def feed_page(
balance_warning: str | None = None,
give_action: Callable[..., Any] | str = "/actions/feed-given",
purchase_action: Callable[..., Any] | str = "/actions/feed-purchased",
give_events: list[tuple[Event, bool]] | None = None,
purchase_events: list[tuple[Event, bool]] | None = None,
feed_per_bird_per_day_g: float | None = None,
purchase_stats: dict | None = None,
location_names: dict[str, str] | None = None,
feed_type_names: dict[str, str] | None = None,
):
"""Create the Feed Quick Capture page with tabbed forms.
@@ -46,11 +54,21 @@ def feed_page(
balance_warning: Warning about negative inventory balance.
give_action: Route function or URL for give feed form.
purchase_action: Route function or URL for purchase feed form.
give_events: Recent FeedGiven events (most recent first).
purchase_events: Recent FeedPurchased events (most recent first).
feed_per_bird_per_day_g: Average feed consumption in g/bird/day.
purchase_stats: Dict with 'total_kg' and 'avg_price_per_kg_cents'.
location_names: Dict mapping location_id to location name.
feed_type_names: Dict mapping feed_type_code to feed type name.
Returns:
Page content with tabbed forms.
"""
give_active = active_tab == "give"
if location_names is None:
location_names = {}
if feed_type_names is None:
feed_type_names = {}
return Div(
H1("Feed", cls="text-2xl font-bold mb-6"),
@@ -73,11 +91,22 @@ def feed_page(
error=give_error,
balance_warning=balance_warning,
action=give_action,
recent_events=give_events,
feed_per_bird_per_day_g=feed_per_bird_per_day_g,
location_names=location_names,
feed_type_names=feed_type_names,
),
cls="uk-active" if give_active else None,
),
Li(
purchase_feed_form(feed_types, error=purchase_error, action=purchase_action),
purchase_feed_form(
feed_types,
error=purchase_error,
action=purchase_action,
recent_events=purchase_events,
purchase_stats=purchase_stats,
feed_type_names=feed_type_names,
),
cls=None if give_active else "uk-active",
),
),
@@ -94,7 +123,11 @@ def give_feed_form(
error: str | None = None,
balance_warning: str | None = None,
action: Callable[..., Any] | str = "/actions/feed-given",
) -> Form:
recent_events: list[tuple[Event, bool]] | None = None,
feed_per_bird_per_day_g: float | None = None,
location_names: dict[str, str] | None = None,
feed_type_names: dict[str, str] | None = None,
) -> Div:
"""Create the Give Feed form.
Args:
@@ -106,10 +139,21 @@ def give_feed_form(
error: Error message to display.
balance_warning: Warning about negative balance.
action: Route function or URL for form submission.
recent_events: Recent (Event, is_deleted) tuples, most recent first.
feed_per_bird_per_day_g: Average feed consumption in g/bird/day.
location_names: Dict mapping location_id to location name.
feed_type_names: Dict mapping feed_type_code to feed type name.
Returns:
Form component for giving feed.
Div containing form and recent events section.
"""
if recent_events is None:
recent_events = []
if location_names is None:
location_names = {}
if feed_type_names is None:
feed_type_names = {}
# Build location options
location_options = [
Option(
@@ -154,7 +198,21 @@ def give_feed_form(
cls="mb-4",
)
return Form(
# Format function for feed given events
def format_give_event(event: Event) -> tuple[str, str]:
amount_kg = event.entity_refs.get("amount_kg", 0)
loc_id = event.entity_refs.get("location_id", "")
feed_code = event.entity_refs.get("feed_type_code", "")
loc_name = location_names.get(loc_id, "Unknown")
feed_name = feed_type_names.get(feed_code, feed_code)
return f"{amount_kg}kg {feed_name} to {loc_name}", event.id
# Build stats text
stat_text = None
if feed_per_bird_per_day_g is not None:
stat_text = f"{feed_per_bird_per_day_g:.1f} g/bird/day (30-day avg)"
form = Form(
H2("Give Feed", cls="text-xl font-bold mb-4"),
error_component,
warning_component,
@@ -200,22 +258,45 @@ def give_feed_form(
cls="space-y-4",
)
return Div(
form,
recent_events_section(
title="Recent Feed Given",
events=recent_events,
format_fn=format_give_event,
stat_text=stat_text,
),
)
def purchase_feed_form(
feed_types: list[FeedType],
error: str | None = None,
action: Callable[..., Any] | str = "/actions/feed-purchased",
) -> Form:
recent_events: list[tuple[Event, bool]] | None = None,
purchase_stats: dict | None = None,
feed_type_names: dict[str, str] | None = None,
) -> Div:
"""Create the Purchase Feed form.
Args:
feed_types: List of active feed types.
error: Error message to display.
action: Route function or URL for form submission.
recent_events: Recent (Event, is_deleted) tuples, most recent first.
purchase_stats: Dict with 'total_kg' and 'avg_price_per_kg_cents'.
feed_type_names: Dict mapping feed_type_code to feed type name.
Returns:
Form component for purchasing feed.
Div containing form and recent events section.
"""
if recent_events is None:
recent_events = []
if purchase_stats is None:
purchase_stats = {}
if feed_type_names is None:
feed_type_names = {}
# Build feed type options
feed_type_options = [Option(ft.name, value=ft.code) for ft in feed_types]
feed_type_options.insert(
@@ -230,7 +311,26 @@ def purchase_feed_form(
cls="mb-4",
)
return Form(
# Format function for purchase events
# Note: entity_refs has total_kg, payload has bag details
def format_purchase_event(event: Event) -> tuple[str, str]:
total_kg = event.entity_refs.get("total_kg", 0)
price_per_kg = event.entity_refs.get("price_per_kg_cents", 0)
total_cents = total_kg * price_per_kg
total_eur = total_cents / 100
feed_code = event.entity_refs.get("feed_type_code", "")
feed_name = feed_type_names.get(feed_code, feed_code)
return f"{total_kg}kg {feed_name} for €{total_eur:.2f}", event.id
# Build stats text
stat_text = None
total_kg = purchase_stats.get("total_kg")
avg_price = purchase_stats.get("avg_price_per_kg_cents")
if total_kg is not None and avg_price is not None:
avg_eur = avg_price / 100
stat_text = f"{total_kg}kg purchased, €{avg_eur:.2f}/kg avg (30-day)"
form = Form(
H2("Purchase Feed", cls="text-xl font-bold mb-4"),
error_component,
# Feed type dropdown - using raw Select to fix value handling
@@ -301,3 +401,13 @@ def purchase_feed_form(
method="post",
cls="space-y-4",
)
return Div(
form,
recent_events_section(
title="Recent Purchases",
events=recent_events,
format_fn=format_purchase_event,
stat_text=stat_text,
),
)

2
src/animaltrack/web/templates/location_detail.py
View File

@@ -98,7 +98,7 @@ def recent_events_section(events: list[dict[str, Any]]) -> Div:
),
href=f"/events/{event.get('event_id')}",
hx_get=f"/events/{event.get('event_id')}",
hx_target="#event-panel",
hx_target="#event-panel-content",
hx_swap="innerHTML",
),
cls="py-1",

46
src/animaltrack/web/templates/move.py
View File

@@ -8,9 +8,11 @@ from fasthtml.common import H2, Div, Form, Hidden, Option, P, Span
from monsterui.all import Alert, AlertT, Button, ButtonT, LabelInput, LabelSelect, LabelTextArea
from ulid import ULID
from animaltrack.models.events import Event
from animaltrack.models.reference import Location
from animaltrack.selection.validation import SelectionDiff
from animaltrack.web.templates.actions import event_datetime_field
from animaltrack.web.templates.recent_events import recent_events_section
def move_form(
@@ -25,7 +27,10 @@ def move_form(
error: str | None = None,
action: Callable[..., Any] | str = "/actions/animal-move",
animals: list | None = None,
) -> Form:
recent_events: list[tuple[Event, bool]] | None = None,
days_since_last_move: int | None = None,
location_names: dict[str, str] | None = None,
) -> Div:
"""Create the Move Animals form.
Args:
@@ -40,9 +45,12 @@ def move_form(
error: Optional error message to display.
action: Route function or URL string for form submission.
animals: List of AnimalListItem for checkbox selection (optional).
recent_events: Recent (Event, is_deleted) tuples, most recent first.
days_since_last_move: Number of days since the last move event.
location_names: Dict mapping location_id to location name.
Returns:
Form component for moving animals.
Div containing form and recent events section.
"""
from animaltrack.web.templates.animal_select import animal_checkbox_list
@@ -50,6 +58,10 @@ def move_form(
resolved_ids = []
if animals is None:
animals = []
if recent_events is None:
recent_events = []
if location_names is None:
location_names = {}
# Build destination location options (exclude from_location if set)
location_options = [Option("Select destination...", value="", disabled=True, selected=True)]
@@ -103,7 +115,25 @@ def move_form(
Hidden(name="resolved_ids", value=animal_id) for animal_id in resolved_ids
]
return Form(
# Format function for move events
# Note: entity_refs stores animal_ids, not resolved_ids
def format_move_event(event: Event) -> tuple[str, str]:
to_loc_id = event.entity_refs.get("to_location_id", "")
to_loc_name = location_names.get(to_loc_id, "Unknown")
count = len(event.entity_refs.get("animal_ids", []))
return f"{count} animals to {to_loc_name}", event.id
# Build stats text
stat_text = None
if days_since_last_move is not None:
if days_since_last_move == 0:
stat_text = "Last move: today"
elif days_since_last_move == 1:
stat_text = "Last move: yesterday"
else:
stat_text = f"Last move: {days_since_last_move} days ago"
form = Form(
H2("Move Animals", cls="text-xl font-bold mb-4"),
# Error message if present
error_component,
@@ -152,6 +182,16 @@ def move_form(
cls="space-y-4",
)
return Div(
form,
recent_events_section(
title="Recent Moves",
events=recent_events,
format_fn=format_move_event,
stat_text=stat_text,
),
)
def diff_panel(
diff: SelectionDiff,

119
src/animaltrack/web/templates/recent_events.py Normal file
View File

@@ -0,0 +1,119 @@
# ABOUTME: Helper components for displaying recent events on forms.
# ABOUTME: Provides event list rendering with humanized timestamps and links.
import time
from collections.abc import Callable
from typing import Any
from fasthtml.common import Div, P, Span
from animaltrack.models.events import Event
# Milliseconds per unit
MS_PER_SECOND = 1000
MS_PER_MINUTE = 60 * MS_PER_SECOND
MS_PER_HOUR = 60 * MS_PER_MINUTE
MS_PER_DAY = 24 * MS_PER_HOUR
def humanize_time_ago(ts_utc: int) -> str:
"""Convert a timestamp to a human-readable relative time.
Args:
ts_utc: Timestamp in milliseconds since epoch.
Returns:
Human-readable string like "2h ago", "3 days ago", "just now".
"""
now_ms = int(time.time() * 1000)
diff_ms = now_ms - ts_utc
if diff_ms < 0:
return "in the future"
if diff_ms < MS_PER_MINUTE:
return "just now"
if diff_ms < MS_PER_HOUR:
minutes = diff_ms // MS_PER_MINUTE
return f"{minutes}m ago"
if diff_ms < MS_PER_DAY:
hours = diff_ms // MS_PER_HOUR
return f"{hours}h ago"
days = diff_ms // MS_PER_DAY
if days == 1:
return "1 day ago"
return f"{days} days ago"
def recent_events_section(
title: str,
events: list[tuple[Event, bool]],
format_fn: Callable[[Event], tuple[str, str]],
stat_text: str | None = None,
) -> Div:
"""Render a section with stats and recent events.
Args:
title: Section title (e.g., "Recent Harvests").
events: List of (Event, is_deleted) tuples, most recent first.
format_fn: Function that takes an Event and returns (description, event_id).
Description is the text to display, event_id for linking.
stat_text: Optional statistics text to show above the event list.
Returns:
Div containing the stats and event list.
"""
children: list[Any] = []
# Stats section
if stat_text:
children.append(
Div(
P(stat_text, cls="text-sm text-stone-600 dark:text-stone-400"),
cls="mb-3 p-2 bg-stone-50 dark:bg-stone-800 rounded",
)
)
# Title
children.append(
P(
title,
cls="text-xs font-semibold text-stone-500 dark:text-stone-400 uppercase tracking-wide mb-2",
)
)
# Event list
if not events:
children.append(
P("No recent events", cls="text-sm text-stone-400 dark:text-stone-500 italic")
)
else:
event_items = []
for event, is_deleted in events:
description, event_id = format_fn(event)
time_ago = humanize_time_ago(event.ts_utc)
# Apply deleted styling if tombstoned
deleted_cls = "line-through opacity-50" if is_deleted else ""
event_items.append(
Div(
Div(
Span(description, cls=f"flex-1 {deleted_cls}"),
Span(
time_ago, cls=f"text-stone-400 dark:text-stone-500 ml-2 {deleted_cls}"
),
cls="flex justify-between items-center",
),
cls="block text-sm py-1 px-2 rounded hover:bg-stone-100 dark:hover:bg-stone-800 transition-colors cursor-pointer",
hx_get=f"/events/{event_id}",
hx_target="#event-panel-content",
hx_swap="innerHTML",
)
)
children.append(Div(*event_items, cls="space-y-1"))
return Div(*children, cls="mt-6 pt-4 border-t border-stone-200 dark:border-stone-700")

57
tests/test_web_eggs.py
View File

@@ -211,3 +211,60 @@ class TestEggCollection:
# 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
class TestEggsRecentEvents:
"""Tests for recent events display on eggs page."""
def test_harvest_tab_shows_recent_events_section(self, client):
"""Harvest tab shows Recent Harvests section."""
resp = client.get("/")
assert resp.status_code == 200
assert "Recent Harvests" in resp.text
def test_sell_tab_shows_recent_events_section(self, client):
"""Sell tab shows Recent Sales section."""
resp = client.get("/?tab=sell")
assert resp.status_code == 200
assert "Recent Sales" in resp.text
def test_harvest_event_appears_in_recent(
self, client, seeded_db, location_strip1_id, ducks_at_strip1
):
"""Newly created harvest event appears in recent events list."""
resp = client.post(
"/actions/product-collected",
data={
"location_id": location_strip1_id,
"quantity": "12",
"nonce": "test-nonce-recent-1",
},
)
assert resp.status_code == 200
# Recent events should include the newly created event
# Check for event link pattern
assert "/events/" in resp.text
def test_harvest_event_links_to_detail(
self, client, seeded_db, location_strip1_id, ducks_at_strip1
):
"""Harvest events in recent list link to event detail page."""
# Create an event
resp = client.post(
"/actions/product-collected",
data={
"location_id": location_strip1_id,
"quantity": "8",
"nonce": "test-nonce-recent-2",
},
)
assert resp.status_code == 200
# Get the event ID from DB
event_row = seeded_db.execute(
"SELECT id FROM events WHERE type = 'ProductCollected' ORDER BY id DESC LIMIT 1"
).fetchone()
event_id = event_row[0]
# The response should contain a link to the event detail
assert f"/events/{event_id}" in resp.text

96
tests/test_web_feed.py
View File

@@ -360,3 +360,99 @@ class TestInventoryWarning:
assert resp.status_code in [200, 302, 303]
# The response should contain a warning about negative inventory
assert "warning" in resp.text.lower() or "negative" in resp.text.lower()
class TestFeedRecentEvents:
"""Tests for recent events display on feed page."""
def test_give_tab_shows_recent_events_section(self, client):
"""Give Feed tab shows Recent Feed Given section."""
resp = client.get("/feed")
assert resp.status_code == 200
assert "Recent Feed Given" in resp.text
def test_purchase_tab_shows_recent_events_section(self, client):
"""Purchase Feed tab shows Recent Purchases section."""
resp = client.get("/feed?tab=purchase")
assert resp.status_code == 200
assert "Recent Purchases" in resp.text
def test_give_feed_event_appears_in_recent(
self, client, seeded_db, location_strip1_id, feed_purchase_in_db
):
"""Newly created feed given event appears in recent events list."""
resp = client.post(
"/actions/feed-given",
data={
"location_id": location_strip1_id,
"feed_type_code": "layer",
"amount_kg": "5",
"nonce": "test-nonce-recent-feed-1",
},
)
assert resp.status_code == 200
# Recent events should include the newly created event
assert "/events/" in resp.text
def test_give_feed_event_links_to_detail(
self, client, seeded_db, location_strip1_id, feed_purchase_in_db
):
"""Feed given events in recent list link to event detail page."""
resp = client.post(
"/actions/feed-given",
data={
"location_id": location_strip1_id,
"feed_type_code": "layer",
"amount_kg": "5",
"nonce": "test-nonce-recent-feed-2",
},
)
assert resp.status_code == 200
# Get the event ID from DB
event_row = seeded_db.execute(
"SELECT id FROM events WHERE type = 'FeedGiven' ORDER BY id DESC LIMIT 1"
).fetchone()
event_id = event_row[0]
# The response should contain a link to the event detail
assert f"/events/{event_id}" in resp.text
def test_purchase_event_appears_in_recent(self, client, seeded_db):
"""Newly created purchase event appears in recent events list."""
resp = client.post(
"/actions/feed-purchased",
data={
"feed_type_code": "layer",
"bag_size_kg": "20",
"bags_count": "2",
"bag_price_euros": "24.00",
"nonce": "test-nonce-recent-purchase-1",
},
)
# The route returns purchase tab active after purchase
assert resp.status_code == 200
assert "/events/" in resp.text
def test_purchase_event_links_to_detail(self, client, seeded_db):
"""Purchase events in recent list link to event detail page."""
resp = client.post(
"/actions/feed-purchased",
data={
"feed_type_code": "layer",
"bag_size_kg": "20",
"bags_count": "2",
"bag_price_euros": "24.00",
"nonce": "test-nonce-recent-purchase-2",
},
)
assert resp.status_code == 200
# Get the event ID from DB
event_row = seeded_db.execute(
"SELECT id FROM events WHERE type = 'FeedPurchased' ORDER BY id DESC LIMIT 1"
).fetchone()
event_id = event_row[0]
# The response should contain a link to the event detail
assert f"/events/{event_id}" in resp.text

113
tests/test_web_move.py
View File

@@ -472,3 +472,116 @@ class TestMoveAnimalMismatch:
payload = json.loads(event_row[0])
# Should have moved 3 animals (5 original - 2 moved by client B)
assert len(payload["resolved_ids"]) == 3
class TestMoveRecentEvents:
"""Tests for recent events display on move page."""
def test_move_form_shows_recent_events_section(self, client):
"""Move form shows Recent Moves section."""
resp = client.get("/move")
assert resp.status_code == 200
assert "Recent Moves" in resp.text
def test_move_event_appears_in_recent(
self,
client,
seeded_db,
animal_service,
location_strip1_id,
location_strip2_id,
ducks_at_strip1,
):
"""Newly created move event appears in recent events list."""
ts_utc = int(time.time() * 1000)
filter_str = 'location:"Strip 1"'
filter_ast = parse_filter(filter_str)
resolution = resolve_filter(seeded_db, filter_ast, ts_utc)
roster_hash = compute_roster_hash(resolution.animal_ids, location_strip1_id)
resp = client.post(
"/actions/animal-move",
data={
"filter": filter_str,
"to_location_id": location_strip2_id,
"resolved_ids": resolution.animal_ids,
"roster_hash": roster_hash,
"from_location_id": location_strip1_id,
"ts_utc": str(ts_utc),
"nonce": "test-nonce-recent-move-1",
},
)
assert resp.status_code == 200
# Recent events should include the newly created event
assert "/events/" in resp.text
def test_move_event_links_to_detail(
self,
client,
seeded_db,
animal_service,
location_strip1_id,
location_strip2_id,
ducks_at_strip1,
):
"""Move events in recent list link to event detail page."""
ts_utc = int(time.time() * 1000)
filter_str = 'location:"Strip 1"'
filter_ast = parse_filter(filter_str)
resolution = resolve_filter(seeded_db, filter_ast, ts_utc)
roster_hash = compute_roster_hash(resolution.animal_ids, location_strip1_id)
resp = client.post(
"/actions/animal-move",
data={
"filter": filter_str,
"to_location_id": location_strip2_id,
"resolved_ids": resolution.animal_ids,
"roster_hash": roster_hash,
"from_location_id": location_strip1_id,
"ts_utc": str(ts_utc),
"nonce": "test-nonce-recent-move-2",
},
)
assert resp.status_code == 200
# Get the event ID from DB
event_row = seeded_db.execute(
"SELECT id FROM events WHERE type = 'AnimalMoved' ORDER BY id DESC LIMIT 1"
).fetchone()
event_id = event_row[0]
# The response should contain a link to the event detail
assert f"/events/{event_id}" in resp.text
def test_days_since_last_move_shows_today(
self,
client,
seeded_db,
animal_service,
location_strip1_id,
location_strip2_id,
ducks_at_strip1,
):
"""After a move today, shows 'Last move: today'."""
ts_utc = int(time.time() * 1000)
filter_str = 'location:"Strip 1"'
filter_ast = parse_filter(filter_str)
resolution = resolve_filter(seeded_db, filter_ast, ts_utc)
roster_hash = compute_roster_hash(resolution.animal_ids, location_strip1_id)
resp = client.post(
"/actions/animal-move",
data={
"filter": filter_str,
"to_location_id": location_strip2_id,
"resolved_ids": resolution.animal_ids,
"roster_hash": roster_hash,
"from_location_id": location_strip1_id,
"ts_utc": str(ts_utc),
"nonce": "test-nonce-recent-move-3",
},
)
assert resp.status_code == 200
# Stats should show "Last move: today"
assert "Last move: today" in resp.text