Files
mealie-mcp/tests/test_server.py
T
fredamn76 42d78f4060 Address recipes by slug on sub-routes, and tags as real objects
Two failures from today's TikTok import, both misread as faults on the
Mealie instance.

set_cover_image passed slug_or_id straight into /api/recipes/{slug}/image.
That path only resolves slugs: /api/recipes/{slug} itself accepts an id, the
routes beneath it do not. With the recipe's id the scrape POST answered 404
and the file upload PUT answered 500 — neither of which says "wrong key", so
the agent concluded the image endpoint was broken and gave up on the cover.
Verified on this instance (v3.22.0): GET /api/recipes/<id>/comments answers
500 while the same call with the slug answers 200. The recipe is now read
first and its stored slug used for the sub-route.

patch_recipe sent tags through untouched, so {"name": "Källa: TikTok"} went
out as-is and came back 422. RecipeTag and RecipeCategory both require name
*and* slug per the instance's OpenAPI schema. Tags and categories are now
resolved before the PATCH: an existing organizer is reused (case-insensitive
on name) so tagging cannot fork a vocabulary that already holds "Källa:
Instagram" and "Källa: YouTube", and a genuinely new name is created through
its own endpoint, which supplies the slug Mealie itself would pick.

Neither path had a test that could have caught this: the cover tests always
passed a slug, and the patch contract test asserted the 422-producing shape
as if it were correct.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
2026-08-17 14:42:45 +02:00

