Add phase 5: link import, Instagram fallback and share target
- SSRF-safe fetcher: public IPs only, connects to the checked address, redirects re-validated, size/time limits, honest user agent - schema.org/Recipe JSON-LD extraction (graphs, HowToSections, ISO durations, yields, keywords, images) without extra dependencies; AI fallback on plain page text when no structured recipe exists - Instagram: one login-free request to the public embed page; when Instagram withholds the caption, the import page explains how to paste the text or import a screenshot instead - Suggested source photo can be downloaded and attached on save - Text import works without AI (text becomes the description) - Installable PWA with manifest, icons and Android share target - Offline tests for SSRF rules, parsers, Instagram and import views Co-Authored-By: Claude Opus 5.5 <[email protected]>
This commit is contained in:
Kyle Dammann
co-authored by
Claude Opus 5.5
Claude Opus 5.5
parent
1bf8ef8b65
commit
60785e2437
+12
-2
@@ -164,9 +164,19 @@ def test_without_key_ai_is_disabled_but_app_works(settings, user_client):
|
||||
|
||||
|
||||
@pytest.mark.django_db
|
||||
def test_without_key_imports_give_clear_message(settings, user_client):
|
||||
def test_without_key_text_import_becomes_description(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"})
|
||||
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()
|
||||
|
||||
|
||||
|
||||
@@ -0,0 +1,330 @@
|
||||
import io
|
||||
import json
|
||||
import socket
|
||||
|
||||
import pytest
|
||||
from django.urls import reverse
|
||||
from PIL import Image
|
||||
|
||||
from recipes import ai, fetch, import_views, instagram, schema_org
|
||||
from recipes import views as recipe_views
|
||||
from recipes.fetch import FetchError, FetchResult
|
||||
from recipes.models import Recipe
|
||||
|
||||
from .recipe_helpers import form_data
|
||||
from .test_ai import ai_answer
|
||||
|
||||
# --- SSRF protection ------------------------------------------------------------------------
|
||||
|
||||
|
||||
def fake_dns(monkeypatch, ip):
|
||||
monkeypatch.setattr(socket, "getaddrinfo", lambda *a, **k: [(socket.AF_INET, socket.SOCK_STREAM, 6, "", (ip, 443))])
|
||||
|
||||
|
||||
@pytest.mark.parametrize("ip", ["127.0.0.1", "10.1.2.3", "192.168.178.1", "172.17.0.1", "169.254.169.254", "100.64.0.1", "0.0.0.0"])
|
||||
def test_internal_addresses_blocked(monkeypatch, ip):
|
||||
fake_dns(monkeypatch, ip)
|
||||
with pytest.raises(FetchError, match="nicht erlaubt"):
|
||||
fetch.fetch("https://harmless-looking.example/recipe")
|
||||
|
||||
|
||||
def test_ipv6_loopback_and_mapped_blocked(monkeypatch):
|
||||
for ip in ["::1", "::ffff:127.0.0.1", "fd00::1"]:
|
||||
monkeypatch.setattr(socket, "getaddrinfo", lambda *a, ip=ip, **k: [(socket.AF_INET6, socket.SOCK_STREAM, 6, "", (ip, 443, 0, 0))])
|
||||
with pytest.raises(FetchError):
|
||||
fetch.fetch("https://x.example/")
|
||||
|
||||
|
||||
@pytest.mark.parametrize("url", ["file:///etc/passwd", "ftp://x.example/", "https://user:[email protected]/", "https://x.example:22/", "gopher://x"])
|
||||
def test_bad_urls_rejected(url):
|
||||
with pytest.raises(FetchError):
|
||||
fetch.fetch(url)
|
||||
|
||||
|
||||
def test_redirect_target_is_checked_again(monkeypatch):
|
||||
class Response:
|
||||
status = 302
|
||||
headers = {"Location": "http://internal.example/admin"}
|
||||
|
||||
def release_conn(self):
|
||||
pass
|
||||
|
||||
def fake_request(url, accept, timeout):
|
||||
if "internal" in url:
|
||||
raise FetchError("Diese Adresse ist nicht erlaubt (internes Netz).")
|
||||
return Response()
|
||||
|
||||
monkeypatch.setattr(fetch, "_request_once", fake_request)
|
||||
with pytest.raises(FetchError, match="nicht erlaubt"):
|
||||
fetch.fetch("https://public.example/")
|
||||
|
||||
|
||||
# --- schema.org parsing -----------------------------------------------------------------------
|
||||
|
||||
RECIPE_LD = {
|
||||
"@context": "https://schema.org",
|
||||
"@graph": [
|
||||
{"@type": "WebPage", "name": "Seite"},
|
||||
{
|
||||
"@type": ["Recipe", "NewsArticle"],
|
||||
"name": "Kürbissuppe & Ingwer",
|
||||
"description": "<p>Wärmt von innen.</p>",
|
||||
"recipeYield": ["4", "4 Portionen"],
|
||||
"prepTime": "PT15M",
|
||||
"cookTime": "PT1H5M",
|
||||
"image": [{"@type": "ImageObject", "url": "https://cdn.example/suppe.jpg"}],
|
||||
"keywords": "Suppe, Herbst, vegan",
|
||||
"recipeCategory": "Hauptgericht",
|
||||
"recipeIngredient": ["1 kg Hokkaido", "2 cm Ingwer", "400 ml Kokosmilch"],
|
||||
"recipeInstructions": [
|
||||
{"@type": "HowToSection", "name": "Vorbereitung", "itemListElement": [
|
||||
{"@type": "HowToStep", "text": "Kürbis würfeln."},
|
||||
]},
|
||||
{"@type": "HowToStep", "text": "Alles <b>pürieren</b>."},
|
||||
],
|
||||
},
|
||||
],
|
||||
}
|
||||
|
||||
|
||||
def page_with(ld, extra=""):
|
||||
return f"""<html><head><title>Suppe</title><meta property="og:image" content="https://cdn.example/og.jpg">
|
||||
<script type="application/ld+json">{{ broken json,</script>
|
||||
<script type="application/ld+json">{json.dumps(ld)}</script>{extra}</head><body><p>Text</p></body></html>"""
|
||||
|
||||
|
||||
def test_schema_org_recipe_extracted():
|
||||
initial, image, _ = schema_org.extract(page_with(RECIPE_LD))
|
||||
assert initial["title"] == "Kürbissuppe & Ingwer"
|
||||
assert initial["description"] == "Wärmt von innen."
|
||||
assert initial["servings"] == 4 and initial["prep_minutes"] == 15 and initial["cook_minutes"] == 65
|
||||
assert initial["ingredients_text"].splitlines() == ["1 kg Hokkaido", "2 cm Ingwer", "400 ml Kokosmilch"]
|
||||
assert initial["steps_text"] == "Kürbis würfeln.\n\nAlles pürieren."
|
||||
assert initial["tags_text"] == "hauptgericht, suppe, herbst, vegan"
|
||||
assert image == "https://cdn.example/suppe.jpg"
|
||||
|
||||
|
||||
def test_schema_org_string_instructions_and_yield():
|
||||
ld = {"@type": "Recipe", "name": "X", "recipeYield": "12 Stück", "recipeIngredient": ["Mehl"],
|
||||
"recipeInstructions": "Erst das.<br>Dann das."}
|
||||
initial, image, _ = schema_org.extract(page_with(ld))
|
||||
assert initial["servings"] == 12 and initial["servings_label"] == "Stück"
|
||||
assert initial["steps_text"] == "Erst das.\n\nDann das."
|
||||
assert image == "https://cdn.example/og.jpg" # falls back to og:image
|
||||
|
||||
|
||||
def test_page_without_recipe():
|
||||
initial, image, page = schema_org.extract("<html><head><title>Blog</title></head><body><nav>Menü</nav><p>Hallo Welt</p><script>x()</script></body></html>")
|
||||
assert initial is None
|
||||
assert "Hallo Welt" in page.visible_text() and "Menü" not in page.visible_text() and "x()" not in page.visible_text()
|
||||
|
||||
|
||||
# --- Instagram ---------------------------------------------------------------------------------
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"url, code",
|
||||
[
|
||||
("https://www.instagram.com/p/C1a2b3c4D5e/", "C1a2b3c4D5e"),
|
||||
("https://instagram.com/reel/C1a2b3c4D5e?igsh=abc", "C1a2b3c4D5e"),
|
||||
("https://www.instagram.com/kochprofi/p/C1a2b3c4D5e/", "C1a2b3c4D5e"),
|
||||
("https://www.instagram.com/reels/DaU7f-CjE1c/", "DaU7f-CjE1c"),
|
||||
("https://www.instagram.com/kochprofi/", None),
|
||||
("https://evil.example/p/C1a2b3c4D5e/", None),
|
||||
("https://instagram.com.evil.example/p/C1a2b3c4D5e/", None),
|
||||
],
|
||||
)
|
||||
def test_shortcode(url, code):
|
||||
assert instagram.shortcode(url) == code
|
||||
|
||||
|
||||
EMBED_HTML = """<html><body><div class="Embed">
|
||||
<img class="EmbeddedMediaImage" src="https://scontent.cdninstagram.com/x.jpg">
|
||||
<div class="Caption"><a class="CaptionUsername" href="#">kochprofi</a><br><br>Einfaches Dal 🍛<br><br>
|
||||
Zutaten:<br>200 g rote Linsen<br>1 Dose Kokosmilch<br><br>Linsen kochen, fertig!
|
||||
<div class="CaptionComments"><a href="#">Alle 23 Kommentare ansehen</a></div></div></div></body></html>"""
|
||||
|
||||
|
||||
def test_parse_embed_caption():
|
||||
caption, image = instagram.parse_embed(EMBED_HTML)
|
||||
assert caption.startswith("Einfaches Dal")
|
||||
assert "200 g rote Linsen" in caption and "kochprofi" not in caption and "Kommentare" not in caption
|
||||
assert image == "https://scontent.cdninstagram.com/x.jpg"
|
||||
|
||||
|
||||
def test_parse_og_description_fallback():
|
||||
html = '<meta property="og:description" content="1.234 likes, 12 comments - kochprofi am 1. März 2026: "Dal mit 200 g Linsen und Kokosmilch"">'
|
||||
caption, _ = instagram.parse_embed(html)
|
||||
assert caption == "Dal mit 200 g Linsen und Kokosmilch"
|
||||
|
||||
|
||||
def test_app_shell_means_unavailable(monkeypatch):
|
||||
monkeypatch.setattr(instagram, "fetch", lambda url: FetchResult(url, 200, "text/html", b"<html><title>Instagram</title></html>"))
|
||||
with pytest.raises(instagram.InstagramUnavailable):
|
||||
instagram.fetch_caption("C1a2b3c4D5e")
|
||||
|
||||
|
||||
# --- Views ---------------------------------------------------------------------------------------
|
||||
|
||||
def stub_fetch(monkeypatch, html, status=200, content_type="text/html; charset=utf-8"):
|
||||
calls = []
|
||||
|
||||
def fake(url, **kwargs):
|
||||
calls.append(url)
|
||||
return FetchResult(url, status, content_type, html.encode())
|
||||
|
||||
monkeypatch.setattr(import_views, "fetch", fake)
|
||||
return calls
|
||||
|
||||
|
||||
@pytest.mark.django_db
|
||||
def test_link_import_uses_schema_org_without_ai(monkeypatch, settings, user_client):
|
||||
settings.OPENAI_API_KEY = ""
|
||||
stub_fetch(monkeypatch, page_with(RECIPE_LD))
|
||||
response = user_client.post(reverse("recipes:import_link"), {"url": "https://rezepte.example/suppe"})
|
||||
assert response.status_code == 200
|
||||
initial = response.context["form"].initial
|
||||
assert initial["title"] == "Kürbissuppe & Ingwer"
|
||||
assert initial["source_url"] == "https://rezepte.example/suppe"
|
||||
html = response.content.decode()
|
||||
assert 'name="import_image_url"' in html and "Bild der Quelle" in html
|
||||
assert not Recipe.objects.exists()
|
||||
|
||||
|
||||
@pytest.mark.django_db
|
||||
def test_link_import_falls_back_to_ai(monkeypatch, settings, user_client):
|
||||
settings.OPENAI_API_KEY = "sk-test"
|
||||
stub_fetch(monkeypatch, "<html><title>Omas Dal</title><body><p>200 g Linsen kochen ...</p></body></html>")
|
||||
seen = []
|
||||
monkeypatch.setattr(ai, "_create_completion", lambda messages: seen.append(messages) or ai_answer())
|
||||
response = user_client.post(reverse("recipes:import_link"), {"url": "https://blog.example/dal"})
|
||||
assert response.context["form"].initial["title"] == "Linsen-Dal"
|
||||
assert "200 g Linsen" in seen[0][1]["content"]
|
||||
|
||||
|
||||
@pytest.mark.django_db
|
||||
def test_link_import_without_recipe_and_without_ai(monkeypatch, settings, user_client):
|
||||
settings.OPENAI_API_KEY = ""
|
||||
stub_fetch(monkeypatch, "<html><body>nichts</body></html>")
|
||||
response = user_client.post(reverse("recipes:import_link"), {"url": "https://blog.example/x"})
|
||||
assert response.status_code == 400
|
||||
assert "kein strukturiertes Rezept" in response.content.decode()
|
||||
|
||||
|
||||
@pytest.mark.django_db
|
||||
def test_link_import_reports_fetch_errors(monkeypatch, user_client):
|
||||
def fail(url, **kwargs):
|
||||
raise FetchError("Die Seite ist nicht erreichbar.")
|
||||
|
||||
monkeypatch.setattr(import_views, "fetch", fail)
|
||||
response = user_client.post(reverse("recipes:import_link"), {"url": "https://down.example/"})
|
||||
assert "nicht erreichbar" in response.content.decode()
|
||||
|
||||
|
||||
@pytest.mark.django_db
|
||||
def test_instagram_failure_offers_paste_fallback(monkeypatch, user_client):
|
||||
def unavailable(code):
|
||||
raise instagram.InstagramUnavailable("nope")
|
||||
|
||||
monkeypatch.setattr(instagram, "fetch_caption", unavailable)
|
||||
url = "https://www.instagram.com/p/C1a2b3c4D5e/"
|
||||
response = user_client.post(reverse("recipes:import_link"), {"url": url}, follow=True)
|
||||
html = response.content.decode()
|
||||
assert "So klappt es mit Instagram" in html
|
||||
assert response.context["text_form"].initial["source_url"] == url
|
||||
|
||||
|
||||
@pytest.mark.django_db
|
||||
def test_instagram_caption_structured_by_ai(monkeypatch, settings, user_client):
|
||||
settings.OPENAI_API_KEY = "sk-test"
|
||||
monkeypatch.setattr(instagram, "fetch_caption", lambda code: ("Dal: 200 g Linsen, Kokosmilch. Kochen!", "https://cdn.example/i.jpg"))
|
||||
seen = []
|
||||
monkeypatch.setattr(ai, "_create_completion", lambda messages: seen.append(messages) or ai_answer())
|
||||
response = user_client.post(reverse("recipes:import_link"), {"url": "https://www.instagram.com/reel/C1a2b3c4D5e/"})
|
||||
initial = response.context["form"].initial
|
||||
assert initial["title"] == "Linsen-Dal" and initial["import_image_url"] == "https://cdn.example/i.jpg"
|
||||
assert "Instagram" in seen[0][1]["content"]
|
||||
|
||||
|
||||
@pytest.mark.django_db
|
||||
def test_instagram_profile_link_rejected(user_client):
|
||||
response = user_client.post(reverse("recipes:import_link"), {"url": "https://www.instagram.com/kochprofi/"})
|
||||
assert "einzelnen Beitrag" in response.content.decode()
|
||||
|
||||
|
||||
def jpeg_bytes():
|
||||
buffer = io.BytesIO()
|
||||
Image.new("RGB", (800, 600), (10, 200, 10)).save(buffer, "JPEG")
|
||||
return buffer.getvalue()
|
||||
|
||||
|
||||
@pytest.mark.django_db
|
||||
def test_saving_draft_downloads_source_image(monkeypatch, user_client):
|
||||
monkeypatch.setattr(recipe_views.fetch, "fetch",
|
||||
lambda url, **kw: FetchResult(url, 200, "image/jpeg", jpeg_bytes()))
|
||||
user_client.post(reverse("recipes:create"), form_data(import_image_url="https://cdn.example/i.jpg", use_import_image="on"))
|
||||
recipe = Recipe.objects.get()
|
||||
assert recipe.images.count() == 1 and recipe.images.get().is_cover
|
||||
|
||||
|
||||
@pytest.mark.django_db
|
||||
def test_source_image_skipped_when_unticked(monkeypatch, user_client):
|
||||
called = []
|
||||
monkeypatch.setattr(recipe_views.fetch, "fetch", lambda url, **kw: called.append(url))
|
||||
user_client.post(reverse("recipes:create"), form_data(import_image_url="https://cdn.example/i.jpg"))
|
||||
assert not called and Recipe.objects.get().images.count() == 0
|
||||
|
||||
|
||||
@pytest.mark.django_db
|
||||
def test_source_image_to_internal_address_is_refused(monkeypatch, user_client):
|
||||
fake_dns(monkeypatch, "10.0.0.5")
|
||||
response = user_client.post(reverse("recipes:create"),
|
||||
form_data(import_image_url="https://nas.example/secret.jpg", use_import_image="on"), follow=True)
|
||||
assert "nicht erlaubt" in response.content.decode()
|
||||
assert Recipe.objects.get().images.count() == 0
|
||||
|
||||
|
||||
@pytest.mark.django_db
|
||||
def test_source_image_must_be_an_image(monkeypatch, user_client):
|
||||
monkeypatch.setattr(recipe_views.fetch, "fetch", lambda url, **kw: FetchResult(url, 200, "text/html", b"<html>"))
|
||||
user_client.post(reverse("recipes:create"), form_data(import_image_url="https://x.example/", use_import_image="on"))
|
||||
assert Recipe.objects.get().images.count() == 0
|
||||
|
||||
|
||||
# --- PWA / share target ----------------------------------------------------------------------------
|
||||
|
||||
|
||||
@pytest.mark.django_db
|
||||
@pytest.mark.parametrize(
|
||||
"params, expected",
|
||||
[
|
||||
({"url": "https://www.instagram.com/p/C1a2b3c4D5e/"}, "https%3A%2F%2Fwww.instagram.com%2Fp%2FC1a2b3c4D5e%2F"),
|
||||
({"text": "Schau mal https://www.instagram.com/reel/XYZ12345/ lecker!"}, "reel%2FXYZ12345%2F"),
|
||||
],
|
||||
)
|
||||
def test_share_target_redirects_to_import(user_client, params, expected):
|
||||
response = user_client.get(reverse("recipes:share"), params)
|
||||
assert response.status_code == 302 and expected in response["Location"]
|
||||
|
||||
|
||||
@pytest.mark.django_db
|
||||
def test_share_target_with_plain_text(user_client):
|
||||
response = user_client.get(reverse("recipes:share"), {"text": "200 g Mehl und 2 Eier verrühren"})
|
||||
assert response.context["text_form"].initial["text"].startswith("200 g Mehl")
|
||||
|
||||
|
||||
@pytest.mark.django_db
|
||||
def test_import_page_prefills_url(user_client):
|
||||
response = user_client.get(reverse("recipes:import"), {"url": "https://x.example/r"})
|
||||
assert response.context["link_form"].initial["url"] == "https://x.example/r"
|
||||
|
||||
|
||||
@pytest.mark.django_db
|
||||
def test_manifest_and_service_worker(client):
|
||||
manifest = client.get("/manifest.webmanifest")
|
||||
assert manifest.status_code == 200 and manifest["Content-Type"] == "application/manifest+json"
|
||||
data = manifest.json()
|
||||
assert data["share_target"]["action"] == reverse("recipes:share")
|
||||
sw = client.get("/sw.js")
|
||||
assert sw.status_code == 200 and "javascript" in sw["Content-Type"]
|
||||
assert "fetch" not in sw.content.decode() # no caching of private pages
|
||||
Reference in New Issue
Block a user