Meal-Planner/backend/tests/test_starter_recipe_catalog.py

"""Offline and PostgreSQL coverage for the bundled starter recipe loader."""
from __future__ import annotations

import copy
import threading
from datetime import date
from queue import Empty, Queue

import pytest
from sqlalchemy import event
from sqlalchemy.orm import sessionmaker

from app.models import (
    Ingredient,
    MealPlan,
    MealPlanItem,
    MealType,
    Recipe,
    RecipeCatalogStatus,
)
from app.services.starter_recipe_catalog import (
    DEFAULT_CATALOG_PATH,
    MAX_CALORIES_PER_SERVING,
    STARTER_CATALOG_VERSION,
    STARTER_DISCOVERY_REASON,
    STARTER_RECIPE_COUNT,
    StarterRecipeCatalogError,
    load_bundled_starter_catalog,
    load_starter_catalog,
    read_starter_catalog,
    validate_starter_catalog,
)


@pytest.fixture(name="db")
def starter_catalog_db(_engine):
    """Use a commit-aware session and remove only rows created by each loader test."""
    if _engine is None:
        pytest.skip("Postgres not reachable")
    Session = sessionmaker(
        bind=_engine,
        autocommit=False,
        autoflush=False,
        expire_on_commit=False,
    )
    baseline = Session()
    try:
        recipe_ids = {recipe.id for recipe in baseline.query(Recipe).all()}
        ingredient_ids = {ingredient.id for ingredient in baseline.query(Ingredient).all()}
        meal_plan_ids = {meal_plan.id for meal_plan in baseline.query(MealPlan).all()}
        meal_plan_item_ids = {
            meal_plan_item.id for meal_plan_item in baseline.query(MealPlanItem).all()
        }
    finally:
        baseline.rollback()
        baseline.close()

    session = Session()
    try:
        yield session
    finally:
        session.rollback()
        session.close()
        cleanup = Session()
        try:
            for meal_plan_item in cleanup.query(MealPlanItem).all():
                if meal_plan_item.id not in meal_plan_item_ids:
                    cleanup.delete(meal_plan_item)
            cleanup.flush()
            for meal_plan in cleanup.query(MealPlan).all():
                if meal_plan.id not in meal_plan_ids:
                    cleanup.delete(meal_plan)
            cleanup.flush()
            for recipe in cleanup.query(Recipe).all():
                if recipe.id not in recipe_ids:
                    cleanup.delete(recipe)
            cleanup.flush()
            for ingredient in cleanup.query(Ingredient).all():
                if ingredient.id not in ingredient_ids:
                    cleanup.delete(ingredient)
            cleanup.commit()
        finally:
            cleanup.rollback()
            cleanup.close()


def _catalog() -> dict[str, object]:
    return dict(read_starter_catalog(DEFAULT_CATALOG_PATH))


def _catalog_recipe(catalog: dict[str, object], name: str) -> dict[str, object]:
    return next(
        recipe for recipe in catalog["recipes"] if recipe["name"] == name  # type: ignore[index]
    )


def _prior_catalog() -> dict[str, object]:
    catalog = copy.deepcopy(_catalog())
    for recipe in catalog["recipes"]:  # type: ignore[index]
        recipe["description"] = "A prior bundled formulation before the version two update."
        recipe["calories_per_serving"] = (
            recipe["calories_per_serving"] - 1
            if recipe["calories_per_serving"] == MAX_CALORIES_PER_SERVING
            else recipe["calories_per_serving"] + 1
        )
        recipe["ingredients"][0]["quantity"] += 1
        recipe["instructions"][0] = (
            "Prepare the prior bundled formulation with measured ingredients and careful cooking."
        )
    return catalog


def _ingredient_values(db, recipe: Recipe) -> list[tuple[str, object, object]]:
    ingredients = {
        str(ingredient.id): ingredient.name_lower
        for ingredient in db.query(Ingredient)
        .filter(Ingredient.id.in_([item["ingredient_id"] for item in recipe.ingredients]))
        .all()
    }
    return [
        (ingredients[item["ingredient_id"]], item["qty"], item["unit"])
        for item in recipe.ingredients
    ]


