Meal-Planner/backend/tests/test_recipe_catalog_audit.py

"""Catalog quarantine migration, audit, and visibility tests."""
from __future__ import annotations

import importlib.util
import os
import subprocess
from datetime import date, datetime, timezone
from pathlib import Path
from urllib.parse import urlparse
from uuid import uuid4

import pytest

from app.models import (
    Ingredient,
    MealPlan,
    Recipe,
    RecipeCatalogStatus,
    RecipeImportStaging,
    RecipeImportStatus,
)
from app.services.recipe_catalog_audit import audit_recipe_catalog


pytestmark = pytest.mark.requires_postgres
BACKEND_ROOT = Path(__file__).resolve().parent.parent
MIGRATION_PATH = BACKEND_ROOT / "alembic" / "versions" / "0018_recipe_catalog_quarantine.py"


def _ingredient(db, name: str) -> Ingredient:
    row = Ingredient(name=name, name_lower=name.casefold(), aliases=[])
    db.add(row)
    db.flush()
    return row


def _recipe(name: str, ingredients: list[Ingredient], **overrides) -> Recipe:
    values = {
        "name": name,
        "description": "A complete test recipe.",
        "prep_time_minutes": 10,
        "cook_time_minutes": 20,
        "servings": 4,
        "ingredients": [
            {"ingredient_id": str(item.id), "qty": 1.0, "unit": "cup"}
            for item in ingredients
        ],
        "instructions": ["Prepare the ingredients.", "Cook until done."],
        "is_manually_added": True,
    }
    values.update(overrides)
    return Recipe(**values)


def _approved_link(recipe: Recipe, **overrides) -> RecipeImportStaging:
    values = {
        "source": "wikibooks",
        "external_id": recipe.external_id,
        "source_url": recipe.source_url,
        "source_revision": str(uuid4()),
        "license_name": "CC BY-SA 4.0",
        "license_url": "https://creativecommons.org/licenses/by-sa/4.0/",
        "attribution": "Wikibooks contributors",
        "raw_payload": {"title": recipe.name},
        "payload_sha256": "a" * 64,
        "normalized_recipe": {"name": recipe.name},
        "validation_report": {"valid": True, "errors": [], "warnings": []},
        "status": RecipeImportStatus.APPROVED,
        "recipe_id": recipe.id,
        "reviewed_at": datetime(2026, 8, 14, tzinfo=timezone.utc),
        "reviewed_by": "test-admin",
    }
    values.update(overrides)
    return RecipeImportStaging(**values)


def test_recipe_catalog_model_and_migration_chain():
    table = Recipe.__table__
    assert table.c.catalog_status.type.enums == ["active", "quarantined"]
    assert table.c.catalog_status.default.arg is RecipeCatalogStatus.ACTIVE
    assert table.c.catalog_status.server_default.arg == "active"
    assert table.c.catalog_findings.nullable is False

    spec = importlib.util.spec_from_file_location("catalog_quarantine", MIGRATION_PATH)
    assert spec and spec.loader
    migration = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(migration)
    assert migration.revision == "0018"
    assert migration.down_revision == "0017"


def test_recipe_catalog_alembic_round_trip_from_0017():
    dsn = os.environ.get("TEST_DATABASE_URL")
    if not dsn:
        pytest.skip("TEST_DATABASE_URL is required to avoid modifying a non-test database")
    database_name = urlparse(dsn).path.rsplit("/", 1)[-1]
    if not database_name.endswith("_test"):
        pytest.skip("TEST_DATABASE_URL must target a database ending in '_test'")

    env = os.environ.copy()
    env["DATABASE_URL"] = dsn

    def alembic(*args: str) -> subprocess.CompletedProcess[str]:
        return subprocess.run(
            ["alembic", *args],
            cwd=str(BACKEND_ROOT),
            env=env,
            capture_output=True,
            text=True,
        )

    try:
        at_head = alembic("upgrade", "head")
        assert at_head.returncode == 0, at_head.stderr
        at_0017 = alembic("downgrade", "0017")
        assert at_0017.returncode == 0, at_0017.stderr
        at_0018 = alembic("upgrade", "0018")
        assert at_0018.returncode == 0, at_0018.stderr
        back_to_0017 = alembic("downgrade", "0017")
        assert back_to_0017.returncode == 0, back_to_0017.stderr
        back_to_0018 = alembic("upgrade", "0018")
        assert back_to_0018.returncode == 0, back_to_0018.stderr
    finally:
        restore_head = alembic("upgrade", "head")
        assert restore_head.returncode == 0, restore_head.stderr


