Public Access
247 lines
7.9 KiB
Python
247 lines
7.9 KiB
Python
"""Tests for app.services.feedback_analyzer."""
|
|
from datetime import date, timedelta
|
|
from uuid import uuid4
|
|
|
|
import pytest
|
|
|
|
from app.models import (
|
|
DenialReason,
|
|
FamilyProfile,
|
|
FamilyMember,
|
|
FamilyMemberRole,
|
|
Feedback,
|
|
MealPlan,
|
|
MealPlanItem,
|
|
MealPlanItemStatus,
|
|
MealType,
|
|
NeverSuggest,
|
|
Recipe,
|
|
WeeklyRun,
|
|
)
|
|
from app.services.feedback_analyzer import FeedbackAnalyzer
|
|
|
|
|
|
@pytest.fixture()
|
|
def make_family(db):
|
|
def _make():
|
|
fp = FamilyProfile(
|
|
id=uuid4(),
|
|
name="TestFamily",
|
|
household_size=2,
|
|
adult_count=2,
|
|
child_count=0,
|
|
pending_approval_policy="approve",
|
|
)
|
|
db.add(fp)
|
|
db.flush()
|
|
return fp
|
|
return _make
|
|
|
|
|
|
@pytest.fixture()
|
|
def make_member(db, make_family):
|
|
def _make(family=None):
|
|
fp = family or make_family()
|
|
m = FamilyMember(
|
|
id=uuid4(),
|
|
family_profile_id=fp.id,
|
|
name="Alice",
|
|
role=FamilyMemberRole.ADULT,
|
|
)
|
|
db.add(m)
|
|
db.flush()
|
|
return m
|
|
return _make
|
|
|
|
|
|
@pytest.fixture()
|
|
def make_recipe(db):
|
|
def _make(**kw):
|
|
r = Recipe(
|
|
id=uuid4(),
|
|
name=kw.get("name", "Test Recipe"),
|
|
servings=kw.get("servings", 4),
|
|
ingredients=[{"name": "ing", "qty": 1, "unit": "cup"}],
|
|
instructions=["cook"],
|
|
cuisine_tags=kw.get("cuisine_tags", []),
|
|
protein_type=kw.get("protein_type", None),
|
|
)
|
|
db.add(r)
|
|
db.flush()
|
|
return r
|
|
return _make
|
|
|
|
|
|
@pytest.fixture()
|
|
def make_meal_plan(db, make_family):
|
|
def _make(family=None, week_start=None):
|
|
week = week_start or date.today()
|
|
mp = MealPlan(
|
|
id=uuid4(),
|
|
family_profile_id=(family or make_family()).id,
|
|
week_start_date=week,
|
|
)
|
|
db.add(mp)
|
|
db.flush()
|
|
return mp
|
|
return _make
|
|
|
|
|
|
@pytest.fixture()
|
|
def make_item(db, make_meal_plan, make_recipe):
|
|
def _make(meal_plan=None, recipe=None, status=MealPlanItemStatus.pending, day_of_week=1):
|
|
mp = meal_plan or make_meal_plan()
|
|
r = recipe or make_recipe()
|
|
item = MealPlanItem(
|
|
id=uuid4(),
|
|
meal_plan_id=mp.id,
|
|
recipe_id=r.id,
|
|
day_of_week=day_of_week,
|
|
meal_type=MealType.DINNER,
|
|
approval_status=status,
|
|
)
|
|
db.add(item)
|
|
db.flush()
|
|
return item
|
|
return _make
|
|
|
|
|
|
class TestFeedbackAnalyzer:
|
|
def test_insufficient_feedback(self, db, make_family):
|
|
family = make_family()
|
|
analyzer = FeedbackAnalyzer(lookback_weeks=4)
|
|
today = date.today()
|
|
result = analyzer.analyze(db, family.id, today=today)
|
|
assert result.total_feedback_count == 0
|
|
assert result.confidence == 0.0
|
|
assert result.discovery_queries == []
|
|
|
|
def test_positive_cuisine_signal(self, db, make_family, make_member, make_recipe, make_meal_plan, make_item):
|
|
family = make_family()
|
|
member = make_member(family=family)
|
|
recipe = make_recipe(cuisine_tags=["mexican"], protein_type="chicken")
|
|
mp = make_meal_plan(family=family)
|
|
item = make_item(meal_plan=mp, recipe=recipe)
|
|
|
|
# 3 feedbacks, avg rating 4 (>= threshold, >= 2 samples)
|
|
for _ in range(3):
|
|
f = Feedback(
|
|
id=uuid4(),
|
|
family_profile_id=family.id,
|
|
meal_plan_item_id=item.id,
|
|
rating=4,
|
|
)
|
|
db.add(f)
|
|
db.flush()
|
|
|
|
analyzer = FeedbackAnalyzer(lookback_weeks=4)
|
|
result = analyzer.analyze(db, family.id)
|
|
assert result.total_feedback_count == 3
|
|
assert result.confidence == 1.0
|
|
|
|
pos = result.positive_signals
|
|
assert len(pos) == 2 # cuisine + protein
|
|
assert any(s.type == "prefer_cuisine" and s.value == "mexican" for s in pos)
|
|
assert any(s.type == "prefer_protein" and s.value == "chicken" for s in pos)
|
|
|
|
def test_never_suggest_blocks_recipe(self, db, make_family, make_member, make_recipe, make_meal_plan, make_item):
|
|
family = make_family()
|
|
member = make_member(family=family)
|
|
recipe = make_recipe(cuisine_tags=["indian"], protein_type="lamb")
|
|
mp = make_meal_plan(family=family)
|
|
item = make_item(meal_plan=mp, recipe=recipe)
|
|
|
|
# Deny with never-suggest
|
|
f = Feedback(
|
|
id=uuid4(),
|
|
family_profile_id=family.id,
|
|
meal_plan_item_id=item.id,
|
|
rating=1,
|
|
never_suggest=True,
|
|
denial_reason=DenialReason.DISLIKED_INGREDIENT,
|
|
)
|
|
db.add(f)
|
|
db.flush()
|
|
|
|
ns = NeverSuggest(
|
|
id=uuid4(),
|
|
family_profile_id=family.id,
|
|
recipe_id=recipe.id,
|
|
)
|
|
db.add(ns)
|
|
db.flush()
|
|
|
|
analyzer = FeedbackAnalyzer(lookback_weeks=4)
|
|
result = analyzer.analyze(db, family.id)
|
|
pos = result.positive_signals
|
|
# 1 feedback < min, therefore no positive signals
|
|
assert len(pos) == 0
|
|
|
|
def test_denial_reason_aggregated(self, db, make_family, make_member, make_recipe, make_meal_plan, make_item):
|
|
family = make_family()
|
|
member = make_member(family=family)
|
|
recipe = make_recipe()
|
|
mp = make_meal_plan(family=family)
|
|
item = make_item(meal_plan=mp, recipe=recipe)
|
|
|
|
for _ in range(3):
|
|
f = Feedback(
|
|
id=uuid4(),
|
|
family_profile_id=family.id,
|
|
meal_plan_item_id=item.id,
|
|
rating=2,
|
|
denial_reason=DenialReason.TOO_EXPENSIVE,
|
|
)
|
|
db.add(f)
|
|
db.flush()
|
|
|
|
analyzer = FeedbackAnalyzer(lookback_weeks=4)
|
|
result = analyzer.analyze(db, family.id)
|
|
negatives = result.negative_signals
|
|
assert any(n.type == "denial_too_expensive" for n in negatives)
|
|
assert negatives[0].count == 3
|
|
|
|
def test_discovery_queries_capped(self, db, make_family, make_member, make_recipe, make_meal_plan, make_item):
|
|
family = make_family()
|
|
member = make_member(family=family)
|
|
cuisines = ["mexican", "italian", "chinese"]
|
|
proteins = ["chicken", "beef", "shrimp"]
|
|
for i, (c, p) in enumerate(zip(cuisines, proteins)):
|
|
recipe = make_recipe(name=f"R{i}", cuisine_tags=[c], protein_type=p)
|
|
mp = make_meal_plan(family=family, week_start=date.today() - timedelta(weeks=i))
|
|
item = make_item(meal_plan=mp, recipe=recipe)
|
|
for _ in range(2):
|
|
db.add(Feedback(
|
|
id=uuid4(),
|
|
family_profile_id=family.id,
|
|
meal_plan_item_id=item.id,
|
|
rating=5,
|
|
))
|
|
db.flush()
|
|
|
|
analyzer = FeedbackAnalyzer(lookback_weeks=4)
|
|
result = analyzer.analyze(db, family.id)
|
|
queries = result.discovery_queries
|
|
assert len(queries) <= 5
|
|
# At least one cuisine+protein cross query
|
|
assert any(" " in q for q in queries)
|
|
|
|
def test_top_rated_sorted(self, db, make_family, make_member, make_recipe, make_meal_plan, make_item):
|
|
family = make_family()
|
|
member = make_member(family=family)
|
|
r1 = make_recipe(name="Awesome Dish")
|
|
r2 = make_recipe(name="Meh Dish")
|
|
mp = make_meal_plan(family=family)
|
|
item1 = make_item(meal_plan=mp, recipe=r1)
|
|
item2 = make_item(meal_plan=mp, recipe=r2, day_of_week=2)
|
|
|
|
db.add(Feedback(id=uuid4(), family_profile_id=family.id, meal_plan_item_id=item1.id, rating=5))
|
|
db.add(Feedback(id=uuid4(), family_profile_id=family.id, meal_plan_item_id=item1.id, rating=5))
|
|
db.add(Feedback(id=uuid4(), family_profile_id=family.id, meal_plan_item_id=item2.id, rating=3))
|
|
db.flush()
|
|
|
|
analyzer = FeedbackAnalyzer(lookback_weeks=4)
|
|
result = analyzer.analyze(db, family.id)
|
|
assert result.top_rated_recipe_names[0] == "Awesome Dish"
|
|
|