def _unrelated_recipes() -> tuple[Recipe, Recipe]:
    approved = Recipe(
        name="Rigatoni Pasta with Ricotta in Tomato Cream Sauce",
        description="An already approved Wikibooks provenance row.",
        servings=4,
        ingredients=[],
        instructions=[
            "Keep the approved provenance recipe unchanged during catalog loading."
        ],
        is_manually_added=False,
        external_source="wikibooks",
        external_id="approved-rigatoni-provenance-row",
        source_url="https://en.wikibooks.org/wiki/Cookbook:Rigatoni",
        discovery_reason="Approved import from revision 1",
        catalog_status=RecipeCatalogStatus.ACTIVE,
        catalog_findings=[],
    )
    legacy = Recipe(
        name="Legacy quarantined recipe remains unchanged",
        description="A historical row outside the bundled catalog.",
        servings=2,
        ingredients=[],
        instructions=[
            "Keep historical links to this quarantined recipe available without changes."
        ],
        is_manually_added=False,
        catalog_status=RecipeCatalogStatus.QUARANTINED,
        catalog_findings=[{"code": "legacy", "severity": "error"}],
    )
    return approved, legacy


def test_default_catalog_path_is_validated_as_the_complete_catalog():
    assert DEFAULT_CATALOG_PATH.is_file()
    catalog = _catalog()

    recipes = validate_starter_catalog(catalog)

    assert catalog["version"] == STARTER_CATALOG_VERSION
    assert len(recipes) == STARTER_RECIPE_COUNT
    assert len({recipe["name"] for recipe in recipes}) == STARTER_RECIPE_COUNT
    assert max(recipe["calories_per_serving"] for recipe in recipes) <= MAX_CALORIES_PER_SERVING


@pytest.mark.parametrize("version", [1, 3])
def test_validation_rejects_catalog_versions_other_than_two(version):
    catalog = copy.deepcopy(_catalog())
    catalog["version"] = version

    with pytest.raises(StarterRecipeCatalogError, match="Catalog version must be 2"):
        validate_starter_catalog(catalog)


def test_validation_rejects_calories_per_serving_over_limit():
    catalog = copy.deepcopy(_catalog())
    catalog["recipes"][0]["calories_per_serving"] = MAX_CALORIES_PER_SERVING + 1  # type: ignore[index]

    with pytest.raises(
        StarterRecipeCatalogError,
        match="calories_per_serving must be an integer between 1 and 500",
    ):
        validate_starter_catalog(catalog)


def test_validation_rejects_partial_catalog_before_database_access():
    catalog = copy.deepcopy(_catalog())
    catalog["recipes"] = catalog["recipes"][:-1]  # type: ignore[index]

    class NoDatabase:
        def __getattr__(self, name):
            raise AssertionError(f"Database access attempted through {name}")

    with pytest.raises(StarterRecipeCatalogError, match="exactly 50"):
        load_starter_catalog(NoDatabase(), catalog)  # type: ignore[arg-type]


def test_nonstandard_json_number_is_a_catalog_error(tmp_path):
    invalid_catalog = tmp_path / "invalid_catalog.json"
    invalid_catalog.write_text('{"version": NaN, "recipes": []}', encoding="utf-8")

    with pytest.raises(StarterRecipeCatalogError, match="not valid JSON"):
        read_starter_catalog(invalid_catalog)


@pytest.mark.requires_postgres
def test_dry_run_validates_actual_catalog_without_writes(db):
    before = (
        db.query(Recipe).count(),
        db.query(Ingredient).count(),
        db.query(Recipe).filter(Recipe.discovery_reason == STARTER_DISCOVERY_REASON).count(),
    )
    db.rollback()

    summary = load_bundled_starter_catalog(db, dry_run=True)

    assert summary["dry_run"] is True
    assert summary["validated_recipes"] == STARTER_RECIPE_COUNT
    assert summary["created_recipes"] == STARTER_RECIPE_COUNT
    after = (
        db.query(Recipe).count(),
        db.query(Ingredient).count(),
        db.query(Recipe).filter(Recipe.discovery_reason == STARTER_DISCOVERY_REASON).count(),
    )
    assert after == before


