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_text_import_becomes_description(settings, user_client): settings.OPENAI_API_KEY = "" text = "200 g Mehl, 2 Eier, alles verrühren und backen" response = user_client.post(reverse("recipes:import_text"), {"text": text}) assert "nicht eingerichtet" in response.content.decode() assert response.context["form"].initial["description"] == text assert not Recipe.objects.exists() @pytest.mark.django_db def test_without_key_photo_import_gives_clear_message(settings, user_client): settings.OPENAI_API_KEY = "" response = user_client.post(reverse("recipes:import_photo"), {"photos": [image_upload()]}, follow=True) 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([]) # --- End to end through the real OpenAI SDK against a local stub server -------------------- @pytest.fixture def stub_openai_server(settings): """A local HTTP server that speaks enough of the Chat Completions API for one request.""" import threading from http.server import BaseHTTPRequestHandler, HTTPServer received = {} reply = {"status": 200, "body": None} class Handler(BaseHTTPRequestHandler): def do_POST(self): length = int(self.headers.get("Content-Length", 0)) received["path"] = self.path received["auth"] = self.headers.get("Authorization") received["body"] = json.loads(self.rfile.read(length)) body = reply["body"] or { "id": "chatcmpl-test", "object": "chat.completion", "created": 0, "model": received["body"]["model"], "choices": [{"index": 0, "finish_reason": "stop", "message": {"role": "assistant", "content": ai_answer(), "refusal": None}}], "usage": {"prompt_tokens": 1, "completion_tokens": 1, "total_tokens": 2}, } data = json.dumps(body).encode() self.send_response(reply["status"]) self.send_header("Content-Type", "application/json") self.send_header("Content-Length", str(len(data))) self.end_headers() self.wfile.write(data) def log_message(self, *args): pass server = HTTPServer(("127.0.0.1", 0), Handler) threading.Thread(target=server.serve_forever, daemon=True).start() settings.OPENAI_API_KEY = "sk-test-local" settings.OPENAI_BASE_URL = f"http://127.0.0.1:{server.server_port}/v1" settings.OPENAI_MODEL = "stub-model" yield received, reply server.shutdown() @pytest.mark.django_db def test_end_to_end_ai_cleanup_through_sdk(stub_openai_server, user_client): received, _ = stub_openai_server response = user_client.post(reverse("recipes:create"), form_data(ai="1")) assert response.status_code == 200 assert response.context["form"].initial["title"] == "Linsen-Dal" assert received["path"] == "/v1/chat/completions" assert received["auth"] == "Bearer sk-test-local" body = received["body"] assert body["model"] == "stub-model" assert body["response_format"]["json_schema"]["strict"] is True assert body["messages"][0]["role"] == "system" assert not Recipe.objects.exists() @pytest.mark.django_db def test_end_to_end_api_error_through_sdk(stub_openai_server, user_client): _, reply = stub_openai_server reply["status"] = 401 reply["body"] = {"error": {"message": "Incorrect API key", "type": "invalid_request_error", "code": "invalid_api_key"}} response = user_client.post(reverse("recipes:create"), form_data(ai="1")) assert "API-Key ist ungültig" in response.content.decode() @pytest.mark.django_db def test_end_to_end_photo_import_through_sdk(stub_openai_server, user_client): received, _ = stub_openai_server response = user_client.post(reverse("recipes:import_photo"), {"photos": [image_upload()]}) assert response.context["form"].initial["title"] == "Linsen-Dal" parts = received["body"]["messages"][1]["content"] assert any(p["type"] == "image_url" and p["image_url"]["url"].startswith("data:image/jpeg;base64,") for p in parts)