nextcloud-mcp-server/tests/server/test_cookbook_mcp.py

import json
import logging
import uuid

import anyio
import pytest
from mcp import ClientSession

from nextcloud_mcp_server.client import NextcloudClient

logger = logging.getLogger(__name__)
pytestmark = pytest.mark.integration


async def test_mcp_cookbook_create_and_read_recipe(
    nc_mcp_client: ClientSession, nc_client: NextcloudClient
):
    """Test creating and reading a recipe via MCP tools with verification via NextcloudClient."""

    unique_suffix = uuid.uuid4().hex[:8]
    recipe_name = f"MCP Test Recipe {unique_suffix}"
    recipe_data = {
        "name": recipe_name,
        "description": "A test recipe created via MCP tools",
        "recipeIngredient": ["100g flour", "2 eggs", "200ml milk"],
        "recipeInstructions": ["Mix ingredients", "Cook for 20 minutes", "Serve hot"],
        "recipeCategory": "MCPTesting",
        "keywords": f"mcp,testing,{unique_suffix}",
        "recipeYield": 4,
        "prepTime": "PT15M",
        "cookTime": "PT20M",
        "totalTime": "PT35M",
    }

    created_recipe_id = None

    try:
        # 1. Create recipe via MCP
        logger.info(f"Creating recipe via MCP: {recipe_name}")
        create_result = await nc_mcp_client.call_tool(
            "nc_cookbook_create_recipe",
            {
                "name": recipe_name,
                "description": recipe_data["description"],
                "ingredients": recipe_data["recipeIngredient"],
                "instructions": recipe_data["recipeInstructions"],
                "category": recipe_data["recipeCategory"],
                "keywords": recipe_data["keywords"],
                "recipe_yield": recipe_data["recipeYield"],
                "prep_time": recipe_data["prepTime"],
                "cook_time": recipe_data["cookTime"],
                "total_time": recipe_data["totalTime"],
            },
        )

        assert create_result.isError is False, (
            f"MCP recipe creation failed: {create_result.content}"
        )

        create_response = json.loads(create_result.content[0].text)
        created_recipe_id = create_response["id"]
        logger.info(f"Recipe created via MCP with ID: {created_recipe_id}")

        # 2. Verify creation via direct NextcloudClient
        direct_recipe = await nc_client.cookbook.get_recipe(created_recipe_id)
        assert direct_recipe["name"] == recipe_name
        assert direct_recipe["description"] == "A test recipe created via MCP tools"
        assert len(direct_recipe["recipeIngredient"]) == 3
        assert len(direct_recipe["recipeInstructions"]) == 3
        assert direct_recipe["recipeCategory"] == "MCPTesting"

        # 3. Read recipe via MCP
        logger.info(f"Reading recipe via MCP: {created_recipe_id}")
        read_result = await nc_mcp_client.call_tool(
            "nc_cookbook_get_recipe", {"recipe_id": created_recipe_id}
        )

        assert read_result.isError is False, (
            f"MCP recipe read failed: {read_result.content}"
        )

        read_recipe = json.loads(read_result.content[0].text)
        assert read_recipe["name"] == recipe_name
        assert read_recipe["description"] == "A test recipe created via MCP tools"
        assert len(read_recipe["recipeIngredient"]) == 3

        logger.info(f"Successfully verified recipe {created_recipe_id} via MCP")

    finally:
        # Cleanup
        if created_recipe_id is not None:
            try:
                await nc_client.cookbook.delete_recipe(created_recipe_id)
                logger.info(f"Cleaned up recipe {created_recipe_id}")
            except Exception as e:
                logger.warning(f"Failed to cleanup recipe: {e}")


