mealie/tests/integration_tests/user_recipe_tests/test_recipe_suggestions.py

import random
from uuid import uuid4

import pytest
from fastapi.testclient import TestClient

from mealie.schema.recipe.recipe import Recipe
from mealie.schema.recipe.recipe_ingredient import IngredientFood, RecipeIngredient, SaveIngredientFood
from mealie.schema.recipe.recipe_settings import RecipeSettings
from mealie.schema.recipe.recipe_tool import RecipeToolOut, RecipeToolSave
from tests.utils import api_routes
from tests.utils.factories import random_int, random_string
from tests.utils.fixture_schemas import TestUser


def create_food(user: TestUser, on_hand: bool = False):
    return user.repos.ingredient_foods.create(
        SaveIngredientFood(id=uuid4(), name=random_string(), group_id=user.group_id, on_hand=on_hand)
    )


def create_tool(user: TestUser, on_hand: bool = False):
    return user.repos.tools.create(
        RecipeToolSave(id=uuid4(), name=random_string(), group_id=user.group_id, on_hand=on_hand)
    )


def create_recipe(
    user: TestUser,
    *,
    foods: list[IngredientFood] | None = None,
    tools: list[RecipeToolOut] | None = None,
    disable_amount: bool = False,
    **kwargs,
):
    if foods:
        ingredients = [RecipeIngredient(food_id=food.id, food=food) for food in foods]
    else:
        ingredients = []

    recipe = user.repos.recipes.create(
        Recipe(
            user_id=user.user_id,
            group_id=user.group_id,
            name=kwargs.pop("name", random_string()),
            recipe_ingredient=ingredients,
            tools=tools or [],
            settings=RecipeSettings(disable_amount=disable_amount),
            **kwargs,
        )
    )

    return recipe


@pytest.fixture(autouse=True)
def base_recipes(unique_user: TestUser, h2_user: TestUser):
    for user in [unique_user, h2_user]:
        for _ in range(10):
            create_recipe(
                user,
                foods=[create_food(user) for _ in range(random_int(5, 10))],
                tools=[create_tool(user) for _ in range(random_int(5, 10))],
            )


@pytest.mark.parametrize("filter_foods", [True, False])
@pytest.mark.parametrize("filter_tools", [True, False])
def test_suggestion_filter(api_client: TestClient, unique_user: TestUser, filter_foods: bool, filter_tools: bool):
    create_params: dict = {}
    api_params: dict = {"maxMissingFoods": 0, "maxMissingTools": 0, "limit": 10}
    if filter_foods:
        known_food = create_food(unique_user)
        create_params["foods"] = [known_food]
        api_params["foods"] = [str(known_food.id)]
    if filter_tools:
        known_tool = create_tool(unique_user)
        create_params["tools"] = [known_tool]
        api_params["tools"] = [str(known_tool.id)]

    recipes = [create_recipe(unique_user, **create_params) for _ in range(3)]
    try:
        expected_recipe_ids = {str(recipe.id) for recipe in recipes if recipe.id}
        response = api_client.get(api_routes.recipes_suggestions, params=api_params, headers=unique_user.token)
        response.raise_for_status()
        data = response.json()

        if not filter_foods and not filter_tools:
            assert len(data["items"]) == 10
        else:
            assert len(data["items"]) == 3
            for item in data["items"]:
                assert item["recipe"]["id"] in expected_recipe_ids
                assert item["missingFoods"] == []
                assert item["missingTools"] == []
    finally:
        for recipe in recipes:
            unique_user.repos.recipes.delete(recipe.slug)


def test_food_suggestion_filter_with_max(api_client: TestClient, unique_user: TestUser):
    food_1, food_2, food_3, food_4 = (create_food(unique_user) for _ in range(4))
    recipe_exact = create_recipe(unique_user, foods=[food_1])
    recipe_missing_one = create_recipe(unique_user, foods=[food_1, food_2])
    recipe_missing_two = create_recipe(unique_user, foods=[food_1, food_2, food_3])
    recipe_missing_three = create_recipe(unique_user, foods=[food_1, food_2, food_3, food_4])

    try:
        response = api_client.get(
            api_routes.recipes_suggestions,
            params={"maxMissingFoods": 1, "includeFoodsOnHand": False, "foods": [str(food_1.id)]},
            headers=unique_user.token,
        )
        response.raise_for_status()
        data = response.json()
        fetched_recipe_ids = {item["recipe"]["id"] for item in data["items"]}
        assert set(fetched_recipe_ids) == {str(recipe_exact.id), str(recipe_missing_one.id)}
        for item in data["items"]:
            missing_food_ids = [food["id"] for food in item["missingFoods"]]
            if item["recipe"]["id"] == str(recipe_exact.id):
                assert missing_food_ids == []
            else:
                assert missing_food_ids == [str(food_2.id)]

    finally:
        for recipe in [recipe_exact, recipe_missing_one, recipe_missing_two, recipe_missing_three]:
            unique_user.repos.recipes.delete(recipe.slug)


