Public Access
370 lines
12 KiB
Python
370 lines
12 KiB
Python
"""Tests for app.services.recipe_discovery and app.services.recipe_ingestion.
|
|
|
|
These are unit tests using mocked HTTP responses; no network calls.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import uuid
|
|
from datetime import date
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
pytestmark = pytest.mark.requires_postgres
|
|
|
|
|
|
class MockResponse:
|
|
"""Minimal stand-in for requests.Response."""
|
|
|
|
def __init__(self, json_data: dict | None = None, status_code: int = 200):
|
|
self._json = json_data or {}
|
|
self.status_code = status_code
|
|
|
|
def raise_for_status(self):
|
|
if self.status_code >= 400:
|
|
raise Exception(f"HTTP {self.status_code}")
|
|
|
|
def json(self):
|
|
return self._json
|
|
|
|
|
|
class TestRecipeDiscoveryService:
|
|
@patch("app.services.recipe_discovery.requests.get")
|
|
def test_disabled_without_api_key(self, mock_get):
|
|
from app.services.recipe_discovery import RecipeDiscoveryService
|
|
|
|
svc = RecipeDiscoveryService()
|
|
svc.enabled = False
|
|
recipes = svc.discover(["chicken"])
|
|
assert recipes == []
|
|
mock_get.assert_not_called()
|
|
|
|
@patch("app.services.recipe_discovery.requests.get")
|
|
def test_quota_stop(self, mock_get):
|
|
from app.services.recipe_discovery import RecipeDiscoveryService
|
|
|
|
svc = RecipeDiscoveryService()
|
|
svc.enabled = True
|
|
# Simulate quota already consumed
|
|
svc._points_used = 140
|
|
|
|
recipes = svc.discover(["query1", "query2"])
|
|
assert recipes == []
|
|
mock_get.assert_not_called()
|
|
|
|
@patch("app.services.recipe_discovery.requests.get")
|
|
def test_single_result_normalization(self, mock_get):
|
|
from app.services.recipe_discovery import RecipeDiscoveryService, ExternalRecipe
|
|
|
|
svc = RecipeDiscoveryService()
|
|
svc.enabled = True
|
|
|
|
search_resp = {
|
|
"results": [
|
|
{
|
|
"id": 123,
|
|
"title": "Spicy Thai Basil Chicken",
|
|
"image": "http://img/1.jpg",
|
|
"cuisines": ["Thai"],
|
|
"diets": ["gluten free"],
|
|
"servings": 4,
|
|
}
|
|
],
|
|
"totalResults": 1,
|
|
}
|
|
|
|
info_resp = {
|
|
"id": 123,
|
|
"title": "Spicy Thai Basil Chicken",
|
|
"extendedIngredients": [
|
|
{"amount": 1.5, "unit": "lb", "name": "chicken breast"},
|
|
{"amount": 2, "unit": "tbsp", "name": "basil"},
|
|
],
|
|
"analyzedInstructions": [{"steps": [{"step": "Cook chicken"}, {"step": "Add basil"}]}],
|
|
"preparationMinutes": 10,
|
|
"cookingMinutes": 20,
|
|
"readyInMinutes": 30,
|
|
"servings": 4,
|
|
"sourceUrl": "http://example.com/recipe",
|
|
"image": "http://img/1.jpg",
|
|
}
|
|
|
|
def _make_resp(*a, **kw):
|
|
url = a[0] if a else ""
|
|
if "complexSearch" in url:
|
|
return MockResponse(search_resp)
|
|
if "information" in url:
|
|
return MockResponse(info_resp)
|
|
return MockResponse({})
|
|
|
|
mock_get.side_effect = _make_resp
|
|
|
|
recipes = svc.discover(["thai chicken"])
|
|
assert len(recipes) == 1
|
|
r = recipes[0]
|
|
assert isinstance(r, ExternalRecipe)
|
|
assert r.name == "Spicy Thai Basil Chicken"
|
|
assert r.external_source == "spoonacular"
|
|
assert r.external_id == "123"
|
|
assert r.cuisine_tags == ["thai"]
|
|
assert r.dietary_tags == ["gluten free"]
|
|
assert r.servings == 4
|
|
assert len(r.ingredients) == 2
|
|
assert r.ingredients[0] == {"name": "chicken breast", "qty": 1.5, "unit": "lb"}
|
|
assert r.instructions == ["Cook chicken", "Add basil"]
|
|
assert r.prep_time_minutes == 10
|
|
assert r.cook_time_minutes == 20
|
|
assert r.source_url == "http://example.com/recipe"
|
|
|
|
@patch("app.services.recipe_discovery.requests.get")
|
|
def test_deduplication_across_queries(self, mock_get):
|
|
from app.services.recipe_discovery import RecipeDiscoveryService
|
|
|
|
svc = RecipeDiscoveryService()
|
|
svc.enabled = True
|
|
|
|
resp = {
|
|
"results": [
|
|
{"id": 1, "title": "A", "image": "", "cuisines": [], "diets": [], "servings": 2}
|
|
],
|
|
"totalResults": 1,
|
|
}
|
|
|
|
def _make_resp(*a, **kw):
|
|
return MockResponse(resp)
|
|
|
|
mock_get.side_effect = _make_resp
|
|
|
|
recipes = svc.discover(["q1", "q2"])
|
|
# Same recipe ID should only appear once even though both queries returned it
|
|
assert len(recipes) == 1
|
|
|
|
@patch("app.services.recipe_discovery.requests.get")
|
|
def test_search_failure_graceful(self, mock_get):
|
|
from app.services.recipe_discovery import RecipeDiscoveryService
|
|
import requests
|
|
|
|
svc = RecipeDiscoveryService()
|
|
svc.enabled = True
|
|
mock_get.side_effect = requests.ConnectionError("network error")
|
|
|
|
recipes = svc.discover(["chicken"])
|
|
assert recipes == []
|
|
|
|
|
|
class TestRecipeIngestionService:
|
|
def test_ingest_skips_duplicate_external(self, db):
|
|
from app.services.recipe_ingestion import RecipeIngestionService
|
|
from app.services.recipe_discovery import ExternalRecipe
|
|
from app.models import Recipe, FamilyProfile
|
|
|
|
svc = RecipeIngestionService()
|
|
family = FamilyProfile(
|
|
id=uuid.uuid4(),
|
|
name="IngestFamily",
|
|
household_size=2,
|
|
adult_count=2,
|
|
child_count=0,
|
|
pending_approval_policy="approve",
|
|
)
|
|
db.add(family)
|
|
db.flush()
|
|
|
|
# Seed existing row with same external_source+external_id
|
|
existing = Recipe(
|
|
id=uuid.uuid4(),
|
|
family_profile_id=family.id,
|
|
name="Already There",
|
|
servings=4,
|
|
ingredients=[],
|
|
instructions=["cook"],
|
|
external_source="spoonacular",
|
|
external_id="99",
|
|
)
|
|
db.add(existing)
|
|
db.flush()
|
|
|
|
ext = ExternalRecipe(
|
|
name="Already There",
|
|
external_source="spoonacular",
|
|
external_id="99",
|
|
image_url=None,
|
|
description=None,
|
|
prep_time_minutes=None,
|
|
cook_time_minutes=None,
|
|
servings=2,
|
|
cuisine_tags=[],
|
|
dietary_tags=[],
|
|
protein_type=None,
|
|
calories_per_serving=None,
|
|
ingredients=[],
|
|
instructions=["cook"],
|
|
source_url=None,
|
|
)
|
|
added = svc.ingest(db, family.id, [ext], {"discovery_queries": [], "positive_signals": []})
|
|
assert added == 0
|
|
|
|
def test_ingest_fuzzy_duplicate_skips(self, db):
|
|
from app.services.recipe_ingestion import RecipeIngestionService
|
|
from app.services.recipe_discovery import ExternalRecipe
|
|
from app.models import Recipe, FamilyProfile
|
|
|
|
svc = RecipeIngestionService()
|
|
family = FamilyProfile(
|
|
id=uuid.uuid4(),
|
|
name="FuzzyFamily",
|
|
household_size=2,
|
|
adult_count=2,
|
|
child_count=0,
|
|
pending_approval_policy="approve",
|
|
)
|
|
db.add(family)
|
|
db.flush()
|
|
|
|
existing = Recipe(
|
|
id=uuid.uuid4(),
|
|
family_profile_id=family.id,
|
|
name="Grilled Salmon with Lemon Butter Sauce",
|
|
servings=4,
|
|
ingredients=[],
|
|
instructions=["cook"],
|
|
)
|
|
db.add(existing)
|
|
db.flush()
|
|
|
|
ext = ExternalRecipe(
|
|
name="Grilled Salmon with Lemon Butter Sauce and Herbs",
|
|
external_source="spoonacular",
|
|
external_id="42",
|
|
image_url=None,
|
|
description=None,
|
|
prep_time_minutes=None,
|
|
cook_time_minutes=None,
|
|
servings=2,
|
|
cuisine_tags=[],
|
|
dietary_tags=[],
|
|
protein_type=None,
|
|
calories_per_serving=None,
|
|
ingredients=[],
|
|
instructions=["cook"],
|
|
source_url=None,
|
|
)
|
|
added = svc.ingest(db, family.id, [ext], {"discovery_queries": [], "positive_signals": []})
|
|
assert added == 0
|
|
|
|
def test_ingest_creates_recipe_and_ingredient(self, db):
|
|
from app.services.recipe_ingestion import RecipeIngestionService
|
|
from app.services.recipe_discovery import ExternalRecipe
|
|
from app.models import Recipe, FamilyProfile, Ingredient
|
|
|
|
svc = RecipeIngestionService()
|
|
family = FamilyProfile(
|
|
id=uuid.uuid4(),
|
|
name="CreateFamily",
|
|
household_size=2,
|
|
adult_count=2,
|
|
child_count=0,
|
|
pending_approval_policy="approve",
|
|
)
|
|
db.add(family)
|
|
db.flush()
|
|
|
|
ext = ExternalRecipe(
|
|
name="Lemon Herb Salmon",
|
|
external_source="spoonacular",
|
|
external_id="77",
|
|
image_url="http://img/salmon.jpg",
|
|
description="<b>Delicious</b> salmon recipe",
|
|
prep_time_minutes=10,
|
|
cook_time_minutes=15,
|
|
servings=2,
|
|
cuisine_tags=["mediterranean"],
|
|
dietary_tags=["pescatarian"],
|
|
protein_type="fish",
|
|
calories_per_serving=350,
|
|
ingredients=[
|
|
{"name": "salmon fillet", "qty": 2.0, "unit": "lb"},
|
|
],
|
|
instructions=["Preheat oven", "Bake till flaky"],
|
|
source_url="http://example.com/77",
|
|
)
|
|
analysis = {
|
|
"discovery_queries": ["mediterranean fish"],
|
|
"positive_signals": [{"type": "prefer_cuisine", "value": "mediterranean"}],
|
|
}
|
|
|
|
added = svc.ingest(db, family.id, [ext], analysis)
|
|
assert added == 1
|
|
|
|
recipes = db.query(Recipe).filter(Recipe.external_id == "77").all()
|
|
assert len(recipes) == 1
|
|
r = recipes[0]
|
|
assert r.name == "Lemon Herb Salmon"
|
|
assert r.external_source == "spoonacular"
|
|
assert r.external_id == "77"
|
|
assert r.protein_type == "fish"
|
|
assert r.is_manually_added is False
|
|
assert r.discovery_reason is not None
|
|
assert "mediterranean fish" in r.discovery_reason
|
|
assert r.cuisine_tags == ["mediterranean"]
|
|
assert r.dietary_tags == ["pescatarian"]
|
|
assert r.calories_per_serving == 350
|
|
assert r.ingredients[0]["name"] == "Salmon Fillet"
|
|
assert r.instructions == ["Preheat oven", "Bake till flaky"]
|
|
|
|
# Ingredient was created
|
|
ings = db.query(Ingredient).filter(Ingredient.name_lower == "salmon fillet").all()
|
|
assert len(ings) == 1
|
|
assert ings[0].name == "Salmon Fillet"
|
|
|
|
def test_ingest_maps_to_existing_ingredient(self, db):
|
|
from app.services.recipe_ingestion import RecipeIngestionService
|
|
from app.services.recipe_discovery import ExternalRecipe
|
|
from app.models import Recipe, FamilyProfile, Ingredient
|
|
|
|
svc = RecipeIngestionService()
|
|
family = FamilyProfile(
|
|
id=uuid.uuid4(),
|
|
name="MapFamily",
|
|
household_size=2,
|
|
adult_count=2,
|
|
child_count=0,
|
|
pending_approval_policy="approve",
|
|
)
|
|
db.add(family)
|
|
|
|
existing_ing = Ingredient(
|
|
id=uuid.uuid4(),
|
|
name="Firm Tofu",
|
|
name_lower="firm tofu",
|
|
aliases=["tofu"],
|
|
)
|
|
db.add(existing_ing)
|
|
db.flush()
|
|
|
|
ext = ExternalRecipe(
|
|
name="Mapo Tofu",
|
|
external_source="spoonacular",
|
|
external_id="88",
|
|
image_url=None,
|
|
description=None,
|
|
prep_time_minutes=None,
|
|
cook_time_minutes=None,
|
|
servings=2,
|
|
cuisine_tags=[],
|
|
dietary_tags=[],
|
|
protein_type=None,
|
|
calories_per_serving=None,
|
|
ingredients=[
|
|
{"name": "Firm Tofu", "qty": 1, "unit": "lb"},
|
|
],
|
|
instructions=["fry"],
|
|
source_url=None,
|
|
)
|
|
added = svc.ingest(db, family.id, [ext], {"discovery_queries": [], "positive_signals": []})
|
|
assert added == 1
|
|
|
|
recipes = db.query(Recipe).filter(Recipe.external_id == "88").all()
|
|
assert recipes[0].ingredients[0]["ingredient_id"] == str(existing_ing.id)
|
|
assert recipes[0].ingredients[0]["name"] == "Firm Tofu"
|