async def test_mcp_cookbook_update_recipe(
    nc_mcp_client: ClientSession, nc_client: NextcloudClient
):
    """Test updating a recipe via MCP tools."""

    unique_suffix = uuid.uuid4().hex[:8]
    recipe_name = f"MCP Update Test {unique_suffix}"
    recipe_data = {
        "name": recipe_name,
        "description": "Original description",
        "recipeIngredient": ["100g flour"],
        "recipeInstructions": ["Mix ingredients"],
        "recipeCategory": "Original",
    }

    created_recipe_id = None

    try:
        # 1. Create recipe via direct client
        logger.info(f"Creating recipe for update test: {recipe_name}")
        created_recipe_id = await nc_client.cookbook.create_recipe(recipe_data)

        # 2. Update recipe via MCP (tool handles fetching current recipe internally)
        logger.info(f"Updating recipe via MCP: {created_recipe_id}")
        update_result = await nc_mcp_client.call_tool(
            "nc_cookbook_update_recipe",
            {
                "recipe_id": created_recipe_id,
                "description": "Updated via MCP",
                "ingredients": ["100g flour", "2 eggs"],
                "instructions": ["Mix ingredients", "Cook"],
                "category": "Updated",
            },
        )

        assert update_result.isError is False, (
            f"MCP recipe update failed: {update_result.content}"
        )

        # 4. Verify update via direct NextcloudClient
        await anyio.sleep(1)  # Allow propagation
        updated_recipe = await nc_client.cookbook.get_recipe(created_recipe_id)
        assert updated_recipe["description"] == "Updated via MCP"
        assert len(updated_recipe["recipeIngredient"]) == 2
        assert len(updated_recipe["recipeInstructions"]) == 2
        assert updated_recipe["recipeCategory"] == "Updated"

        logger.info(f"Successfully updated recipe {created_recipe_id} via MCP")

    finally:
        # Cleanup
        if created_recipe_id is not None:
            try:
                await nc_client.cookbook.delete_recipe(created_recipe_id)
                logger.info(f"Cleaned up recipe {created_recipe_id}")
            except Exception as e:
                logger.warning(f"Failed to cleanup recipe: {e}")


async def test_mcp_cookbook_delete_recipe(
    nc_mcp_client: ClientSession, nc_client: NextcloudClient
):
    """Test deleting a recipe via MCP tools."""

    unique_suffix = uuid.uuid4().hex[:8]
    recipe_name = f"MCP Delete Test {unique_suffix}"
    recipe_data = {
        "name": recipe_name,
        "description": "Recipe to be deleted",
        "recipeIngredient": ["test"],
        "recipeInstructions": ["test"],
    }

    created_recipe_id = None

    try:
        # 1. Create recipe via direct client
        logger.info(f"Creating recipe for delete test: {recipe_name}")
        created_recipe_id = await nc_client.cookbook.create_recipe(recipe_data)

        # 2. Delete recipe via MCP
        logger.info(f"Deleting recipe via MCP: {created_recipe_id}")
        delete_result = await nc_mcp_client.call_tool(
            "nc_cookbook_delete_recipe", {"recipe_id": created_recipe_id}
        )

        assert delete_result.isError is False, (
            f"MCP recipe deletion failed: {delete_result.content}"
        )

        # 3. Verify deletion via direct NextcloudClient
        try:
            await nc_client.cookbook.get_recipe(created_recipe_id)
            pytest.fail("Recipe should have been deleted but was still found")
        except Exception:
            # Expected - recipe should be deleted
            logger.info(f"Successfully verified recipe {created_recipe_id} was deleted")
            created_recipe_id = None  # Mark as cleaned up

    finally:
        # Cleanup in case of test failure
        if created_recipe_id is not None:
            try:
                await nc_client.cookbook.delete_recipe(created_recipe_id)
                logger.info(f"Cleaned up recipe {created_recipe_id}")
            except Exception as e:
                logger.warning(f"Failed to cleanup recipe: {e}")


