dammann/Kochbuch
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
main
Kochbuch/tests/test_link_import.py
T
Kyle DammannandClaude Opus 5.5 60785e2437 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]>
2026-10-30 21:53:28 +01:00

331 lines
14 KiB
Python
Raw Permalink Blame History

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 &amp; 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: &quot;Dal mit 200 g Linsen und Kokosmilch&quot;">'
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 View Git Blame Copy Permalink