@pytest.mark.requires_postgres
def test_loader_uses_actual_catalog_and_preserves_unrelated_rows(db):
    catalog = _catalog()
    approved, legacy = _unrelated_recipes()
    db.add_all([approved, legacy])
    db.commit()
    approved_before = {
        "name": approved.name,
        "external_source": approved.external_source,
        "external_id": approved.external_id,
        "source_url": approved.source_url,
        "discovery_reason": approved.discovery_reason,
        "catalog_status": approved.catalog_status,
    }
    legacy_before = {
        "name": legacy.name,
        "catalog_status": legacy.catalog_status,
        "catalog_findings": legacy.catalog_findings,
    }
    db.rollback()

    first = load_bundled_starter_catalog(db)
    assert first["created_recipes"] == STARTER_RECIPE_COUNT
    assert (
        db.query(Recipe)
        .filter(Recipe.discovery_reason == STARTER_DISCOVERY_REASON)
        .count()
        == STARTER_RECIPE_COUNT
    )
    loaded = (
        db.query(Recipe)
        .filter(Recipe.name == catalog["recipes"][0]["name"])  # type: ignore[index]
        .one()
    )
    expected = _catalog_recipe(catalog, loaded.name)
    assert _ingredient_values(db, loaded) == [
        (" ".join(item["name"].split()).casefold(), item["quantity"], item["unit"])
        for item in expected["ingredients"]
    ]
    approved_after = db.get(Recipe, approved.id)
    legacy_after = db.get(Recipe, legacy.id)
    assert {
        "name": approved_after.name,
        "external_source": approved_after.external_source,
        "external_id": approved_after.external_id,
        "source_url": approved_after.source_url,
        "discovery_reason": approved_after.discovery_reason,
        "catalog_status": approved_after.catalog_status,
    } == approved_before
    assert {
        "name": legacy_after.name,
        "catalog_status": legacy_after.catalog_status,
        "catalog_findings": legacy_after.catalog_findings,
    } == legacy_before

    db.rollback()
    second = load_bundled_starter_catalog(db)
    assert second["created_recipes"] == 0
    assert second["existing_recipes"] == STARTER_RECIPE_COUNT

@pytest.mark.requires_postgres
@pytest.mark.parametrize("field", ["description", "instructions", "ingredients"])
def test_loader_rejects_drifted_bundled_content(db, field):
    catalog = _catalog()
    load_bundled_starter_catalog(db)
    drifted = (
        db.query(Recipe)
        .filter(Recipe.name == catalog["recipes"][0]["name"])  # type: ignore[index]
        .one()
    )
    if field == "description":
        drifted.description = "This description was changed after the initial catalog load."
    elif field == "instructions":
        drifted.instructions = ["This instruction was changed after the initial catalog load."]
    else:
        ingredients = copy.deepcopy(drifted.ingredients)
        ingredients[0]["qty"] += 1
        drifted.ingredients = ingredients
    db.commit()

    with pytest.raises(StarterRecipeCatalogError, match="content has drifted"):
        load_bundled_starter_catalog(db)


@pytest.mark.requires_postgres
def test_update_existing_preserves_recipe_and_meal_plan_item_ids(db):
    catalog = _catalog()
    load_starter_catalog(db, _prior_catalog())
    target = (
        db.query(Recipe)
        .filter(Recipe.name == catalog["recipes"][0]["name"])  # type: ignore[index]
        .one()
    )
    recipe_id = target.id
    prior_ingredients = copy.deepcopy(target.ingredients)
    meal_plan = MealPlan(week_start_date=date(2026, 1, 5))
    db.add(meal_plan)
    db.flush()
    meal_plan_item = MealPlanItem(
        meal_plan_id=meal_plan.id,
        recipe_id=recipe_id,
        day_of_week=1,
        meal_type=MealType.DINNER,
    )
    db.add(meal_plan_item)
    db.commit()

    with pytest.raises(StarterRecipeCatalogError, match="content has drifted"):
        load_bundled_starter_catalog(db)

    summary = load_bundled_starter_catalog(db, update_existing=True)
    assert summary["created_recipes"] == 0
    assert summary["updated_recipes"] == STARTER_RECIPE_COUNT

    db.expire_all()
    updated = db.get(Recipe, recipe_id)
    expected = _catalog_recipe(catalog, updated.name)
    assert updated.id == recipe_id
    assert db.get(MealPlanItem, meal_plan_item.id).recipe_id == recipe_id
    assert updated.description == expected["description"]
    assert updated.calories_per_serving == expected["calories_per_serving"]
    assert updated.instructions == expected["instructions"]
    assert updated.ingredients != prior_ingredients
    assert _ingredient_values(db, updated) == [
        (" ".join(item["name"].split()).casefold(), item["quantity"], item["unit"])
        for item in expected["ingredients"]
    ]

    db.rollback()
    assert load_bundled_starter_catalog(db, update_existing=True)["updated_recipes"] == 0