528 lines
22 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""Server-layer behaviour, driven through a mocked HTTP transport."""
from __future__ import annotations
import json
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
def test_text_is_never_posted_to_the_scraper_endpoint(self, mock_mealie, monkeypatch):
"""The scraper answers 400 for anything without schema.org markup.
Plain Swedish text has none, so every text import through
``/api/recipes/create/html-or-json`` failed. Pin the endpoints instead of
the payload, because that is what the mocked-out tests could not see.
"""
def handler(request: httpx.Request) -> httpx.Response:
if request.url.path == "/api/recipes/create/html-or-json":
return httpx.Response(400, json={"detail": "no recipe data found"})
if request.method == "POST":
return httpx.Response(201, json="testsoppa")
return httpx.Response(200, json={"slug": "testsoppa"})
recorded = mock_mealie(handler)
monkeypatch.setattr(server, "get_recipe", lambda _slug: {"slug": "testsoppa"})
server.import_recipe_text(
"Titel: Testsoppa\n\nPortioner: 2\n\nIngredienser:\n1 dl vatten\n\n"
"Gör så här:\nBlanda."
)
paths = [(r.method, r.url.path) for r in recorded]
assert ("POST", "/api/recipes") in paths
assert ("PATCH", "/api/recipes/testsoppa") in paths
assert all(path != "/api/recipes/create/html-or-json" for _method, path in paths)
def test_sections_become_readable_lines_steps_and_yield(self, mock_mealie, monkeypatch):
def handler(request: httpx.Request) -> httpx.Response:
if request.method == "POST":
return httpx.Response(201, json="testsoppa")
return httpx.Response(200, json={"slug": "testsoppa"})
recorded = mock_mealie(handler)
monkeypatch.setattr(server, "get_recipe", lambda _slug: {"slug": "testsoppa"})
server.import_recipe_text(
"Titel: Testsoppa\n\nPortioner: 4-6\n\nIngredienser:\n1 gul lök\n2 dl riven ost\n\n"
"Gör så här:\n1. Hacka löken.\n2. Rör ner osten.",
source_url="https://example.test/klipp",
)
create = json.loads(recorded[0].content)
patch = json.loads(recorded[1].content)
assert create == {"name": "Testsoppa"}
assert [item["note"] for item in patch["recipeIngredient"]] == ["1 gul lök", "2 dl riven ost"]
assert [step["text"] for step in patch["recipeInstructions"]] == [
"Hacka löken.",
"Rör ner osten.",
]
assert patch["recipeYield"] == "4-6"
assert patch["orgURL"] == "https://example.test/klipp"
def test_a_marked_reconstruction_heading_is_still_a_heading(self, mock_mealie, monkeypatch):
"""``Gör så här (REKONSTRUERAD):`` marks the source, not the section name.
The old exact-match check rejected it, which is how a correctly marked
import got turned away before it reached Mealie.
"""
def handler(request: httpx.Request) -> httpx.Response:
if request.method == "POST":
return httpx.Response(201, json="testsoppa")
return httpx.Response(200, json={"slug": "testsoppa"})
recorded = mock_mealie(handler)
monkeypatch.setattr(server, "get_recipe", lambda _slug: {"slug": "testsoppa"})
server.import_recipe_text(
"Titel: Testsoppa\n\nIngredienser:\n1 dl vatten\n\n"
"Gör så här (REKONSTRUERAD - står INTE i källan):\nBlanda."
)
patch = json.loads(recorded[1].content)
assert [step["text"] for step in patch["recipeInstructions"]] == ["Blanda."]
def test_a_failed_content_patch_names_the_empty_recipe_it_left(
self, mock_mealie, monkeypatch
):
def handler(request: httpx.Request) -> httpx.Response:
if request.method == "POST":
return httpx.Response(201, json="testsoppa")
return httpx.Response(500, json={"detail": "boom"})
mock_mealie(handler)
monkeypatch.setattr(server, "get_recipe", lambda _slug: pytest.fail("patch failed"))
with pytest.raises(RuntimeError, match="testsoppa"):
server.import_recipe_text(
"Titel: Testsoppa\n\nIngredienser:\n1 dl vatten\n\nGör så här:\nBlanda."
)
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": "4710 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": "4710 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 TestOrganizerPatch:
"""Tags and categories must reach Mealie as full RecipeTag/RecipeCategory objects.
Verified against the instance's OpenAPI schema (v3.22.0): both require ``name``
*and* ``slug``, so a bare ``{"name": ...}`` is answered with 422.
"""
def _mealie(self, mock_mealie, existing, recipe=None):
recipe = recipe or {"id": "stable-id", "slug": "soppa"}
bodies: list[dict] = []
def handler(request: httpx.Request) -> httpx.Response:
path = request.url.path
if path.startswith("/api/organizers/"):
if request.method == "GET":
return httpx.Response(200, json={"items": existing})
created = json.loads(request.content)
made = {"id": "new-id", "name": created["name"], "slug": "kalla-tiktok"}
bodies.append({"created": made})
return httpx.Response(201, json=made)
if request.method == "PATCH":
bodies.append({"patch": json.loads(request.content)})
return httpx.Response(200, json=recipe)
return bodies, mock_mealie(handler)
def test_a_bare_name_is_created_and_patched_with_its_slug(self, mock_mealie):
bodies, _ = self._mealie(mock_mealie, existing=[])
server.patch_recipe("soppa", {"tags": [{"name": "Källa: TikTok"}]})
created = [b["created"] for b in bodies if "created" in b]
patch = next(b["patch"] for b in bodies if "patch" in b)
assert created == [{"id": "new-id", "name": "Källa: TikTok", "slug": "kalla-tiktok"}]
assert patch["tags"] == created
def test_an_existing_organizer_is_reused_instead_of_forking_the_vocabulary(self, mock_mealie):
existing = [{"id": "tag-1", "name": "Källa: TikTok", "slug": "kalla-tiktok"}]
bodies, recorded = self._mealie(mock_mealie, existing=existing)
server.patch_recipe("soppa", {"tags": ["källa: tiktok"]})
patch = next(b["patch"] for b in bodies if "patch" in b)
assert patch["tags"] == existing
assert not any(r.method == "POST" for r in recorded)
def test_a_complete_object_is_passed_through_untouched(self, mock_mealie):
tag = {"id": "tag-1", "name": "📅 Vardag", "slug": "vardag"}
bodies, recorded = self._mealie(mock_mealie, existing=[])
server.patch_recipe("soppa", {"tags": [tag], "description": "Av: Matprinsessan"})
patch = next(b["patch"] for b in bodies if "patch" in b)
assert patch == {"tags": [tag], "description": "Av: Matprinsessan"}
assert not any(r.url.path.startswith("/api/organizers/") for r in recorded)
def test_categories_go_through_their_own_endpoint(self, mock_mealie):
_, recorded = self._mealie(mock_mealie, existing=[])
server.patch_recipe("soppa", {"recipeCategory": ["Såser"]})
lookups = [r.url.path for r in recorded if r.url.path.startswith("/api/organizers/")]
assert lookups == ["/api/organizers/categories", "/api/organizers/categories"]
class TestCoverImage:
def test_an_id_is_resolved_to_the_slug_before_hitting_the_image_route(self, mock_mealie):
# Verified on v3.22.0: /api/recipes/{slug} takes an id, but the sub-routes
# under it are slug-only — an id answered 404 on POST .../image and 500 on
# PUT .../image, which is what silently killed the TikTok cover.
recipe_id = "1488dd5c-abbe-4660-b61b-eeca3f6cfb91"
def handler(request: httpx.Request) -> httpx.Response:
if request.url.path.endswith("/image"):
assert request.url.path == "/api/recipes/skolans-ost-broccolisas/image"
return httpx.Response(200, json=None)
return httpx.Response(200, json={"slug": "skolans-ost-broccolisas", "image": "img"})
recorded = mock_mealie(handler)
result = server.set_cover_image(recipe_id, source_url="https://cdn.test/cover.jpg")
assert result["slug"] == "skolans-ost-broccolisas"
assert not any(recipe_id in str(r.url) and r.url.path.endswith("/image") for r in recorded)
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&amp;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