def test_tool_suggestion_filter_with_max(api_client: TestClient, unique_user: TestUser):
    tool_1, tool_2, tool_3, tool_4 = (create_tool(unique_user) for _ in range(4))
    recipe_exact = create_recipe(unique_user, tools=[tool_1])
    recipe_missing_one = create_recipe(unique_user, tools=[tool_1, tool_2])
    recipe_missing_two = create_recipe(unique_user, tools=[tool_1, tool_2, tool_3])
    recipe_missing_three = create_recipe(unique_user, tools=[tool_1, tool_2, tool_3, tool_4])

    try:
        response = api_client.get(
            api_routes.recipes_suggestions,
            params={"maxMissingTools": 1, "includeToolsOnHand": False, "tools": [str(tool_1.id)]},
            headers=unique_user.token,
        )
        response.raise_for_status()

        data = response.json()
        fetched_recipe_ids = {item["recipe"]["id"] for item in data["items"]}
        assert set(fetched_recipe_ids) == {str(recipe_exact.id), str(recipe_missing_one.id)}
        for item in data["items"]:
            missing_tool_ids = [tool["id"] for tool in item["missingTools"]]
            if item["recipe"]["id"] == str(recipe_exact.id):
                assert missing_tool_ids == []
            else:
                assert missing_tool_ids == [str(tool_2.id)]

    finally:
        for recipe in [recipe_exact, recipe_missing_one, recipe_missing_two, recipe_missing_three]:
            unique_user.repos.recipes.delete(recipe.slug)


def test_ignore_empty_food_filter(api_client: TestClient, unique_user: TestUser):
    known_tool = create_tool(unique_user)
    recipe = create_recipe(
        unique_user, foods=[create_food(unique_user) for _ in range(random_int(3, 5))], tools=[known_tool]
    )

    try:
        response = api_client.get(
            api_routes.recipes_suggestions,
            params={"maxMissingFoods": 0, "maxMissingTools": 0, "tools": [str(known_tool.id)]},
            headers=unique_user.token,
        )
        response.raise_for_status()

        data = response.json()
        assert len(data["items"]) == 1
        item = data["items"][0]
        assert item["recipe"]["id"] == str(recipe.id)
        assert item["missingFoods"] == []
        assert item["missingTools"] == []

    finally:
        unique_user.repos.recipes.delete(recipe.slug)


def test_ignore_empty_tool_filter(api_client: TestClient, unique_user: TestUser):
    known_food = create_food(unique_user)
    recipe = create_recipe(
        unique_user, foods=[known_food], tools=[create_tool(unique_user) for _ in range(random_int(3, 5))]
    )

    try:
        response = api_client.get(
            api_routes.recipes_suggestions,
            params={"maxMissingFoods": 0, "maxMissingTools": 0, "foods": [str(known_food.id)]},
            headers=unique_user.token,
        )
        response.raise_for_status()

        data = response.json()
        assert len(data["items"]) == 1
        item = data["items"][0]
        assert item["recipe"]["id"] == str(recipe.id)
        assert item["missingFoods"] == []
        assert item["missingTools"] == []

    finally:
        unique_user.repos.recipes.delete(recipe.slug)


@pytest.mark.parametrize("include_on_hand", [True, False])
def test_include_foods_on_hand(api_client: TestClient, unique_user: TestUser, include_on_hand: bool):
    on_hand_food = create_food(unique_user, on_hand=True)
    off_hand_food = create_food(unique_user, on_hand=False)
    recipe = create_recipe(unique_user, foods=[on_hand_food, off_hand_food])

    try:
        response = api_client.get(
            api_routes.recipes_suggestions,
            params={
                "maxMissingFoods": 0,
                "maxMissingTools": 0,
                "includeFoodsOnHand": include_on_hand,
                "foods": [str(off_hand_food.id)],
            },
            headers=unique_user.token,
        )
        response.raise_for_status()

        data = response.json()
        if not include_on_hand:
            assert len(data["items"]) == 0
        else:
            assert len(data["items"]) == 1
            item = data["items"][0]
            assert item["recipe"]["id"] == str(recipe.id)
            assert item["missingFoods"] == []

    finally:
        unique_user.repos.recipes.delete(recipe.slug)