@pytest.mark.requires_postgres
def test_update_existing_dry_run_reports_updates_without_writes(db):
    catalog = _catalog()
    load_starter_catalog(db, _prior_catalog())
    target = (
        db.query(Recipe)
        .filter(Recipe.name == catalog["recipes"][0]["name"])  # type: ignore[index]
        .one()
    )
    recipe_id = target.id
    before = (
        target.description,
        target.calories_per_serving,
        copy.deepcopy(target.ingredients),
        list(target.instructions),
    )
    db.rollback()

    summary = load_bundled_starter_catalog(db, dry_run=True, update_existing=True)

    assert summary["dry_run"] is True
    assert summary["created_recipes"] == 0
    assert summary["updated_recipes"] == STARTER_RECIPE_COUNT
    db.expire_all()
    unchanged = db.get(Recipe, recipe_id)
    assert (
        unchanged.description,
        unchanged.calories_per_serving,
        unchanged.ingredients,
        unchanged.instructions,
    ) == before


@pytest.mark.requires_postgres
def test_update_existing_rejects_an_incomplete_marker_owned_catalog(db):
    catalog = _catalog()
    load_starter_catalog(db, _prior_catalog())
    missing = (
        db.query(Recipe)
        .filter(Recipe.name == catalog["recipes"][0]["name"])  # type: ignore[index]
        .one()
    )
    missing.discovery_reason = None
    db.commit()

    with pytest.raises(StarterRecipeCatalogError, match="exactly all 50"):
        load_bundled_starter_catalog(db, update_existing=True)


@pytest.mark.requires_postgres
def test_update_existing_rolls_back_all_content_changes_on_flush_failure(db):
    catalog = _catalog()
    load_starter_catalog(db, _prior_catalog())
    target = (
        db.query(Recipe)
        .filter(Recipe.name == catalog["recipes"][0]["name"])  # type: ignore[index]
        .one()
    )
    recipe_id = target.id
    before = (
        target.description,
        target.calories_per_serving,
        copy.deepcopy(target.ingredients),
        list(target.instructions),
    )
    db.rollback()

    def fail_recipe_update(mapper, connection, target):
        raise RuntimeError("forced recipe update failure")

    event.listen(Recipe, "before_update", fail_recipe_update)
    try:
        with pytest.raises(RuntimeError, match="forced recipe update failure"):
            load_bundled_starter_catalog(db, update_existing=True)
    finally:
        event.remove(Recipe, "before_update", fail_recipe_update)

    db.expire_all()
    unchanged = db.get(Recipe, recipe_id)
    assert (
        unchanged.description,
        unchanged.calories_per_serving,
        unchanged.ingredients,
        unchanged.instructions,
    ) == before


@pytest.mark.requires_postgres
def test_loader_rolls_back_all_catalog_writes_if_recipe_flush_fails(db):
    before = (
        db.query(Recipe).count(),
        db.query(Ingredient).count(),
        db.query(Recipe).filter(Recipe.discovery_reason == STARTER_DISCOVERY_REASON).count(),
    )
    db.rollback()

    def fail_recipe_flush(mapper, connection, target):
        raise RuntimeError("forced recipe flush failure")

    event.listen(Recipe, "before_insert", fail_recipe_flush)
    try:
        with pytest.raises(RuntimeError, match="forced recipe flush failure"):
            load_bundled_starter_catalog(db)
    finally:
        event.remove(Recipe, "before_insert", fail_recipe_flush)

    after = (
        db.query(Recipe).count(),
        db.query(Ingredient).count(),
        db.query(Recipe).filter(Recipe.discovery_reason == STARTER_DISCOVERY_REASON).count(),
    )
    assert after == before