def test_audit_reports_without_mutation_and_applies_quarantine(db):
    pork = _ingredient(db, f"Audit Pork {uuid4()}")
    salt = _ingredient(db, f"Audit Salt {uuid4()}")
    manual = _recipe("Audit Pork Salt", [pork, salt])
    unsafe = _recipe(
        "Audit Apple Pork Glaze",
        [pork, salt],
        is_manually_added=False,
        external_source="spoonacular",
        external_id=str(uuid4()),
        source_url="https://example.invalid/source",
    )
    db.add_all([manual, unsafe])
    db.flush()

    report = audit_recipe_catalog(
        db, apply=False, reviewed_at=datetime(2026, 8, 14, tzinfo=timezone.utc)
    )
    records = {row["id"]: row for row in report["recipes"]}
    assert records[str(manual.id)]["catalog_status"] == "active"
    assert records[str(unsafe.id)]["catalog_status"] == "quarantined"
    unsafe_codes = {item["code"] for item in records[str(unsafe.id)]["findings"]}
    assert "unlicensed_external_source" in unsafe_codes
    assert "title_ingredient_mismatch" in unsafe_codes
    assert unsafe.catalog_status is RecipeCatalogStatus.ACTIVE

    audit_recipe_catalog(
        db, apply=True, reviewed_at=datetime(2026, 8, 14, tzinfo=timezone.utc)
    )
    assert manual.catalog_status is RecipeCatalogStatus.ACTIVE
    assert unsafe.catalog_status is RecipeCatalogStatus.QUARANTINED
    assert unsafe.catalog_reviewed_at is not None
    assert unsafe.catalog_findings


def test_audit_requires_exact_approved_wikibooks_provenance(db):
    first = _ingredient(db, f"Provenance First {uuid4()}")
    second = _ingredient(db, f"Provenance Second {uuid4()}")
    recipe = _recipe(
        "Provenance First Second",
        [first, second],
        is_manually_added=False,
        external_source="wikibooks",
        external_id=str(uuid4()),
        source_url="https://en.wikibooks.org/wiki/Cookbook:Provenance_Test",
    )
    db.add(recipe)
    db.flush()
    db.add(_approved_link(recipe, external_id="unrelated-page"))
    db.flush()

    report = audit_recipe_catalog(
        db, reviewed_at=datetime(2026, 8, 14, tzinfo=timezone.utc)
    )
    record = next(row for row in report["recipes"] if row["id"] == str(recipe.id))
    assert "missing_import_provenance" in {
        finding["code"] for finding in record["findings"]
    }

    db.add(_approved_link(recipe))
    db.flush()
    report = audit_recipe_catalog(
        db, reviewed_at=datetime(2026, 8, 14, tzinfo=timezone.utc)
    )
    record = next(row for row in report["recipes"] if row["id"] == str(recipe.id))
    assert "missing_import_provenance" not in {
        finding["code"] for finding in record["findings"]
    }


def test_audit_preserves_operator_quarantine_for_valid_wikibooks_recipe(db):
    first = _ingredient(db, f"Operator First {uuid4()}")
    second = _ingredient(db, f"Operator Second {uuid4()}")
    operator_finding = {
        "code": "operator_quarantine",
        "severity": "error",
        "field": "catalog_status",
        "message": "Quarantined by an operator.",
    }
    recipe = _recipe(
        "Operator First Second",
        [first, second],
        is_manually_added=False,
        external_source="wikibooks",
        external_id=str(uuid4()),
        source_url="https://en.wikibooks.org/wiki/Cookbook:Operator_Test",
        catalog_findings=[operator_finding],
    )
    db.add(recipe)
    db.flush()
    db.add(_approved_link(recipe))
    db.flush()

    report = audit_recipe_catalog(
        db, apply=False, reviewed_at=datetime(2026, 8, 14, tzinfo=timezone.utc)
    )
    record = next(row for row in report["recipes"] if row["id"] == str(recipe.id))
    assert record["catalog_status"] == "quarantined"
    assert operator_finding in record["findings"]

    applied_report = audit_recipe_catalog(
        db, apply=True, reviewed_at=datetime(2026, 8, 14, tzinfo=timezone.utc)
    )
    applied_record = next(
        row for row in applied_report["recipes"] if row["id"] == str(recipe.id)
    )
    assert applied_record["catalog_status"] == "quarantined"
    assert operator_finding in applied_record["findings"]
    assert recipe.catalog_status is RecipeCatalogStatus.QUARANTINED
    assert operator_finding in recipe.catalog_findings


@pytest.mark.parametrize(
    "prior_findings",
    [
        1,
        {"code": "operator_quarantine", "severity": "error"},
        None,
    ],
    ids=["truthy-scalar", "truthy-object", "null"],
)
def test_audit_discards_malformed_prior_catalog_findings(db, prior_findings):
    first = _ingredient(db, f"Malformed First {uuid4()}")
    second = _ingredient(db, f"Malformed Second {uuid4()}")
    recipe = _recipe(
        "Malformed First Second",
        [first, second],
        catalog_findings=prior_findings,
    )
    db.add(recipe)
    db.flush()

    report = audit_recipe_catalog(
        db, apply=True, reviewed_at=datetime(2026, 8, 14, tzinfo=timezone.utc)
    )
    record = next(row for row in report["recipes"] if row["id"] == str(recipe.id))

    assert record["catalog_status"] == "active"
    assert record["findings"] == []
    assert recipe.catalog_findings == []