@pytest.mark.parametrize("include_on_hand", [True, False])
def test_include_tools_on_hand(api_client: TestClient, unique_user: TestUser, include_on_hand: bool):
    on_hand_tool = create_tool(unique_user, on_hand=True)
    off_hand_tool = create_tool(unique_user, on_hand=False)
    recipe = create_recipe(unique_user, tools=[on_hand_tool, off_hand_tool])

    try:
        response = api_client.get(
            api_routes.recipes_suggestions,
            params={
                "maxMissingFoods": 0,
                "maxMissingTools": 0,
                "includeToolsOnHand": include_on_hand,
                "tools": [str(off_hand_tool.id)],
            },
            headers=unique_user.token,
        )
        response.raise_for_status()

        data = response.json()
        if not include_on_hand:
            assert len(data["items"]) == 0
        else:
            assert len(data["items"]) == 1
            item = data["items"][0]
            assert item["recipe"]["id"] == str(recipe.id)
            assert item["missingTools"] == []

    finally:
        unique_user.repos.recipes.delete(recipe.slug)


def test_exclude_recipes_with_no_foods(api_client: TestClient, unique_user: TestUser):
    known_food = create_food(unique_user)
    recipe_with_foods = create_recipe(unique_user, foods=[known_food])
    recipe_without_foods = create_recipe(unique_user, foods=[])

    try:
        response = api_client.get(
            api_routes.recipes_suggestions,
            params={"maxMissingFoods": 0, "maxMissingTools": 0, "foods": [str(known_food.id)]},
            headers=unique_user.token,
        )
        response.raise_for_status()

        data = response.json()
        assert {item["recipe"]["id"] for item in data["items"]} == {str(recipe_with_foods.id)}
        for item in data["items"]:
            assert item["missingFoods"] == []

    finally:
        for recipe in [recipe_with_foods, recipe_without_foods]:
            unique_user.repos.recipes.delete(recipe.slug)


def test_include_recipes_with_no_tools(api_client: TestClient, unique_user: TestUser):
    known_tool = create_tool(unique_user)
    recipe_with_tools = create_recipe(unique_user, tools=[known_tool])
    recipe_without_tools = create_recipe(unique_user, tools=[])

    try:
        response = api_client.get(
            api_routes.recipes_suggestions,
            params={"maxMissingFoods": 0, "maxMissingTools": 0, "tools": [str(known_tool.id)]},
            headers=unique_user.token,
        )
        response.raise_for_status()

        data = response.json()
        assert {item["recipe"]["id"] for item in data["items"]} == {
            str(recipe_with_tools.id),
            str(recipe_without_tools.id),
        }
        for item in data["items"]:
            assert item["missingTools"] == []

    finally:
        for recipe in [recipe_with_tools, recipe_without_tools]:
            unique_user.repos.recipes.delete(recipe.slug)


def test_ignore_recipes_with_ingredient_amounts_disabled_with_foods(api_client: TestClient, unique_user: TestUser):
    known_food = create_food(unique_user)
    recipe_with_amounts = create_recipe(unique_user, foods=[known_food])
    recipe_without_amounts = create_recipe(unique_user, foods=[known_food], disable_amount=True)

    try:
        response = api_client.get(
            api_routes.recipes_suggestions,
            params={"maxMissingFoods": 0, "maxMissingTools": 0, "foods": [str(known_food.id)]},
            headers=unique_user.token,
        )
        response.raise_for_status()

        data = response.json()
        assert {item["recipe"]["id"] for item in data["items"]} == {str(recipe_with_amounts.id)}
        for item in data["items"]:
            assert item["missingFoods"] == []

    finally:
        for recipe in [recipe_with_amounts, recipe_without_amounts]:
            unique_user.repos.recipes.delete(recipe.slug)


