fix(meal-planner): validate meal_type path param, enforce store whitelist safety, add week_start date validation, make PrepTask frozen

- upsert_slot: raise 422 immediately if meal_type not in VALID_MEAL_TYPES
- update_prep_task: assert whitelist safety contract after dict comprehension
- CreatePlanRequest: week_start typed as date with must_be_monday validator; str() cast at call site
- PrepTask: frozen=True; build_prep_tasks rewired to use (priority, kwargs) tuples so frozen instances are built with correct sequence_order in one pass (no post-construction mutation)
- Move deferred import json to file-level in meal_plans.py
- Fix test dates: "2026-04-14" was a Tuesday; updated request bodies to "2026-04-13" (Monday)
This commit is contained in:
pyr0ball 2026-04-12 13:51:50 -07:00
parent bfc63f1fc9
commit 482666907b
5 changed files with 41 additions and 24 deletions

View file

@ -3,6 +3,7 @@
from __future__ import annotations
import asyncio
import json
from datetime import date
from fastapi import APIRouter, Depends, HTTPException
@ -48,7 +49,6 @@ def _slot_summary(row: dict) -> SlotSummary:
def _plan_summary(plan: dict, slots: list[dict]) -> PlanSummary:
meal_types = plan.get("meal_types") or ["dinner"]
if isinstance(meal_types, str):
import json
meal_types = json.loads(meal_types)
return PlanSummary(
id=plan["id"],
@ -87,7 +87,7 @@ async def create_plan(
else:
meal_types = ["dinner"]
plan = await asyncio.to_thread(store.create_meal_plan, req.week_start, meal_types)
plan = await asyncio.to_thread(store.create_meal_plan, str(req.week_start), meal_types)
slots = await asyncio.to_thread(store.get_plan_slots, plan["id"])
return _plan_summary(plan, slots)
@ -124,6 +124,8 @@ async def upsert_slot(
session: CloudUser = Depends(get_session),
store: Store = Depends(get_store),
) -> SlotSummary:
if meal_type not in VALID_MEAL_TYPES:
raise HTTPException(status_code=422, detail=f"Invalid meal_type '{meal_type}'.")
plan = await asyncio.to_thread(store.get_meal_plan, plan_id)
if plan is None:
raise HTTPException(status_code=404, detail="Plan not found.")

View file

@ -1115,6 +1115,7 @@ class Store:
def update_prep_task(self, task_id: int, **kwargs: object) -> dict | None:
allowed = {"duration_minutes", "sequence_order", "notes", "equipment"}
updates = {k: v for k, v in kwargs.items() if k in allowed and v is not None}
assert all(k in allowed for k in updates), f"Unexpected column(s): {set(updates) - allowed}"
if not updates:
return self._fetch_one("SELECT * FROM prep_tasks WHERE id = ?", (task_id,))
set_clause = ", ".join(f"{k} = ?" for k in updates)

View file

@ -2,16 +2,25 @@
"""Pydantic schemas for meal planning endpoints."""
from __future__ import annotations
from pydantic import BaseModel, Field
from datetime import date as _date
from pydantic import BaseModel, Field, field_validator
VALID_MEAL_TYPES = {"breakfast", "lunch", "dinner", "snack"}
class CreatePlanRequest(BaseModel):
week_start: str # ISO date string, e.g. "2026-04-14" (must be Monday)
week_start: _date
meal_types: list[str] = Field(default_factory=lambda: ["dinner"])
@field_validator("week_start")
@classmethod
def must_be_monday(cls, v: _date) -> _date:
if v.weekday() != 0:
raise ValueError("week_start must be a Monday (weekday 0)")
return v
class UpsertSlotRequest(BaseModel):
recipe_id: int | None = None

View file

@ -6,13 +6,13 @@ Pure function — no DB or network calls. Sorts tasks by equipment priority
"""
from __future__ import annotations
from dataclasses import dataclass, field
from dataclasses import dataclass
_EQUIPMENT_PRIORITY = {"oven": 0, "stovetop": 1, "cold": 2, "no-heat": 3}
_DEFAULT_PRIORITY = 4
@dataclass
@dataclass(frozen=True)
class PrepTask:
recipe_id: int | None
slot_id: int | None
@ -57,7 +57,7 @@ def build_prep_tasks(slots: list[dict], recipes: list[dict]) -> list[PrepTask]:
return []
recipe_map: dict[int, dict] = {r["id"]: r for r in recipes}
raw_tasks: list[tuple[int, PrepTask]] = [] # (priority, task)
raw_tasks: list[tuple[int, dict]] = [] # (priority, kwargs)
for slot in slots:
recipe_id = slot.get("recipe_id")
@ -69,18 +69,23 @@ def build_prep_tasks(slots: list[dict], recipes: list[dict]) -> list[PrepTask]:
eq = _equipment(recipe)
priority = _EQUIPMENT_PRIORITY.get(eq or "", _DEFAULT_PRIORITY)
task = PrepTask(
recipe_id=recipe_id,
slot_id=slot.get("id"),
task_label=recipe.get("name", f"Recipe {recipe_id}"),
duration_minutes=_total_minutes(recipe),
sequence_order=0, # filled below
equipment=eq,
)
raw_tasks.append((priority, task))
raw_tasks.append((priority, {
"recipe_id": recipe_id,
"slot_id": slot.get("id"),
"task_label": recipe.get("name", f"Recipe {recipe_id}"),
"duration_minutes": _total_minutes(recipe),
"equipment": eq,
}))
raw_tasks.sort(key=lambda t: t[0])
for i, (_, task) in enumerate(raw_tasks, 1):
task.sequence_order = i
return [t for _, t in raw_tasks]
return [
PrepTask(
recipe_id=kw["recipe_id"],
slot_id=kw["slot_id"],
task_label=kw["task_label"],
duration_minutes=kw["duration_minutes"],
sequence_order=i,
equipment=kw["equipment"],
)
for i, (_, kw) in enumerate(raw_tasks, 1)
]

View file

@ -69,20 +69,20 @@ def paid_session():
def test_create_plan_free_tier_locks_to_dinner(free_session):
resp = client.post("/api/v1/meal-plans/", json={
"week_start": "2026-04-14", "meal_types": ["breakfast", "dinner"]
"week_start": "2026-04-13", "meal_types": ["breakfast", "dinner"]
})
assert resp.status_code == 200
# Free tier forced to dinner-only regardless of request
free_session.create_meal_plan.assert_called_once_with("2026-04-14", ["dinner"])
free_session.create_meal_plan.assert_called_once_with("2026-04-13", ["dinner"])
def test_create_plan_paid_tier_respects_meal_types(paid_session):
resp = client.post("/api/v1/meal-plans/", json={
"week_start": "2026-04-14", "meal_types": ["breakfast", "lunch", "dinner"]
"week_start": "2026-04-13", "meal_types": ["breakfast", "lunch", "dinner"]
})
assert resp.status_code == 200
paid_session.create_meal_plan.assert_called_once_with(
"2026-04-14", ["breakfast", "lunch", "dinner"]
"2026-04-13", ["breakfast", "lunch", "dinner"]
)