def test_audit_discards_stale_normal_findings(db):
    first = _ingredient(db, f"Stale First {uuid4()}")
    second = _ingredient(db, f"Stale Second {uuid4()}")
    stale_finding = {
        "code": "obsolete_validation_error",
        "severity": "error",
        "field": "name",
        "message": "No longer applies.",
    }
    recipe = _recipe(
        "Stale First Second",
        [first, second],
        catalog_findings=[stale_finding],
    )
    db.add(recipe)
    db.flush()

    report = audit_recipe_catalog(
        db, apply=False, reviewed_at=datetime(2026, 8, 14, tzinfo=timezone.utc)
    )
    record = next(row for row in report["recipes"] if row["id"] == str(recipe.id))

    assert record["catalog_status"] == "active"
    assert record["findings"] == []
    assert recipe.catalog_findings == [stale_finding]

    audit_recipe_catalog(
        db, apply=True, reviewed_at=datetime(2026, 8, 14, tzinfo=timezone.utc)
    )
    assert recipe.catalog_findings == []


def test_public_catalog_hides_quarantined_recipes(client, db):
    first = _ingredient(db, f"Visible First {uuid4()}")
    second = _ingredient(db, f"Visible Second {uuid4()}")
    active = _recipe("Visible First Second", [first, second])
    quarantined = _recipe(
        "Hidden First Second",
        [first, second],
        catalog_status=RecipeCatalogStatus.QUARANTINED,
        catalog_findings=[{"code": "test_quarantine", "severity": "error"}],
    )
    db.add_all([active, quarantined])
    db.flush()

    response = client.get("/api/recipes")
    assert response.status_code == 200
    ids = {row["id"] for row in response.json()}
    assert str(active.id) in ids
    assert str(quarantined.id) not in ids
    assert client.get(f"/api/recipes/{quarantined.id}").status_code == 404


def test_meal_plan_creation_rejects_quarantined_recipe(client, db):
    first = _ingredient(db, f"Plan First {uuid4()}")
    second = _ingredient(db, f"Plan Second {uuid4()}")
    quarantined = _recipe(
        "Plan First Second",
        [first, second],
        catalog_status=RecipeCatalogStatus.QUARANTINED,
    )
    db.add(quarantined)
    db.flush()
    login = client.post("/api/auth/login", json={"password": "test-family-password"})
    assert login.status_code == 204, login.text

    response = client.post(
        "/api/meals",
        json={
            "week_start_date": "2031-01-03",
            "items": [
                {
                    "recipe_id": str(quarantined.id),
                    "day_of_week": 1,
                    "meal_type": "dinner",
                }
            ],
        },
    )

    assert response.status_code == 400
    assert response.json()["detail"] == "Meal plan contains unavailable recipes"


def test_generate_single_item_reports_empty_active_catalog(client, db):
    db.query(Recipe).update(
        {Recipe.catalog_status: RecipeCatalogStatus.QUARANTINED},
        synchronize_session=False,
    )
    plan = MealPlan(week_start_date=date(2031, 1, 13))
    db.add(plan)
    db.flush()
    login = client.post("/api/auth/login", json={"password": "test-family-password"})
    assert login.status_code == 204, login.text

    response = client.post(
        f"/api/meals/{plan.id}/generate-item",
        params={"day_of_week": 4, "meal_type": "dinner"},
    )

    assert response.status_code == 409
    assert response.json()["detail"] == (
        "No active recipes are available. Approve a staged recipe before generating meals."
    )


def test_llm_library_includes_active_global_and_excludes_quarantined(db):
    from app.models import FamilyProfile
    from app.api.llm_plan import _serialize_library

    family = FamilyProfile(
        name=f"Audit Family {uuid4()}",
        household_size=2,
        adult_count=2,
        child_count=0,
    )
    first = _ingredient(db, f"Library First {uuid4()}")
    second = _ingredient(db, f"Library Second {uuid4()}")
    active = _recipe("Library First Second", [first, second], family_profile_id=None)
    quarantined = _recipe(
        "Quarantined First Second",
        [first, second],
        family_profile_id=None,
        catalog_status=RecipeCatalogStatus.QUARANTINED,
    )
    db.add_all([family, active, quarantined])
    db.flush()

    ids = {row["id"] for row in _serialize_library(db, family.id)}
    assert str(active.id) in ids
    assert str(quarantined.id) not in ids