def test_include_recipes_with_ingredient_amounts_disabled_without_foods(api_client: TestClient, unique_user: TestUser):
    known_tool = create_tool(unique_user)
    recipe_with_amounts = create_recipe(unique_user, tools=[known_tool])
    recipe_without_amounts = create_recipe(unique_user, tools=[known_tool], disable_amount=True)

    try:
        response = api_client.get(
            api_routes.recipes_suggestions,
            params={
                "maxMissingFoods": 0,
                "maxMissingTools": 0,
                "includeFoodsOnHand": False,
                "tools": [str(known_tool.id)],
            },
            headers=unique_user.token,
        )
        response.raise_for_status()

        data = response.json()
        assert {item["recipe"]["id"] for item in data["items"]} == {
            str(recipe_with_amounts.id),
            str(recipe_without_amounts.id),
        }
        for item in data["items"]:
            assert item["missingFoods"] == []

    finally:
        for recipe in [recipe_with_amounts, recipe_without_amounts]:
            unique_user.repos.recipes.delete(recipe.slug)


def test_exclude_recipes_with_no_user_foods(api_client: TestClient, unique_user: TestUser):
    known_food = create_food(unique_user)
    food_on_hand = create_food(unique_user, on_hand=True)
    recipe_with_user_food = create_recipe(unique_user, foods=[known_food])
    recipe_with_on_hand_food = create_recipe(unique_user, foods=[food_on_hand])

    try:
        response = api_client.get(
            api_routes.recipes_suggestions,
            params={"maxMissingFoods": 10, "includeFoodsOnHand": True, "foods": [str(known_food.id)]},
            headers=unique_user.token,
        )
        response.raise_for_status()

        data = response.json()
        assert {item["recipe"]["id"] for item in data["items"]} == {str(recipe_with_user_food.id)}
        assert data["items"][0]["missingFoods"] == []

    finally:
        for recipe in [recipe_with_user_food, recipe_with_on_hand_food]:
            unique_user.repos.recipes.delete(recipe.slug)


def test_recipe_order(api_client: TestClient, unique_user: TestUser):
    user_food_1, user_food_2, other_food_1, other_food_2, other_food_3 = (create_food(unique_user) for _ in range(5))
    user_tool_1, other_tool_1, other_tool_2 = (create_tool(unique_user) for _ in range(3))
    food_on_hand = create_food(unique_user, on_hand=True)

    recipe_lambdas = [
        # No missing tools or foods
        (0, lambda: create_recipe(unique_user, tools=[user_tool_1], foods=[user_food_1])),
        # No missing tools, one missing food
        (1, lambda: create_recipe(unique_user, tools=[user_tool_1], foods=[user_food_1, other_food_1])),
        # One missing tool, no missing foods
        (2, lambda: create_recipe(unique_user, tools=[user_tool_1, other_tool_1], foods=[user_food_1])),
        # One missing tool, one missing food
        (3, lambda: create_recipe(unique_user, tools=[user_tool_1, other_tool_1], foods=[user_food_1, other_food_1])),
        # Two missing tools, two missing foods, two user foods
        (
            4,
            lambda: create_recipe(
                unique_user,
                tools=[user_tool_1, other_tool_1, other_tool_2],
                foods=[user_food_1, user_food_2, other_food_1, other_food_2],
            ),
        ),
        # Two missing tools, two missing foods, one user food
        (
            5,
            lambda: create_recipe(
                unique_user,
                tools=[user_tool_1, other_tool_1, other_tool_2],
                foods=[user_food_1, other_food_1, other_food_2],
            ),
        ),
        # Two missing tools, three missing foods, two user foods, don't include food on hand
        (
            6,
            lambda: create_recipe(
                unique_user,
                tools=[user_tool_1, other_tool_1, other_tool_2],
                foods=[user_food_1, user_food_2, other_food_1, other_food_2, other_food_3],
            ),
        ),
        # Two missing tools, three missing foods, one user food, include food on hand
        (
            7,
            lambda: create_recipe(
                unique_user,
                tools=[user_tool_1, other_tool_1, other_tool_2],
                foods=[food_on_hand, user_food_1, other_food_1, other_food_2, other_food_3],
            ),
        ),
    ]

    # create recipes in a random order
    random.shuffle(recipe_lambdas)
    recipe_tuples: list[tuple[int, Recipe]] = []
    for i, recipe_lambda in recipe_lambdas:
        recipe_tuples.append((i, recipe_lambda()))

    recipe_tuples.sort(key=lambda x: x[0])
    recipes = [recipe_tuple[1] for recipe_tuple in recipe_tuples]

    try:
        response = api_client.get(
            api_routes.recipes_suggestions,
            params={
                "maxMissingFoods": 3,
                "maxMissingTools": 3,
                "includeFoodsOnHand": True,
                "includeToolsOnHand": True,
                "limit": 10,
                "foods": [str(user_food_1.id), str(user_food_2.id)],
                "tools": [str(user_tool_1.id)],
            },
            headers=unique_user.token,
        )
        response.raise_for_status()

        data = response.json()
        assert len(data["items"]) == len(recipes)
        for i, (item, recipe) in enumerate(zip(data["items"], recipes, strict=True)):
            try:
                assert item["recipe"]["id"] == str(recipe.id)
            except AssertionError as e:
                raise AssertionError(f"Recipe in position {i} was incorrect") from e

    finally:
        for recipe in recipes:
            unique_user.repos.recipes.delete(recipe.slug)


