mealie-mcp/tests/test_server_contracts.py

"""Boundary contracts for the verified Mealie instance quirks."""
from __future__ import annotations

import asyncio

import httpx
import pytest
from mcp.server.mcpserver import MCPServer

from mealie_mcp import server


def response(status: int, text: str = "") -> httpx.Response:
    return httpx.Response(status, text=text, request=httpx.Request("GET", "https://example.test"))


class TestAuthenticationFailures:
    def test_401_is_an_auth_problem(self):
        with pytest.raises(server.MealieAuthError):
            server._raise_for_status(response(401))

    def test_cloudflare_1010_is_not_reported_as_bad_token(self):
        with pytest.raises(server.MealieBlockedError):
            server._raise_for_status(response(403, "error code: 1010"))


def test_read_only_server_removes_all_mutating_tools():
    test_server = MCPServer("test")

    @test_server.tool()
    def search_recipes(query: str) -> list[str]:
        return [query]

    @test_server.tool()
    def patch_recipe(value: str) -> str:
        return value

    @test_server.tool()
    def future_mutating_tool(value: str) -> str:
        return value

    server._configure_read_only(test_server, frozenset({"search_recipes"}))

    tool_names = {tool.name for tool in asyncio.run(test_server.list_tools())}

    assert tool_names == {"search_recipes"}


def test_read_only_tool_allowlist_is_explicit():
    assert server.READ_ONLY_TOOL_NAMES == {
        "check_auth",
        "find_by_source_url",
        "get_recipe",
        "list_organizers",
        "resolve_foods",
        "scale_ingredients",
        "search_recipes",
        "suggest_recipes",
        "verify_recipe",
    }


def test_import_profile_adds_the_pipeline_but_never_delete():
    assert server.IMPORT_TOOL_NAMES == server.READ_ONLY_TOOL_NAMES | {
        "import_recipe_url",
        "import_recipe_text",
        "import_recipe_image",
        "patch_recipe",
        "parse_ingredients",
        "set_cover_image",
    }
    assert "delete_recipe" not in server.IMPORT_TOOL_NAMES
    assert "shopping_list_add" not in server.IMPORT_TOOL_NAMES


class TestUrlImportResponse:
    def test_slug_only_import_response_is_resolved_before_reporting(self, monkeypatch):
        recipe = {"slug": "lax-med-citron", "name": "Lax med citron", "recipeIngredient": []}
        calls: list[tuple[str, str]] = []

        def request(method, path, **kwargs):
            calls.append((method, path))
            assert kwargs["json"] == {"url": "https://example.test/lax"}
            return "lax-med-citron"

        monkeypatch.setattr(server, "find_by_source_url", lambda _url: [])
        monkeypatch.setattr(server, "_request", request)
        monkeypatch.setattr(server, "get_recipe", lambda slug: recipe if slug == "lax-med-citron" else None)

        result = server.import_recipe_url("https://example.test/lax")

        assert calls == [("POST", "/api/recipes/create/url")]
        assert result["imported"] is True
        assert result["recipe"] == recipe
        assert result["report"]["slug"] == "lax-med-citron"

    def test_duplicate_source_url_stops_before_creating(self, monkeypatch):
        monkeypatch.setattr(
            server,
            "find_by_source_url",
            lambda _url: [{"slug": "redan-finns", "name": "Redan finns"}],
        )
        monkeypatch.setattr(server, "_request", lambda *_args, **_kwargs: pytest.fail("must not import"))

        result = server.import_recipe_url("https://example.test/recept")

        assert result == {
            "imported": False,
            "reason": "duplicate",
            "existing": [{"slug": "redan-finns", "name": "Redan finns"}],
        }


class TestPatchContract:
    def test_patch_recipe_reads_back_the_final_object(self, monkeypatch):
        seen = []
        before = {"id": "stable-id", "slug": "soppa", "name": "Soppa"}
        updated = {
            "id": "stable-id",
            "slug": "ny-soppa",
            "name": "Ny soppa",
            "tags": [{"name": "📅 Vardag"}],
        }

        def request(method, path, **kwargs):
            seen.append((method, path, kwargs))
            return None

        monkeypatch.setattr(server, "_request", request)
        monkeypatch.setattr(
            server,
            "get_recipe",
            lambda key: before if key == "soppa" else updated if key == "stable-id" else None,
        )

        assert server.patch_recipe("soppa", {"tags": updated["tags"]}) == updated
        assert seen == [("PATCH", "/api/recipes/soppa", {"json": {"tags": updated["tags"]}})]