@pytest.mark.requires_postgres
def test_concurrent_sessions_create_one_bundled_catalog(db):
    engine = db.get_bind().engine
    Session = sessionmaker(bind=engine, autocommit=False, autoflush=False)
    baseline = Session()
    try:
        initial_ingredient_ids = {ingredient.id for ingredient in baseline.query(Ingredient).all()}
        assert (
            baseline.query(Recipe)
            .filter(Recipe.discovery_reason == STARTER_DISCOVERY_REASON)
            .count()
            == 0
        )
    finally:
        baseline.rollback()
        baseline.close()

    barrier = threading.Barrier(2)
    results: Queue[dict[str, object]] = Queue()
    errors: Queue[BaseException] = Queue()

    def load_in_session() -> None:
        session = Session()
        try:
            barrier.wait(timeout=5)
            results.put(load_bundled_starter_catalog(session))
        except BaseException as exc:
            session.rollback()
            errors.put(exc)
        finally:
            session.close()

    threads = [threading.Thread(target=load_in_session) for _ in range(2)]
    for thread in threads:
        thread.start()
    for thread in threads:
        thread.join(timeout=20)

    try:
        assert all(not thread.is_alive() for thread in threads)
        if not errors.empty():
            raise errors.get_nowait()
        try:
            summaries = [results.get_nowait(), results.get_nowait()]
        except Empty as exc:
            raise AssertionError("concurrent catalog load did not return two summaries") from exc
        assert sorted(summary["created_recipes"] for summary in summaries) == [0, 50]

        verifier = Session()
        try:
            bundled = (
                verifier.query(Recipe)
                .filter(Recipe.discovery_reason == STARTER_DISCOVERY_REASON)
                .all()
            )
            assert len(bundled) == STARTER_RECIPE_COUNT
            assert len({recipe.name for recipe in bundled}) == STARTER_RECIPE_COUNT
        finally:
            verifier.rollback()
            verifier.close()
    finally:
        cleanup = Session()
        try:
            bundled = (
                cleanup.query(Recipe)
                .filter(Recipe.discovery_reason == STARTER_DISCOVERY_REASON)
                .all()
            )
            for recipe in bundled:
                cleanup.delete(recipe)
            cleanup.flush()
            for ingredient in cleanup.query(Ingredient).all():
                if ingredient.id not in initial_ingredient_ids:
                    cleanup.delete(ingredient)
            cleanup.commit()
        finally:
            cleanup.rollback()
            cleanup.close()


@pytest.mark.requires_postgres
def test_concurrent_update_existing_calls_serialize(db):
    catalog = _catalog()
    load_starter_catalog(db, _prior_catalog())
    engine = db.get_bind().engine
    Session = sessionmaker(bind=engine, autocommit=False, autoflush=False)
    barrier = threading.Barrier(2)
    results: Queue[dict[str, object]] = Queue()
    errors: Queue[BaseException] = Queue()

    def update_in_session() -> None:
        session = Session()
        try:
            barrier.wait(timeout=5)
            results.put(load_bundled_starter_catalog(session, update_existing=True))
        except BaseException as exc:
            session.rollback()
            errors.put(exc)
        finally:
            session.close()

    threads = [threading.Thread(target=update_in_session) for _ in range(2)]
    for thread in threads:
        thread.start()
    for thread in threads:
        thread.join(timeout=20)

    assert all(not thread.is_alive() for thread in threads)
    if not errors.empty():
        raise errors.get_nowait()
    try:
        summaries = [results.get_nowait(), results.get_nowait()]
    except Empty as exc:
        raise AssertionError("concurrent catalog update did not return two summaries") from exc
    assert sorted(summary["updated_recipes"] for summary in summaries) == [0, 50]

    verifier = Session()
    try:
        target = (
            verifier.query(Recipe)
            .filter(Recipe.name == catalog["recipes"][0]["name"])  # type: ignore[index]
            .one()
        )
        expected = _catalog_recipe(catalog, target.name)
        assert target.description == expected["description"]
        assert target.calories_per_serving == expected["calories_per_serving"]
    finally:
        verifier.rollback()
        verifier.close()