"""Tests for menu-item / knowledge-document linkage. Covers: - menu_item table schema presence and column types (migration verification) - JSONB metadata linkage persistence in knowledge_document - Tenant-scoped isolation for both menu items and linkage queries - KnowledgeDocumentRead model_validator extracts linkage fields """ import asyncio import os import unittest import uuid from typing import Any from sqlalchemy import bindparam, text from sqlalchemy.dialects.postgresql import JSONB from sqlalchemy.ext.asyncio import ( AsyncSession, async_sessionmaker, create_async_engine, ) from app.models.knowledge import KnowledgeDocumentRead def _neon_url() -> str | None: raw = os.environ.get("NEON_DATABASE_URL", "") if not raw or "localhost" in raw or "127.0.0.1" in raw: return None return raw.replace("postgresql://", "postgresql+asyncpg://", 1).split("?")[0] def _run(coro): # noqa: ANN001, ANN202 try: loop = asyncio.get_event_loop() except RuntimeError: loop = asyncio.new_event_loop() asyncio.set_event_loop(loop) return loop.run_until_complete(coro) _DB_URL = _neon_url() _SKIP = "NEON_DATABASE_URL not set" _INSERT_ITEM = ( "INSERT INTO menu_item " "(id, restaurant_id, name, price_cents, category, " "is_active, sort_order, created_at) " "VALUES (:id, :rid, :name, :price, :cat, true, 0, now())" ) _INSERT_DOC = ( "INSERT INTO knowledge_document " "(id, restaurant_id, content, source, created_at, metadata) " "VALUES (:id, :rid, :content, :source, now(), :meta)" ) @unittest.skipIf(_DB_URL is None, _SKIP) class TestMenuItemTableSchema(unittest.TestCase): """Verify menu_item table exists with expected columns.""" def test_menu_item_table_has_required_columns(self) -> None: async def _check() -> dict[str, tuple[str, str]]: assert _DB_URL is not None engine = create_async_engine( _DB_URL, connect_args={"ssl": "require"}, ) factory = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False) async with factory() as s: r = await s.execute( text( "SELECT column_name, data_type, is_nullable " "FROM information_schema.columns " "WHERE table_name = 'menu_item' " "ORDER BY ordinal_position" ) ) cols = {row[0]: (row[1], row[2]) for row in r} await engine.dispose() return cols cols = _run(_check()) for name in ( "id", "restaurant_id", "name", "price_cents", "category", "photo_url", "is_active", "sort_order", "created_at", "updated_at", ): self.assertIn(name, cols) # Non-nullable constraints for name in ("id", "restaurant_id", "name", "price_cents", "is_active"): self.assertEqual(cols[name][1], "NO", f"{name} should be NOT NULL") def test_menu_item_has_indexes(self) -> None: async def _check() -> list[str]: assert _DB_URL is not None engine = create_async_engine( _DB_URL, connect_args={"ssl": "require"}, ) factory = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False) async with factory() as s: r = await s.execute( text("SELECT indexname FROM pg_indexes WHERE tablename = 'menu_item'") ) names = [row[0] for row in r] await engine.dispose() return names names = _run(_check()) self.assertIn("ix_menu_item_restaurant_id", names) self.assertIn("ix_menu_item_restaurant_category", names) @unittest.skipIf(_DB_URL is None, _SKIP) class TestMenuLinkagePersistence(unittest.TestCase): """Verify menu-item / knowledge-document linkage round-trips.""" @classmethod def setUpClass(cls) -> None: assert _DB_URL is not None cls._engine = create_async_engine( _DB_URL, connect_args={"ssl": "require"}, ) cls._factory = async_sessionmaker(cls._engine, class_=AsyncSession, expire_on_commit=False) async def _get_rid() -> str: async with cls._factory() as s: r = await s.execute(text("SELECT id FROM restaurant LIMIT 1")) row = r.first() if row is None: raise unittest.SkipTest("No restaurant in database") return row[0] cls._rid = _run(_get_rid()) cls._item_id = str(uuid.uuid4()) cls._doc_id = str(uuid.uuid4()) cls._linkage = { "menu_item_links": { "item_ids": [cls._item_id], "status": "linked", } } async def _seed() -> None: async with cls._factory() as s: await s.execute( text(_INSERT_ITEM), { "id": cls._item_id, "rid": cls._rid, "name": "Test Linkage Item", "price": 999, "cat": "test", }, ) await s.commit() async with cls._factory() as s: stmt = text(_INSERT_DOC).bindparams(bindparam("meta", type_=JSONB)) await s.execute( stmt, { "id": cls._doc_id, "rid": cls._rid, "content": "Test doc", "source": "menu", "meta": cls._linkage, }, ) await s.commit() _run(_seed()) @classmethod def tearDownClass(cls) -> None: async def _cleanup() -> None: async with cls._factory() as s: await s.execute( text("DELETE FROM knowledge_document WHERE id = :id"), {"id": cls._doc_id}, ) await s.execute( text("DELETE FROM menu_item WHERE id = :id"), {"id": cls._item_id}, ) await s.commit() await cls._engine.dispose() _run(_cleanup()) def test_linkage_item_ids_persisted(self) -> None: async def _check() -> Any: async with self._factory() as s: r = await s.execute( text( "SELECT " "metadata->'menu_item_links'->'item_ids', " "metadata->'menu_item_links'->>'status' " "FROM knowledge_document " "WHERE id = :id AND restaurant_id = :rid" ), {"id": self._doc_id, "rid": self._rid}, ) return r.first() row = _run(_check()) self.assertIsNotNone(row) self.assertIn(self._item_id, row[0]) self.assertEqual(row[1], "linked") def test_linkage_invisible_to_other_restaurant(self) -> None: async def _check() -> int: fake = str(uuid.uuid4()) async with self._factory() as s: r = await s.execute( text( "SELECT count(*) FROM knowledge_document " "WHERE restaurant_id = :rid " "AND metadata->'menu_item_links' IS NOT NULL" ), {"rid": fake}, ) return r.scalar() self.assertEqual(_run(_check()), 0) def test_menu_item_invisible_to_other_restaurant(self) -> None: async def _check() -> int: fake = str(uuid.uuid4()) async with self._factory() as s: r = await s.execute( text("SELECT count(*) FROM menu_item WHERE restaurant_id = :rid"), {"rid": fake}, ) return r.scalar() self.assertEqual(_run(_check()), 0) class TestKnowledgeDocumentReadLinkageDTO(unittest.TestCase): """Verify model_validator extracts linkage fields.""" def _make(self, meta: dict | None = None) -> KnowledgeDocumentRead: return KnowledgeDocumentRead( id="d1", restaurant_id="r1", content="c", source="s", metadata_=meta, ) def test_extracts_item_ids_and_count(self) -> None: dto = self._make( { "menu_item_links": { "item_ids": ["i1", "i2"], "status": "linked", } } ) self.assertEqual(dto.menu_item_count, 2) self.assertEqual(dto.menu_item_ids, ["i1", "i2"]) def test_handles_none_metadata(self) -> None: dto = self._make(None) self.assertEqual(dto.menu_item_count, 0) self.assertEqual(dto.menu_item_ids, []) def test_handles_empty_metadata(self) -> None: dto = self._make({}) self.assertEqual(dto.menu_item_count, 0) self.assertEqual(dto.menu_item_ids, []) def test_handles_metadata_without_linkage_key(self) -> None: dto = self._make({"other_key": "val"}) self.assertEqual(dto.menu_item_count, 0) self.assertEqual(dto.menu_item_ids, []) def test_handles_empty_item_ids(self) -> None: dto = self._make({"menu_item_links": {"item_ids": [], "status": "linked"}}) self.assertEqual(dto.menu_item_count, 0) self.assertEqual(dto.menu_item_ids, []) def test_filename_becomes_title(self) -> None: dto = self._make({"filename": "menu-2026.pdf"}) self.assertEqual(dto.title, "menu-2026.pdf") def test_title_falls_back_to_first_content_line(self) -> None: dto = KnowledgeDocumentRead( id="d1", restaurant_id="r1", content="House special: truffle risotto\nLong description here", source="other", metadata_=None, ) self.assertEqual(dto.title, "House special: truffle risotto") def test_common_question_extracts_question_and_answer(self) -> None: dto = KnowledgeDocumentRead( id="d1", restaurant_id="r1", content="Q: Heeft u een terras?\nA: Ja, op de eerste verdieping.", source="common-question", metadata_={ "common_question_id": "cq-1", "question_key": "outdoor_seating", "source_type": "common-question", }, ) self.assertEqual(dto.source_type, "common-question") self.assertEqual(dto.question_text, "Heeft u een terras?") self.assertEqual(dto.answer, "Ja, op de eerste verdieping.") self.assertEqual(dto.common_question_id, "cq-1") self.assertEqual(dto.question_key, "outdoor_seating") self.assertEqual(dto.title, "Heeft u een terras?") def test_common_question_multiline_answer(self) -> None: dto = KnowledgeDocumentRead( id="d1", restaurant_id="r1", content="Q: Allergenen?\nA: Bel ons gerust.\nWe verwelkomen alle dieetwensen.", source="common-question", metadata_={"common_question_id": "cq-2"}, ) self.assertEqual(dto.answer, "Bel ons gerust.\nWe verwelkomen alle dieetwensen.") def test_status_defaults_to_ready(self) -> None: dto = self._make(None) self.assertEqual(dto.status, "ready")