Kyle DammannandClaude Opus 5.5
1bf8ef8b65
- recipes/ai.py: Chat Completions with strict JSON schema (Structured Outputs), Pydantic re-validation, plausibility clamping, prompt treats input as data only - "Mit KI aufbereiten" on create/edit form returns an editable draft - Import page: photo/screenshot (vision, downscaled JPEG, not stored) and pasted text; drafts pre-fill the recipe form with a preview - Clear UI messages for timeout, rate limit, quota, auth, model and connection errors; without OPENAI_API_KEY AI features show a notice - Optional per-user and global daily AI limits (default unlimited) - Tests for response parsing, schema strictness, error mapping, views Co-Authored-By: Claude Opus 5.5 <[email protected]>
306 lines
12 KiB
Python
306 lines
12 KiB
Python
import json
|
|
|
|
import httpx2
|
|
import openai
|
|
import pytest
|
|
from django.urls import reverse
|
|
|
|
from recipes import ai
|
|
from recipes.models import Recipe
|
|
from recipes.parsing import parse_ingredients
|
|
|
|
from .recipe_helpers import form_data, image_upload
|
|
|
|
VALID = {
|
|
"is_recipe": True,
|
|
"title": "Linsen-Dal",
|
|
"description": "Cremig und würzig.",
|
|
"servings": 4,
|
|
"servings_label": "Portionen",
|
|
"prep_minutes": 10,
|
|
"cook_minutes": 25,
|
|
"ingredients": [
|
|
{"group": "", "amount": 200, "amount_max": None, "unit": "g", "name": "rote Linsen", "note": "gewaschen"},
|
|
{"group": "", "amount": 0.5, "amount_max": None, "unit": "TL", "name": "Kurkuma", "note": ""},
|
|
{"group": "Zum Servieren", "amount": 2, "amount_max": 3, "unit": "EL", "name": "Joghurt", "note": ""},
|
|
{"group": "Zum Servieren", "amount": None, "amount_max": None, "unit": "", "name": "Koriander", "note": ""},
|
|
],
|
|
"steps": ["Zwiebeln anbraten.", " ", "Linsen zugeben und 20 Minuten köcheln."],
|
|
"tags": ["Indisch", "#Vegetarisch", "indisch"],
|
|
"author_notes": "",
|
|
}
|
|
|
|
|
|
def ai_answer(**changes):
|
|
return json.dumps({**VALID, **changes})
|
|
|
|
|
|
# --- Parsing & validation of AI answers ----------------------------------------------
|
|
|
|
|
|
def test_parse_valid_response():
|
|
recipe = ai.parse_response(ai_answer())
|
|
assert recipe.title == "Linsen-Dal"
|
|
assert recipe.steps == ["Zwiebeln anbraten.", "Linsen zugeben und 20 Minuten köcheln."]
|
|
assert recipe.tags == ["indisch", "vegetarisch"]
|
|
|
|
|
|
@pytest.mark.parametrize("content", ["", "kein json", "[]", '{"title": "x"}', json.dumps({**VALID, "ingredients": "200 g"})])
|
|
def test_invalid_response_rejected(content):
|
|
with pytest.raises(ai.AIError, match="ungültige Antwort"):
|
|
ai.parse_response(content)
|
|
|
|
|
|
def test_not_a_recipe():
|
|
with pytest.raises(ai.AIError, match="kein Rezept"):
|
|
ai.parse_response(ai_answer(is_recipe=False))
|
|
with pytest.raises(ai.AIError, match="kein Rezept"):
|
|
ai.parse_response(ai_answer(ingredients=[], steps=[]))
|
|
|
|
|
|
def test_implausible_values_are_dropped():
|
|
recipe = ai.parse_response(ai_answer(
|
|
servings=0, prep_minutes=-5, cook_minutes=99999,
|
|
ingredients=[{"group": "", "amount": -3, "amount_max": None, "unit": "g", "name": "Mehl", "note": ""}],
|
|
))
|
|
assert recipe.servings is None and recipe.prep_minutes is None and recipe.cook_minutes is None
|
|
assert recipe.ingredients[0].amount is None
|
|
|
|
|
|
def test_form_initial_roundtrips_into_data_model():
|
|
initial = ai.parse_response(ai_answer()).to_form_initial()
|
|
items = parse_ingredients(initial["ingredients_text"])
|
|
assert [(i.group, str(i.amount) if i.amount is not None else None, i.unit, i.name, i.note) for i in items] == [
|
|
("", "200", "g", "rote Linsen", "gewaschen"),
|
|
("", "0.5", "TL", "Kurkuma", ""),
|
|
("Zum Servieren", "2", "EL", "Joghurt", ""),
|
|
("Zum Servieren", None, "", "Koriander", ""),
|
|
]
|
|
assert items[2].amount_max == 3
|
|
assert initial["servings"] == 4 and initial["tags_text"] == "indisch, vegetarisch"
|
|
|
|
|
|
def test_schema_is_strict_structured_output():
|
|
def check(node):
|
|
if node.get("type") == "object":
|
|
assert node["additionalProperties"] is False
|
|
assert set(node["required"]) == set(node["properties"])
|
|
for child in node["properties"].values():
|
|
check(child)
|
|
if node.get("type") == "array":
|
|
check(node["items"])
|
|
|
|
check(ai.RECIPE_JSON_SCHEMA)
|
|
|
|
|
|
# --- Error handling ----------------------------------------------------------------------
|
|
|
|
REQUEST = httpx2.Request("POST", "https://api.openai.com/v1/chat/completions")
|
|
|
|
|
|
def api_error(cls, status, message="error", body=None):
|
|
return cls(message, response=httpx2.Response(status, request=REQUEST), body=body)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"error, text",
|
|
[
|
|
(openai.APITimeoutError(request=REQUEST), "zu lange"),
|
|
(api_error(openai.RateLimitError, 429, "Rate limit reached"), "ausgelastet"),
|
|
(api_error(openai.RateLimitError, 429, "You exceeded your current quota", {"code": "insufficient_quota"}), "Guthaben"),
|
|
(api_error(openai.AuthenticationError, 401), "API-Key"),
|
|
(api_error(openai.NotFoundError, 404), "OPENAI_MODEL"),
|
|
(api_error(openai.BadRequestError, 400), "nicht verarbeiten"),
|
|
(openai.APIConnectionError(request=REQUEST), "nicht erreichbar"),
|
|
(api_error(openai.InternalServerError, 500), "Fehler aufgetreten"),
|
|
],
|
|
)
|
|
def test_api_errors_are_mapped(error, text):
|
|
assert text in ai.map_api_error(error).user_message
|
|
|
|
|
|
@pytest.fixture
|
|
def ai_on(settings):
|
|
settings.OPENAI_API_KEY = "sk-test-not-real"
|
|
|
|
|
|
@pytest.fixture
|
|
def fake_openai(monkeypatch, ai_on):
|
|
calls = []
|
|
|
|
def fake(messages):
|
|
calls.append(messages)
|
|
return ai_answer()
|
|
|
|
monkeypatch.setattr(ai, "_create_completion", fake)
|
|
return calls
|
|
|
|
|
|
@pytest.mark.django_db
|
|
def test_api_error_shown_in_ui(monkeypatch, ai_on, user_client):
|
|
def boom(messages):
|
|
raise openai.APITimeoutError(request=REQUEST)
|
|
|
|
monkeypatch.setattr(ai, "_create_completion", boom)
|
|
response = user_client.post(reverse("recipes:create"), form_data(ai="1"), follow=True)
|
|
assert "zu lange gebraucht" in response.content.decode()
|
|
assert not Recipe.objects.exists()
|
|
|
|
|
|
# --- Without API key -------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.django_db
|
|
def test_without_key_ai_is_disabled_but_app_works(settings, user_client):
|
|
settings.OPENAI_API_KEY = ""
|
|
form_page = user_client.get(reverse("recipes:create")).content.decode()
|
|
assert "Mit KI aufbereiten" not in form_page
|
|
assert "nicht eingerichtet" in user_client.get(reverse("recipes:import")).content.decode()
|
|
response = user_client.post(reverse("recipes:create"), form_data(ai="1"))
|
|
assert "nicht eingerichtet" in response.content.decode()
|
|
assert not Recipe.objects.exists()
|
|
# Saving manually still works
|
|
assert user_client.post(reverse("recipes:create"), form_data()).status_code == 302
|
|
|
|
|
|
@pytest.mark.django_db
|
|
def test_without_key_imports_give_clear_message(settings, user_client):
|
|
settings.OPENAI_API_KEY = ""
|
|
response = user_client.post(reverse("recipes:import_text"), {"text": "200 g Mehl, 2 Eier, alles verrühren und backen"})
|
|
assert "nicht eingerichtet" in response.content.decode()
|
|
|
|
|
|
# --- Views with a (fake) model ------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.django_db
|
|
def test_ai_cleanup_returns_draft_without_saving(fake_openai, user_client):
|
|
response = user_client.post(reverse("recipes:create"), form_data(ai="1", visibility="public"))
|
|
assert response.status_code == 200
|
|
assert not Recipe.objects.exists()
|
|
form = response.context["form"]
|
|
assert form.initial["title"] == "Linsen-Dal" and form.initial["visibility"] == "public"
|
|
assert "rote Linsen" in form.initial["ingredients_text"]
|
|
assert "bitte prüfen" in response.content.decode()
|
|
# user text went to the model as data
|
|
assert "Pfannkuchen" in fake_openai[0][1]["content"]
|
|
|
|
|
|
@pytest.mark.django_db
|
|
def test_ai_cleanup_on_edit_keeps_recipe_until_saved(fake_openai, user_client):
|
|
user_client.post(reverse("recipes:create"), form_data())
|
|
recipe = Recipe.objects.get()
|
|
response = user_client.post(reverse("recipes:edit", args=[recipe.pk]), form_data(ai="1"))
|
|
assert response.context["form"].initial["title"] == "Linsen-Dal"
|
|
recipe.refresh_from_db()
|
|
assert recipe.title == "Pfannkuchen"
|
|
|
|
|
|
@pytest.mark.django_db
|
|
def test_text_import_creates_draft(fake_openai, user_client):
|
|
response = user_client.post(reverse("recipes:import_text"), {
|
|
"text": "Dal: 200g rote Linsen, Kurkuma, Joghurt. Linsen kochen.", "source_url": "https://www.instagram.com/p/abc/",
|
|
})
|
|
assert response.status_code == 200
|
|
assert response.context["form"].initial["source_url"] == "https://www.instagram.com/p/abc/"
|
|
assert response.context["form_action"] == reverse("recipes:create")
|
|
assert not Recipe.objects.exists()
|
|
|
|
|
|
@pytest.mark.django_db
|
|
def test_photo_import_sends_downscaled_jpeg(fake_openai, user_client):
|
|
response = user_client.post(reverse("recipes:import_photo"), {"photos": [image_upload(size=(4000, 3000))]})
|
|
assert response.status_code == 200
|
|
parts = fake_openai[0][1]["content"]
|
|
image_part = next(p for p in parts if p["type"] == "image_url")
|
|
assert image_part["image_url"]["url"].startswith("data:image/jpeg;base64,")
|
|
assert response.context["form"].initial["title"] == "Linsen-Dal"
|
|
assert not Recipe.objects.exists()
|
|
|
|
|
|
@pytest.mark.django_db
|
|
def test_photo_import_rejects_non_image(fake_openai, user_client):
|
|
from django.core.files.uploadedfile import SimpleUploadedFile
|
|
|
|
bad = SimpleUploadedFile("x.jpg", b"nope", content_type="image/jpeg")
|
|
response = user_client.post(reverse("recipes:import_photo"), {"photos": [bad]})
|
|
assert "kein unterstütztes Bild" in response.content.decode()
|
|
assert not fake_openai
|
|
|
|
|
|
@pytest.mark.django_db
|
|
def test_ai_rate_limit_per_user(fake_openai, user_client, settings):
|
|
settings.AI_RATE_LIMIT_PER_USER_PER_DAY = 2
|
|
for _ in range(3):
|
|
response = user_client.post(reverse("recipes:create"), form_data(ai="1"))
|
|
assert len(fake_openai) == 2
|
|
assert "Tageslimit" in response.content.decode()
|
|
|
|
|
|
@pytest.mark.django_db
|
|
def test_ai_global_limit(fake_openai, client, user, other_user, settings):
|
|
settings.AI_RATE_LIMIT_GLOBAL_PER_DAY = 1
|
|
for u in (user, other_user):
|
|
client.force_login(u)
|
|
response = client.post(reverse("recipes:create"), form_data(ai="1"))
|
|
assert len(fake_openai) == 1
|
|
assert "Kontingent" in response.content.decode()
|
|
|
|
|
|
@pytest.mark.django_db
|
|
def test_ai_unlimited_by_default(fake_openai, user_client):
|
|
for _ in range(5):
|
|
user_client.post(reverse("recipes:create"), form_data(ai="1"))
|
|
assert len(fake_openai) == 5
|
|
|
|
|
|
def test_text_too_short():
|
|
with pytest.raises(ai.AIError, match="mehr Text"):
|
|
ai.structure_text(None, "Mehl")
|
|
|
|
|
|
# --- The real request builder against a stub client --------------------------------------
|
|
|
|
|
|
class StubClient:
|
|
def __init__(self, content=None, refusal=None, finish_reason="stop"):
|
|
from types import SimpleNamespace
|
|
|
|
self.kwargs = None
|
|
message = SimpleNamespace(content=content, refusal=refusal)
|
|
self._response = SimpleNamespace(choices=[SimpleNamespace(message=message, finish_reason=finish_reason)])
|
|
self.chat = SimpleNamespace(completions=SimpleNamespace(create=self._create))
|
|
|
|
def _create(self, **kwargs):
|
|
self.kwargs = kwargs
|
|
return self._response
|
|
|
|
|
|
def test_request_uses_strict_json_schema(monkeypatch, settings):
|
|
settings.OPENAI_MODEL = "test-model"
|
|
settings.OPENAI_REASONING_EFFORT = "low"
|
|
stub = StubClient(content=ai_answer())
|
|
monkeypatch.setattr(ai, "_client", lambda: stub)
|
|
assert ai._create_completion([{"role": "user", "content": "x"}]) == ai_answer()
|
|
assert stub.kwargs["model"] == "test-model"
|
|
assert stub.kwargs["reasoning_effort"] == "low"
|
|
fmt = stub.kwargs["response_format"]
|
|
assert fmt["type"] == "json_schema" and fmt["json_schema"]["strict"] is True
|
|
|
|
|
|
def test_reasoning_effort_omitted_when_empty(monkeypatch, settings):
|
|
settings.OPENAI_REASONING_EFFORT = ""
|
|
stub = StubClient(content=ai_answer())
|
|
monkeypatch.setattr(ai, "_client", lambda: stub)
|
|
ai._create_completion([])
|
|
assert "reasoning_effort" not in stub.kwargs
|
|
|
|
|
|
def test_refusal_and_truncation(monkeypatch):
|
|
monkeypatch.setattr(ai, "_client", lambda: StubClient(refusal="nein"))
|
|
with pytest.raises(ai.AIError, match="abgelehnt"):
|
|
ai._create_completion([])
|
|
monkeypatch.setattr(ai, "_client", lambda: StubClient(content='{"is_re', finish_reason="length"))
|
|
with pytest.raises(ai.AIError, match="zu lang"):
|
|
ai._create_completion([])
|