Fizz
c8775ac713
Mealie's POST /api/recipes/{slug}/image downloads the URL it is given and
rejects anything that is not an image, so handing it a recipe page answered
400 {"message": "Url is not an image"}. This was read as a source-specific
failure on ica.se, but that page's own og:image uploads fine (verified live,
200 + 74608 bytes of image/webp), and httpbin's JPEG works too — the endpoint
simply never accepted pages.
Callers naturally pass the recipe page they just imported from, so resolve a
page to its advertised og:image and retry rather than making every caller know
the distinction. A page with no discoverable image now raises instead of
leaving the recipe silently coverless.
Co-authored-by: fredamn76 <fredrik.fallman@gmail.com>
Signed-off-by: fredamn76 <fredrik.fallman@gmail.com>
351 lines
14 KiB
Python
351 lines
14 KiB
Python
"""Server-layer behaviour, driven through a mocked HTTP transport."""
|
||
from __future__ import annotations
|
||
|
||
import httpx
|
||
import pytest
|
||
|
||
from mealie_mcp import server
|
||
|
||
|
||
@pytest.fixture
|
||
def mock_mealie(monkeypatch):
|
||
"""Install a fake Mealie and record every request the server makes."""
|
||
def install(handler):
|
||
recorded: list[httpx.Request] = []
|
||
|
||
def wrapped(request: httpx.Request) -> httpx.Response:
|
||
recorded.append(request)
|
||
return handler(request)
|
||
|
||
def fake_client() -> httpx.Client:
|
||
return httpx.Client(
|
||
base_url="https://mealie.test",
|
||
headers={"Authorization": "Bearer test", "User-Agent": server.UA},
|
||
transport=httpx.MockTransport(wrapped),
|
||
)
|
||
|
||
monkeypatch.setattr(server, "_client", fake_client)
|
||
return recorded
|
||
|
||
return install
|
||
|
||
|
||
class TestErrorMapping:
|
||
def test_401_is_reported_as_an_auth_problem(self, mock_mealie):
|
||
mock_mealie(lambda r: httpx.Response(401, json={"detail": "Could not validate credentials"}))
|
||
result = server.check_auth()
|
||
assert result["ok"] is False
|
||
assert result["reason"] == "auth"
|
||
assert "token" in result["detail"]
|
||
|
||
def test_cloudflare_1010_is_not_mistaken_for_bad_credentials(self, mock_mealie):
|
||
# Verified failure: the external hostname returns 403 + "error code: 1010"
|
||
# for blocked client fingerprints. Calling that an auth error sends
|
||
# debugging down the wrong path.
|
||
mock_mealie(lambda r: httpx.Response(403, text="error code: 1010"))
|
||
result = server.check_auth()
|
||
assert result["reason"] == "waf"
|
||
assert "1010" in result["detail"]
|
||
|
||
def test_healthy_instance_reports_ok(self, mock_mealie):
|
||
mock_mealie(lambda r: httpx.Response(200, json={"username": "fredrik"}))
|
||
result = server.check_auth()
|
||
assert result["ok"] is True
|
||
assert result["user"] == "fredrik"
|
||
|
||
|
||
class TestUserAgent:
|
||
def test_requests_never_use_the_default_user_agent(self, mock_mealie):
|
||
recorded = mock_mealie(lambda r: httpx.Response(200, json={"username": "fredrik"}))
|
||
server.check_auth()
|
||
agent = recorded[0].headers["user-agent"]
|
||
assert agent == server.UA
|
||
assert not agent.startswith(("python-httpx", "Python-urllib"))
|
||
|
||
|
||
class TestDuplicateDetection:
|
||
def test_import_stops_when_the_source_url_already_exists(self, mock_mealie):
|
||
existing = {"slug": "kycklinggryta", "name": "Kycklinggryta", "orgURL": "https://koket.se/a"}
|
||
mock_mealie(lambda r: httpx.Response(200, json={"items": [existing]}))
|
||
result = server.import_recipe_url("https://koket.se/a")
|
||
assert result["imported"] is False
|
||
assert result["reason"] == "duplicate"
|
||
|
||
def test_trailing_slash_does_not_hide_a_duplicate(self, mock_mealie):
|
||
existing = {"slug": "x", "name": "X", "orgURL": "https://koket.se/a/"}
|
||
mock_mealie(lambda r: httpx.Response(200, json={"items": [existing]}))
|
||
assert server.find_by_source_url("https://koket.se/a")
|
||
|
||
|
||
class TestImportReport:
|
||
def test_report_flags_failed_extraction_and_junk_title(self):
|
||
recipe = {
|
||
"slug": "flaskfile",
|
||
"name": "Fläskfilé med svampsås - se & gör",
|
||
"recipeIngredient": [{"display": "Could not detect ingredients"}],
|
||
"recipeInstructions": [],
|
||
"image": "",
|
||
}
|
||
report = server._import_report(recipe)
|
||
assert report["extraction_failed"] is True
|
||
assert report["suggested_title"] == "Fläskfilé med svampsås"
|
||
assert report["has_image_field"] is False
|
||
|
||
def test_clean_import_suggests_no_title_change(self):
|
||
recipe = {"slug": "lax", "name": "Lax i ugn", "recipeIngredient": [], "image": "img"}
|
||
assert server._import_report(recipe)["suggested_title"] is None
|
||
|
||
|
||
class TestTextImport:
|
||
def test_rejects_text_without_explicit_sections_before_writing(self, monkeypatch):
|
||
monkeypatch.setattr(
|
||
server,
|
||
"_request",
|
||
lambda *_args, **_kwargs: pytest.fail("must not create a junk recipe"),
|
||
)
|
||
|
||
with pytest.raises(ValueError, match="Titel"):
|
||
server.import_recipe_text(
|
||
"Namnlöst recept\n\nIngredienser:\n1 dl vatten\n\nGör så här:\nBlanda."
|
||
)
|
||
|
||
def test_accepts_the_documented_swedish_section_format(self, monkeypatch):
|
||
text = (
|
||
"Titel: Testsoppa\n\nPortioner: 2\n\nIngredienser:\n1 dl vatten\n\n"
|
||
"Gör så här:\nBlanda."
|
||
)
|
||
recipe = {
|
||
"slug": "testsoppa",
|
||
"name": "Testsoppa",
|
||
"recipeIngredient": [{"display": "1 dl vatten"}],
|
||
"recipeInstructions": [{"text": "Blanda."}],
|
||
}
|
||
monkeypatch.setattr(server, "_request", lambda *_args, **_kwargs: "testsoppa")
|
||
monkeypatch.setattr(server, "get_recipe", lambda _slug: recipe)
|
||
|
||
assert server.import_recipe_text(text)["recipe"] == recipe
|
||
|
||
|
||
class TestDeleteRecipe:
|
||
def test_requires_the_exact_stored_slug(self, monkeypatch):
|
||
monkeypatch.setattr(
|
||
server,
|
||
"get_recipe",
|
||
lambda _slug: {"slug": "stored-slug", "name": "Test"},
|
||
)
|
||
monkeypatch.setattr(
|
||
server,
|
||
"_client",
|
||
lambda: pytest.fail("must not delete without exact confirmation"),
|
||
)
|
||
|
||
with pytest.raises(ValueError, match="stored-slug"):
|
||
server.delete_recipe("recipe-id", "wrong-slug")
|
||
|
||
def test_deletes_and_proves_the_recipe_is_gone(self, mock_mealie):
|
||
recipe = {"slug": "zzz-test", "name": "ZZZ Test"}
|
||
|
||
def handler(request: httpx.Request) -> httpx.Response:
|
||
if request.method == "DELETE":
|
||
return httpx.Response(200, json={})
|
||
if request.url.path == "/api/recipes/zzz-test":
|
||
# First GET is get_recipe; second GET is the deletion proof.
|
||
if getattr(handler, "read_once", False):
|
||
return httpx.Response(404, json={"detail": "Not Found"})
|
||
handler.read_once = True
|
||
return httpx.Response(200, json=recipe)
|
||
return httpx.Response(500)
|
||
|
||
mock_mealie(handler)
|
||
result = server.delete_recipe("zzz-test", "zzz-test")
|
||
|
||
assert result == {
|
||
"deleted": True,
|
||
"slug": "zzz-test",
|
||
"name": "ZZZ Test",
|
||
"delete_status": 200,
|
||
"readback_status": 404,
|
||
}
|
||
|
||
|
||
class TestParseIngredients:
|
||
def test_reverts_structure_when_mealie_rewords_the_swedish_line(self, mock_mealie):
|
||
canonical = "4,7 dl basmatiris"
|
||
recipe = {
|
||
"slug": "ris",
|
||
"recipeIngredient": [{"display": canonical, "note": canonical}],
|
||
}
|
||
patches: list[list[dict]] = []
|
||
reads = 0
|
||
|
||
def handler(request: httpx.Request) -> httpx.Response:
|
||
import json as _json
|
||
|
||
nonlocal reads
|
||
if request.url.path == "/api/parser/ingredients":
|
||
return httpx.Response(200, json=[{
|
||
"ingredient": {
|
||
"quantity": 4.7,
|
||
"unit": {"id": "liter", "name": "liter"},
|
||
"food": {"id": "f", "name": "basmatiris"},
|
||
}
|
||
}])
|
||
if request.url.path == "/api/units":
|
||
return httpx.Response(
|
||
200,
|
||
json={"items": [{"id": "dl", "name": "dl", "abbreviation": "dl"}]},
|
||
)
|
||
if request.method == "PATCH":
|
||
patches.append(_json.loads(request.content)["recipeIngredient"])
|
||
return httpx.Response(200, json={})
|
||
reads += 1
|
||
if reads == 1:
|
||
return httpx.Response(200, json=recipe)
|
||
if len(patches) == 1:
|
||
# This is what Mealie rendered from the first structured patch.
|
||
return httpx.Response(200, json={
|
||
"slug": "ris",
|
||
"recipeIngredient": [{
|
||
**patches[0][0],
|
||
"display": "47⁄10 dl basmatiris",
|
||
}],
|
||
})
|
||
return httpx.Response(200, json=recipe)
|
||
|
||
mock_mealie(handler)
|
||
result = server.parse_ingredients("ris")
|
||
|
||
assert len(patches) == 2
|
||
assert patches[0][0]["unit"]["id"] == "dl"
|
||
assert patches[1][0]["quantity"] == 0
|
||
assert patches[1][0]["unit"] is None
|
||
assert patches[1][0]["food"] is None
|
||
assert patches[1][0]["note"] == canonical
|
||
assert result["lines_unchanged"] is True
|
||
assert result["left_unstructured"] == [{
|
||
"line": canonical,
|
||
"mealie_would_show": "47⁄10 dl basmatiris",
|
||
}]
|
||
|
||
def test_failed_structured_patch_preserves_the_readable_import(self, mock_mealie):
|
||
recipe = {"slug": "ris", "recipeIngredient": [{"display": "2 dl grädde"}]}
|
||
|
||
def handler(request: httpx.Request) -> httpx.Response:
|
||
if request.url.path == "/api/parser/ingredients":
|
||
return httpx.Response(200, json=[{"ingredient": {"quantity": 2}}])
|
||
if request.method == "PATCH":
|
||
# Verified: this PATCH can 500 with a ValueError on this instance.
|
||
return httpx.Response(500, json={"detail": "ValueError"})
|
||
return httpx.Response(200, json=recipe)
|
||
|
||
mock_mealie(handler)
|
||
result = server.parse_ingredients("ris")
|
||
assert result["parsed"] is False
|
||
assert result["canonical_lines"] == ["2 dl grädde"]
|
||
|
||
|
||
class TestCoverImage:
|
||
def test_local_upload_sends_the_required_extension_field(self, mock_mealie, tmp_path):
|
||
# Verified: without the multipart `extension` field the upload fails validation.
|
||
image = tmp_path / "cover.webp"
|
||
image.write_bytes(b"fake")
|
||
recorded = mock_mealie(lambda r: httpx.Response(200, json={"slug": "x", "image": "img"}))
|
||
server.set_cover_image("x", image_path=str(image))
|
||
upload = next(r for r in recorded if r.method == "PUT")
|
||
body = upload.content.decode("utf-8", errors="replace")
|
||
assert 'name="extension"' in body
|
||
assert "webp" in body
|
||
|
||
def test_requires_a_source(self):
|
||
with pytest.raises(ValueError):
|
||
server.set_cover_image("x")
|
||
|
||
def test_a_recipe_page_is_resolved_to_its_hero_image(self, mock_mealie, monkeypatch):
|
||
# Verified live: Mealie answers 400 "Url is not an image" for a recipe page,
|
||
# and 200 for that same page's og:image.
|
||
page = "https://www.ica.se/recept/kalsoppa-722069/"
|
||
hero = "https://assets.icanet.se/imagevaultfiles/kalsoppa.jpg"
|
||
posted: list[str] = []
|
||
|
||
def handler(request: httpx.Request) -> httpx.Response:
|
||
import json as _json
|
||
|
||
if request.method == "POST" and request.url.path.endswith("/image"):
|
||
url = _json.loads(request.content)["url"]
|
||
posted.append(url)
|
||
if url == page:
|
||
return httpx.Response(
|
||
400, json={"detail": {"message": "Url is not an image", "error": True}}
|
||
)
|
||
return httpx.Response(200, json=None)
|
||
return httpx.Response(200, json={"slug": "soppa", "image": "abcd"})
|
||
|
||
mock_mealie(handler)
|
||
monkeypatch.setattr(
|
||
server, "_hero_image_url", lambda url: hero if url == page else None
|
||
)
|
||
|
||
result = server.set_cover_image("soppa", source_url=page)
|
||
|
||
# The page is tried first, then retried with the image it advertises.
|
||
assert posted == [page, hero]
|
||
assert result["method"] == "scrape (resolved og:image)"
|
||
assert result["resolved_image_url"] == hero
|
||
|
||
def test_a_direct_image_url_is_sent_unchanged(self, mock_mealie, monkeypatch):
|
||
recorded = mock_mealie(lambda r: httpx.Response(200, json={"slug": "x", "image": "img"}))
|
||
monkeypatch.setattr(server, "_hero_image_url", lambda url: pytest.fail("not needed"))
|
||
|
||
result = server.set_cover_image("x", source_url="https://example.com/hero.jpg")
|
||
|
||
assert result["method"] == "scrape"
|
||
assert "resolved_image_url" not in result
|
||
assert any(r.method == "POST" for r in recorded)
|
||
|
||
def test_a_page_with_no_hero_image_fails_loudly(self, mock_mealie, monkeypatch):
|
||
def handler(request: httpx.Request) -> httpx.Response:
|
||
if request.method == "POST" and request.url.path.endswith("/image"):
|
||
return httpx.Response(400, json={"detail": {"message": "Url is not an image"}})
|
||
return httpx.Response(200, json={"slug": "x"})
|
||
|
||
mock_mealie(handler)
|
||
monkeypatch.setattr(server, "_hero_image_url", lambda url: None)
|
||
|
||
with pytest.raises(ValueError, match="no og:image"):
|
||
server.set_cover_image("x", source_url="https://example.com/page")
|
||
|
||
|
||
class TestHeroImageUrl:
|
||
def test_reads_og_image_in_either_attribute_order(self, monkeypatch):
|
||
pages = {
|
||
"a": '<meta property="og:image" content="https://x.test/a.jpg">',
|
||
"b": '<meta content="https://x.test/b.jpg" property="og:image">',
|
||
"c": '<meta name="twitter:image" content="https://x.test/c.jpg">',
|
||
"d": "<html><body>no image here</body></html>",
|
||
}
|
||
|
||
def fake_get(url, **kwargs):
|
||
return httpx.Response(200, text=pages[url], request=httpx.Request("GET", "http://t"))
|
||
|
||
monkeypatch.setattr(server.httpx, "get", fake_get)
|
||
assert server._hero_image_url("a") == "https://x.test/a.jpg"
|
||
assert server._hero_image_url("b") == "https://x.test/b.jpg"
|
||
assert server._hero_image_url("c") == "https://x.test/c.jpg"
|
||
assert server._hero_image_url("d") is None
|
||
|
||
def test_unescapes_entities_in_the_url(self, monkeypatch):
|
||
# Query strings in og:image arrive HTML-escaped and must be decoded.
|
||
page = '<meta property="og:image" content="https://x.test/a.jpg?w=1&h=2">'
|
||
monkeypatch.setattr(
|
||
server.httpx, "get",
|
||
lambda url, **kw: httpx.Response(200, text=page, request=httpx.Request("GET", "http://t")),
|
||
)
|
||
assert server._hero_image_url("p") == "https://x.test/a.jpg?w=1&h=2"
|
||
|
||
def test_an_unreachable_page_is_not_an_error(self, monkeypatch):
|
||
def boom(url, **kwargs):
|
||
raise httpx.ConnectError("no route")
|
||
|
||
monkeypatch.setattr(server.httpx, "get", boom)
|
||
assert server._hero_image_url("https://example.com/page") is None
|