async def test_mcp_cookbook_import_recipe_from_url(
    nc_mcp_client: ClientSession,
    nc_client: NextcloudClient,
):
    """Test importing a recipe from a URL via MCP tools.

    This is the key feature test - importing recipes from URLs using schema.org metadata.
    Uses an nginx container to serve reliable, controlled test data.
    """
    # Use the nginx container hostname within the Docker network
    test_url = "http://recipes/black-pepper-tofu"

    created_recipe_id = None

    try:
        # 1. Import recipe via MCP
        logger.info(f"Importing recipe from nginx container via MCP: {test_url}")
        import_result = await nc_mcp_client.call_tool(
            "nc_cookbook_import_recipe", {"url": test_url}
        )

        assert import_result.isError is False, (
            f"MCP recipe import failed: {import_result.content}"
        )

        import_response = json.loads(import_result.content[0].text)
        created_recipe_id = int(import_response["recipe_id"])
        imported_recipe = import_response["recipe"]

        logger.info(f"Successfully imported recipe via MCP: {imported_recipe['name']}")

        # 2. Verify basic recipe structure
        assert imported_recipe["name"] == "Black Pepper Tofu"
        assert imported_recipe.get("description")
        assert len(imported_recipe.get("recipeIngredient", [])) > 0
        assert len(imported_recipe.get("recipeInstructions", [])) > 0
        assert imported_recipe.get("recipeCategory") == "Main Course"
        assert "tofu" in imported_recipe.get("keywords", "").lower()

        # 3. Verify we can read it back via direct NextcloudClient
        retrieved = await nc_client.cookbook.get_recipe(created_recipe_id)
        assert retrieved["name"] == imported_recipe["name"]
        logger.info(f"Verified imported recipe ID: {created_recipe_id}")

    finally:
        # Cleanup
        if created_recipe_id is not None:
            try:
                await nc_client.cookbook.delete_recipe(created_recipe_id)
                logger.info(f"Cleaned up imported recipe {created_recipe_id}")
            except Exception as e:
                logger.warning(f"Failed to cleanup imported recipe: {e}")


async def test_mcp_cookbook_search_recipes(
    nc_mcp_client: ClientSession, nc_client: NextcloudClient
):
    """Test searching recipes via MCP tools."""

    unique_keyword = f"mcptestkeyword{uuid.uuid4().hex[:8]}"
    recipe_name = f"MCP Search Test {uuid.uuid4().hex[:8]}"
    recipe_data = {
        "name": recipe_name,
        "description": f"Recipe for testing MCP search with {unique_keyword}",
        "keywords": unique_keyword,
        "recipeIngredient": ["test ingredient"],
        "recipeInstructions": ["test instruction"],
    }

    created_recipe_id = None

    try:
        # 1. Create recipe via direct client
        logger.info(f"Creating recipe for search test with keyword: {unique_keyword}")
        created_recipe_id = await nc_client.cookbook.create_recipe(recipe_data)

        # 2. Allow time for indexing
        await anyio.sleep(2)

        # 3. Search for the recipe via MCP
        logger.info(f"Searching for recipes via MCP with keyword: {unique_keyword}")
        search_result = await nc_mcp_client.call_tool(
            "nc_cookbook_search_recipes", {"query": unique_keyword}
        )

        assert search_result.isError is False, (
            f"MCP recipe search failed: {search_result.content}"
        )

        search_response = json.loads(search_result.content[0].text)
        search_results = search_response["recipes"]

        assert isinstance(search_results, list)
        assert len(search_results) > 0

        # 4. Verify our recipe is in the results
        found = any(str(r.get("id")) == str(created_recipe_id) for r in search_results)
        assert found, f"Recipe {created_recipe_id} not found in search results"
        logger.info(
            f"Successfully found recipe {created_recipe_id} in MCP search results"
        )

    finally:
        # Cleanup
        if created_recipe_id is not None:
            try:
                await nc_client.cookbook.delete_recipe(created_recipe_id)
                logger.info(f"Cleaned up recipe {created_recipe_id}")
            except Exception as e:
                logger.warning(f"Failed to cleanup recipe: {e}")


async def test_mcp_cookbook_list_recipes(
    nc_mcp_client: ClientSession, nc_client: NextcloudClient
):
    """Test listing all recipes via MCP tools."""

    logger.info("Listing all recipes via MCP")
    list_result = await nc_mcp_client.call_tool("nc_cookbook_list_recipes", {})

    assert list_result.isError is False, (
        f"MCP list recipes failed: {list_result.content}"
    )

    list_response = json.loads(list_result.content[0].text)
    recipes = list_response["recipes"]

    assert isinstance(recipes, list)
    logger.info(f"Found {len(recipes)} recipes via MCP")


