Files

624 lines
26 KiB
Python
Raw Permalink Normal View History

2026-07-31 07:31:10 +02:00
"""Server-layer behaviour, driven through a mocked HTTP transport."""
from __future__ import annotations
import json
2026-07-31 07:31:10 +02:00
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"
# A range takes its lower bound; without a number Mealie cannot scale
# ingredients or show nutrition per serving.
assert patch["recipeServings"] == 4
def test_a_yield_without_a_number_leaves_servings_alone(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: en stor kastrull\n\nIngredienser:\n1 dl vatten\n\n"
"Gör så här:\nBlanda."
)
patch = json.loads(recorded[1].content)
assert patch["recipeYield"] == "en stor kastrull"
assert "recipeServings" not in patch
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,
}
2026-07-31 07:31:10 +02:00
class TestParseIngredients:
def test_reverts_structure_when_mealie_rewords_the_swedish_line(self, mock_mealie):
2026-07-31 07:31:10 +02:00
canonical = "4,7 dl basmatiris"
recipe = {
"slug": "ris",
"recipeIngredient": [{"display": canonical, "note": canonical}],
}
patches: list[list[dict]] = []
reads = 0
2026-07-31 07:31:10 +02:00
def handler(request: httpx.Request) -> httpx.Response:
import json as _json
nonlocal reads
2026-07-31 07:31:10 +02:00
if request.url.path == "/api/parser/ingredients":
return httpx.Response(200, json=[{
"ingredient": {
"quantity": 4.7,
"unit": {"id": "liter", "name": "liter"},
2026-07-31 07:31:10 +02:00
"food": {"id": "f", "name": "basmatiris"},
}
}])
if request.url.path == "/api/units":
return httpx.Response(
200,
json={"items": [{"id": "dl", "name": "dl", "abbreviation": "dl"}]},
)
2026-07-31 07:31:10 +02:00
if request.method == "PATCH":
patches.append(_json.loads(request.content)["recipeIngredient"])
2026-07-31 07:31:10 +02:00
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",
}],
})
2026-07-31 07:31:10 +02:00
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",
}]
2026-07-31 07:31:10 +02:00
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 TestNutritionPatch:
"""Mealie validates nothing here: it stores strings and renders them as fact."""
def _mealie(self, mock_mealie, recipe):
sent: list[dict] = []
def handler(request: httpx.Request) -> httpx.Response:
if request.method == "PATCH":
sent.append(json.loads(request.content))
return httpx.Response(200, json=recipe)
mock_mealie(handler)
return sent
def test_a_consistent_estimate_is_normalized_to_bare_numbers(self, mock_mealie):
sent = self._mealie(mock_mealie, {"id": "stable-id", "slug": "sas", "recipeServings": 4})
server.patch_recipe(
"sas",
{"nutrition": {
"calories": "459 kcal",
"proteinContent": 13,
"fatContent": "38 g",
"carbohydrateContent": 15.4,
}},
)
assert sent[0]["nutrition"] == {
"calories": "459",
"proteinContent": "13",
"fatContent": "38",
"carbohydrateContent": "15",
}
def test_energy_that_contradicts_the_macros_is_refused(self, mock_mealie):
sent = self._mealie(mock_mealie, {"id": "stable-id", "slug": "sas", "recipeServings": 4})
# 13 g protein + 38 g fat + 15 g carbohydrate is ~455 kcal, not 150.
with pytest.raises(ValueError, match="does not match the macros"):
server.patch_recipe(
"sas",
{"nutrition": {
"calories": 150,
"proteinContent": 13,
"fatContent": 38,
"carbohydrateContent": 15,
}},
)
assert sent == []
def test_nutrition_without_a_serving_count_is_refused(self, mock_mealie):
# The TikTok import left recipeServings 0 and only a "4-6 personer" string,
# so per-serving figures would have had nothing to divide by.
sent = self._mealie(mock_mealie, {"id": "stable-id", "slug": "sas", "recipeServings": 0})
with pytest.raises(ValueError, match="recipeServings"):
server.patch_recipe("sas", {"nutrition": {"calories": 459}})
assert sent == []
def test_servings_patched_in_the_same_call_satisfy_the_requirement(self, mock_mealie):
sent = self._mealie(mock_mealie, {"id": "stable-id", "slug": "sas", "recipeServings": 0})
server.patch_recipe("sas", {"recipeServings": 4, "nutrition": {"calories": 459}})
assert sent[0]["recipeServings"] == 4
assert sent[0]["nutrition"] == {"calories": "459"}
def test_a_misspelt_field_names_the_right_one_instead_of_vanishing(self, mock_mealie):
sent = self._mealie(mock_mealie, {"id": "stable-id", "slug": "sas", "recipeServings": 4})
with pytest.raises(ValueError, match="carbohydrateContent"):
server.patch_recipe("sas", {"nutrition": {"carbs": 15}})
assert sent == []
2026-07-31 07:31:10 +02:00
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)
2026-07-31 07:31:10 +02:00
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