Kyle DammannandClaude Opus 5.5
015a0a5821
docker compose passes an empty .env line as OPENAI_BASE_URL="", and with base_url=None the openai SDK fell back to that empty value, so every call failed with "Connection error". Always pass the base URL explicitly. Also disable gunicorn's unused control socket, which logged a permission error because /app is read-only for the app user. Found while deploying to the homelab server. Co-Authored-By: Claude Opus 5.5 <[email protected]>
405 lines
16 KiB
Python
405 lines
16 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"),
|
|
# Real-world message seen from the API (2026-10) for an account without credit
|
|
(api_error(openai.RateLimitError, 429, "Error code: 429 - You have no credits remaining. Add credits to "
|
|
"continue using the API at https://platform.openai.com/settings/organization/billing/."), "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)
|
|
|
|
|
|
def test_empty_base_url_env_falls_back_to_openai(monkeypatch, settings):
|
|
"""docker compose sets OPENAI_BASE_URL="" for an empty .env line; the SDK must not pick that up."""
|
|
monkeypatch.setenv("OPENAI_BASE_URL", "")
|
|
settings.OPENAI_API_KEY = "sk-test"
|
|
settings.OPENAI_BASE_URL = ""
|
|
assert str(ai._client().base_url).rstrip("/") == "https://api.openai.com/v1"
|
|
settings.OPENAI_BASE_URL = "http://proxy.example/v1"
|
|
assert str(ai._client().base_url).rstrip("/") == "http://proxy.example/v1"
|