async def test_mcp_cookbook_categories_workflow(
    nc_mcp_client: ClientSession, nc_client: NextcloudClient
):
    """Test category listing and filtering via MCP tools."""

    unique_category = f"MCPTestCategory{uuid.uuid4().hex[:8]}"
    recipe_name = f"MCP Category Test {uuid.uuid4().hex[:8]}"
    recipe_data = {
        "name": recipe_name,
        "recipeCategory": unique_category,
        "recipeIngredient": ["test"],
        "recipeInstructions": ["test"],
    }

    created_recipe_id = None

    try:
        # 1. Create recipe in test category
        logger.info(f"Creating recipe in category: {unique_category}")
        created_recipe_id = await nc_client.cookbook.create_recipe(recipe_data)

        # 2. Allow time for indexing
        await anyio.sleep(2)

        # 3. List categories via MCP
        logger.info("Listing categories via MCP")
        categories_result = await nc_mcp_client.call_tool(
            "nc_cookbook_list_categories", {}
        )

        assert categories_result.isError is False, (
            f"MCP list categories failed: {categories_result.content}"
        )

        categories_response = json.loads(categories_result.content[0].text)
        categories = categories_response["categories"]

        assert isinstance(categories, list)
        logger.info(f"Found {len(categories)} categories via MCP")

        # 4. Get recipes in this category via MCP
        logger.info(f"Getting recipes in category via MCP: {unique_category}")
        category_recipes_result = await nc_mcp_client.call_tool(
            "nc_cookbook_get_recipes_in_category", {"category": unique_category}
        )

        assert category_recipes_result.isError is False, (
            f"MCP get recipes in category failed: {category_recipes_result.content}"
        )

        category_recipes_response = json.loads(category_recipes_result.content[0].text)
        recipes_in_category = category_recipes_response["recipes"]

        assert isinstance(recipes_in_category, list)
        assert len(recipes_in_category) > 0

        # 5. Verify our recipe is in the results
        found = any(
            str(r.get("id")) == str(created_recipe_id) for r in recipes_in_category
        )
        assert found, (
            f"Recipe {created_recipe_id} not found in category {unique_category}"
        )
        logger.info(f"Successfully found recipe in category {unique_category} via MCP")

    finally:
        # Cleanup
        if created_recipe_id is not None:
            try:
                await nc_client.cookbook.delete_recipe(created_recipe_id)
                logger.info(f"Cleaned up recipe {created_recipe_id}")
            except Exception as e:
                logger.warning(f"Failed to cleanup recipe: {e}")


