"""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()