def test_respect_user_sort(api_client: TestClient, unique_user: TestUser):
    known_food = create_food(unique_user)

    # Create recipes with names A, B, C, D out of order
    recipe_b = create_recipe(unique_user, foods=[known_food], name="B")
    recipe_c = create_recipe(unique_user, foods=[known_food, create_food(unique_user)], name="C")
    recipe_a = create_recipe(unique_user, foods=[known_food, create_food(unique_user)], name="A")
    recipe_d = create_recipe(unique_user, foods=[known_food, create_food(unique_user)], name="D")

    try:
        response = api_client.get(
            api_routes.recipes_suggestions,
            params={"maxMissingFoods": 1, "foods": [str(known_food.id)], "orderBy": "name", "orderDirection": "desc"},
            headers=unique_user.token,
        )
        response.raise_for_status()

        data = response.json()
        assert len(data["items"]) == 4

        # "B" should come first because it matches all foods, even though the user sort would put it last
        assert [item["recipe"]["name"] for item in data["items"]] == ["B", "D", "C", "A"]

    finally:
        for recipe in [recipe_a, recipe_b, recipe_c, recipe_d]:
            unique_user.repos.recipes.delete(recipe.slug)


def test_limit_param(api_client: TestClient, unique_user: TestUser):
    known_food = create_food(unique_user)
    limit = random_int(12, 20)
    recipes = [create_recipe(unique_user, foods=[known_food]) for _ in range(limit)]

    try:
        response = api_client.get(
            api_routes.recipes_suggestions,
            params={"maxMissingFoods": 0, "foods": [str(known_food.id)], "limit": limit},
            headers=unique_user.token,
        )
        response.raise_for_status()
        assert len(response.json()["items"]) == limit

    finally:
        for recipe in recipes:
            unique_user.repos.recipes.delete(recipe.slug)


def test_query_filter(api_client: TestClient, unique_user: TestUser):
    known_food = create_food(unique_user)
    recipes_with_prefix = [
        create_recipe(unique_user, foods=[known_food], name=f"MY_PREFIX{random_string()}") for _ in range(10)
    ]
    recipes_without_prefix = [
        create_recipe(unique_user, foods=[known_food], name=f"MY_OTHER_PREFIX{random_string()}") for _ in range(10)
    ]

    try:
        response = api_client.get(
            api_routes.recipes_suggestions,
            params={"maxMissingFoods": 0, "foods": [str(known_food.id)], "queryFilter": 'name LIKE "MY_PREFIX%"'},
            headers=unique_user.token,
        )
        response.raise_for_status()
        assert len(response.json()["items"]) == len(recipes_with_prefix)
        assert {item["recipe"]["id"] for item in response.json()["items"]} == {
            str(recipe.id) for recipe in recipes_with_prefix
        }

    finally:
        for recipe in recipes_with_prefix + recipes_without_prefix:
            unique_user.repos.recipes.delete(recipe.slug)


def test_include_cross_household_recipes(api_client: TestClient, unique_user: TestUser, h2_user: TestUser):
    known_food = create_food(unique_user)
    recipe = create_recipe(unique_user, foods=[known_food])
    other_recipe = create_recipe(h2_user, foods=[known_food])

    try:
        response = api_client.get(
            api_routes.recipes_suggestions,
            params={"maxMissingFoods": 0, "foods": [str(known_food.id)], "includeCrossHousehold": True},
            headers=h2_user.token,
        )
        response.raise_for_status()
        data = response.json()
        assert len(data["items"]) == 2
        assert {item["recipe"]["id"] for item in data["items"]} == {str(recipe.id), str(other_recipe.id)}

    finally:
        unique_user.repos.recipes.delete(recipe.slug)
        h2_user.repos.recipes.delete(other_recipe.slug)