async def test_mcp_cookbook_keywords_workflow(
    nc_mcp_client: ClientSession, nc_client: NextcloudClient
):
    """Test keyword listing and filtering via MCP tools."""

    unique_keyword = f"mcptesttag{uuid.uuid4().hex[:8]}"
    recipe_name = f"MCP Keyword Test {uuid.uuid4().hex[:8]}"
    recipe_data = {
        "name": recipe_name,
        "keywords": f"{unique_keyword},mcptesting",
        "recipeIngredient": ["test"],
        "recipeInstructions": ["test"],
    }

    created_recipe_id = None

    try:
        # 1. Create recipe with test keywords
        logger.info(f"Creating recipe with keyword: {unique_keyword}")
        created_recipe_id = await nc_client.cookbook.create_recipe(recipe_data)

        # 2. Allow extra time for indexing and trigger reindex
        await anyio.sleep(3)
        await nc_client.cookbook.reindex()
        await anyio.sleep(2)

        # 3. List keywords via MCP
        logger.info("Listing keywords via MCP")
        keywords_result = await nc_mcp_client.call_tool("nc_cookbook_list_keywords", {})

        assert keywords_result.isError is False, (
            f"MCP list keywords failed: {keywords_result.content}"
        )

        keywords_response = json.loads(keywords_result.content[0].text)
        keywords = keywords_response["keywords"]

        assert isinstance(keywords, list)
        logger.info(f"Found {len(keywords)} keywords via MCP")

        # 4. Get recipes with this keyword via MCP
        logger.info(f"Getting recipes with keyword via MCP: {unique_keyword}")
        keyword_recipes_result = await nc_mcp_client.call_tool(
            "nc_cookbook_get_recipes_with_keywords", {"keywords": [unique_keyword]}
        )

        assert keyword_recipes_result.isError is False, (
            f"MCP get recipes with keywords failed: {keyword_recipes_result.content}"
        )

        keyword_recipes_response = json.loads(keyword_recipes_result.content[0].text)
        recipes_with_keywords = keyword_recipes_response["recipes"]

        assert isinstance(recipes_with_keywords, list)

        # Keyword filtering might not find recipes immediately due to indexing
        if len(recipes_with_keywords) > 0:
            # Verify our recipe is in the results if any are found
            found = any(
                str(r.get("id")) == str(created_recipe_id)
                for r in recipes_with_keywords
            )
            if found:
                logger.info(
                    f"Successfully found recipe with keyword {unique_keyword} via MCP"
                )
            else:
                logger.warning(
                    f"Recipe {created_recipe_id} not in keyword results via MCP, but other recipes found"
                )
        else:
            logger.warning(
                f"No recipes found with keyword {unique_keyword} via MCP - may be indexing delay"
            )

    finally:
        # Cleanup
        if created_recipe_id is not None:
            try:
                await nc_client.cookbook.delete_recipe(created_recipe_id)
                logger.info(f"Cleaned up recipe {created_recipe_id}")
            except Exception as e:
                logger.warning(f"Failed to cleanup recipe: {e}")


async def test_mcp_cookbook_config_and_version(
    nc_mcp_client: ClientSession, nc_client: NextcloudClient
):
    """Test getting Cookbook configuration and version via MCP resources."""

    # 1. Get version via MCP resource
    logger.info("Getting Cookbook version via MCP resource")
    version_result = await nc_mcp_client.read_resource("cookbook://version")

    assert len(version_result.contents) > 0
    version_response = json.loads(version_result.contents[0].text)
    assert "cookbook_version" in version_response
    assert "api_version" in version_response
    logger.info(f"Cookbook version from MCP: {version_response}")

    # 2. Verify version via direct NextcloudClient
    direct_version = await nc_client.cookbook.get_version()
    assert direct_version["cookbook_version"] == version_response["cookbook_version"]
    assert (
        direct_version["api_version"]["epoch"]
        == version_response["api_version"]["epoch"]
    )

    # 3. Get config via MCP resource
    logger.info("Getting Cookbook config via MCP resource")
    config_result = await nc_mcp_client.read_resource("cookbook://config")

    assert len(config_result.contents) > 0
    config_response = json.loads(config_result.contents[0].text)
    assert isinstance(config_response, dict)
    logger.info(f"Cookbook config from MCP: {config_response}")

    # 4. Verify config via direct NextcloudClient
    direct_config = await nc_client.cookbook.get_config()
    # Both should be dicts - exact match may vary based on config
    assert isinstance(config_response, dict)
    assert isinstance(direct_config, dict)

    logger.info("Successfully verified Cookbook version and config via MCP")


async def test_mcp_cookbook_reindex(
    nc_mcp_client: ClientSession, nc_client: NextcloudClient
):
    """Test triggering a recipe reindex via MCP tools."""

    logger.info("Triggering recipe reindex via MCP")
    reindex_result = await nc_mcp_client.call_tool("nc_cookbook_reindex", {})

    assert reindex_result.isError is False, (
        f"MCP reindex failed: {reindex_result.content}"
    )

    reindex_response = json.loads(reindex_result.content[0].text)
    assert isinstance(reindex_response["message"], str)
    logger.info(f"Reindex result from MCP: {reindex_response['message']}")