from __future__ import annotations from pathlib import Path from unittest import TestCase _REPO_ROOT = Path(__file__).resolve().parents[2] class TestFaqApiContracts(TestCase): def _read_faq_router(self) -> str: path = _REPO_ROOT / "backend/app/routers/faq_entries.py" return path.read_text(encoding="utf-8") def _read_faq_model(self) -> str: path = _REPO_ROOT / "backend/app/models/faq_entry.py" return path.read_text(encoding="utf-8") def _read_knowledge_model(self) -> str: path = _REPO_ROOT / "backend/app/models/knowledge.py" return path.read_text(encoding="utf-8") def _read_knowledge_router(self) -> str: path = _REPO_ROOT / "backend/app/routers/knowledge.py" return path.read_text(encoding="utf-8") def _read_main(self) -> str: path = _REPO_ROOT / "backend/app/main.py" return path.read_text(encoding="utf-8") def _read_models_init(self) -> str: path = _REPO_ROOT / "backend/app/models/__init__.py" return path.read_text(encoding="utf-8") def _read_migration(self) -> str: path = _REPO_ROOT / "backend/alembic/versions/0010_add_faq_entry_table.py" return path.read_text(encoding="utf-8") # --- Model contracts --- def test_faq_model_has_all_schema_classes(self) -> None: source = self._read_faq_model() self.assertIn("class FaqEntryBase(SQLModel)", source) self.assertIn("class FaqEntry(FaqEntryBase, TenantModel, table=True)", source) self.assertIn("class FaqEntryCreate(SQLModel)", source) self.assertIn("class FaqEntryRead(FaqEntryBase)", source) self.assertIn("class FaqEntryUpdate(SQLModel)", source) def test_faq_model_inherits_tenant_model(self) -> None: source = self._read_faq_model() self.assertIn("from app.db.base import TenantModel", source) self.assertIn("TenantModel", source) def test_faq_model_uses_text_columns_for_question_answer(self) -> None: source = self._read_faq_model() self.assertIn("Column(Text, nullable=False)", source) self.assertIn("question: str", source) self.assertIn("answer: str", source) def test_faq_model_has_category_and_is_active(self) -> None: source = self._read_faq_model() self.assertIn("category: str | None", source) self.assertIn("is_active: bool", source) def test_faq_model_exports_in_init(self) -> None: source = self._read_models_init() self.assertIn("from .faq_entry import FaqEntry", source) self.assertIn('"FaqEntry"', source) self.assertIn('"FaqEntryCreate"', source) self.assertIn('"FaqEntryRead"', source) self.assertIn('"FaqEntryUpdate"', source) # --- Migration contracts --- def test_migration_creates_faq_entry_table(self) -> None: source = self._read_migration() self.assertIn('"faq_entry"', source) self.assertIn("op.create_table", source) def test_migration_has_correct_columns(self) -> None: source = self._read_migration() self.assertIn('"id"', source) self.assertIn('"restaurant_id"', source) self.assertIn('"question"', source) self.assertIn('"answer"', source) self.assertIn('"category"', source) self.assertIn('"is_active"', source) self.assertIn('"created_at"', source) self.assertIn('"updated_at"', source) def test_migration_has_restaurant_id_fk(self) -> None: source = self._read_migration() self.assertIn('ForeignKeyConstraint(["restaurant_id"], ["restaurant.id"])', source) def test_migration_has_indexes(self) -> None: source = self._read_migration() self.assertIn("ix_faq_entry_restaurant_id", source) self.assertIn("ix_faq_entry_restaurant_updated", source) def test_migration_has_downgrade(self) -> None: source = self._read_migration() self.assertIn("def downgrade()", source) self.assertIn("op.drop_table", source) # --- Router contracts --- def test_router_is_registered_in_main(self) -> None: source = self._read_main() self.assertIn("faq_entries", source) self.assertIn("faq_entries.router", source) def test_list_endpoint_is_tenant_scoped(self) -> None: source = self._read_faq_router() self.assertIn("FaqEntry.restaurant_id == restaurant.id", source) def test_list_endpoint_orders_by_updated_at(self) -> None: source = self._read_faq_router() self.assertIn("updated_at.desc()", source) def test_list_endpoint_has_pagination(self) -> None: source = self._read_faq_router() self.assertIn("limit: int = Query(", source) self.assertIn("offset: int = Query(", source) def test_list_endpoint_has_category_filter(self) -> None: source = self._read_faq_router() self.assertIn("category: str | None = Query(", source) self.assertIn("FaqEntry.category == category", source) def test_list_endpoint_has_is_active_filter(self) -> None: source = self._read_faq_router() self.assertIn("is_active: bool | None = Query(", source) self.assertIn("FaqEntry.is_active == is_active", source) def test_create_endpoint_validates_non_empty(self) -> None: source = self._read_faq_router() self.assertIn("HTTP_201_CREATED", source) self.assertIn("Question and answer must be non-empty", source) def test_update_endpoint_has_ownership_check(self) -> None: source = self._read_faq_router() self.assertIn("FaqEntry.id == entry_id", source) self.assertIn("FaqEntry.restaurant_id == restaurant.id", source) self.assertIn("FAQ entry not found", source) def test_update_endpoint_validates_non_empty_fields(self) -> None: source = self._read_faq_router() self.assertIn("Question must be non-empty", source) self.assertIn("Answer must be non-empty", source) def test_delete_endpoint_has_ownership_check(self) -> None: source = self._read_faq_router() self.assertIn("@router.delete", source) self.assertIn("await session.delete(entry)", source) def test_delete_cleans_up_knowledge_documents(self) -> None: source = self._read_faq_router() self.assertIn('KnowledgeDocument.source == "faq"', source) self.assertIn('KnowledgeDocument.metadata_["faq_entry_id"]', source) # --- Bulk import contracts --- def test_bulk_import_endpoint_exists(self) -> None: source = self._read_faq_router() self.assertIn('@router.post("/import")', source) self.assertIn("FaqImportRequest", source) def test_bulk_import_has_duplicate_prevention(self) -> None: source = self._read_faq_router() self.assertIn(".strip().lower()", source) self.assertIn("is_dup", source) def test_bulk_import_returns_counts(self) -> None: source = self._read_faq_router() self.assertIn('"imported"', source) self.assertIn('"skipped"', source) self.assertIn('"total"', source) # --- RAG indexing contracts --- def test_rag_sync_helper_exists(self) -> None: source = self._read_faq_router() self.assertIn("_sync_faq_embedding", source) self.assertIn("generate_embedding", source) def test_rag_sync_creates_knowledge_document(self) -> None: source = self._read_faq_router() self.assertIn('source="faq"', source) self.assertIn("faq_entry_id", source) def test_rag_sync_handles_deactivation(self) -> None: source = self._read_faq_router() self.assertIn("not entry.is_active", source) self.assertIn("await session.delete(existing_doc)", source) def test_rag_sync_handles_embedding_failure(self) -> None: source = self._read_faq_router() self.assertIn("faq_embedding_failed", source) def test_create_triggers_rag_sync(self) -> None: source = self._read_faq_router() # After create endpoint commit, _sync_faq_embedding is called self.assertIn("await _sync_faq_embedding(entry, session)", source) def test_update_triggers_rag_sync(self) -> None: source = self._read_faq_router() # Ensure sync is called in update path too lines = source.split("\n") sync_calls = [i for i, line in enumerate(lines) if "_sync_faq_embedding" in line] # At least 3 occurrences: definition + create call + update call self.assertGreaterEqual(len(sync_calls), 3) # --- Knowledge lifecycle contracts --- def test_knowledge_model_has_source_type_field(self) -> None: source = self._read_knowledge_model() self.assertIn("source_type", source) def test_knowledge_router_has_source_filter(self) -> None: source = self._read_knowledge_router() self.assertIn("source: str | None = Query(", source) self.assertIn("KnowledgeDocument.source == source", source)