import importlib from fastapi.testclient import TestClient from sqlalchemy import select ai_module = importlib.import_module("services.ai_orchestrator_service.app") ai_app = ai_module.app from services.interaction_service.app import app as interaction_app from services.shared.core import new_id, utc_now_iso from services.shared.db import get_session from services.shared.sql_models import AISessionRow, KBArticleRow, KBCategoryRow, WhatsAppThreadRow from services.whatsapp_adapter_service import app as whatsapp_module from services.whatsapp_adapter_service.app import app as whatsapp_app def admin_headers(): return {"X-User": "admin", "X-Role": "admin"} def _u(value: str) -> str: return value.encode("ascii").decode("unicode_escape") def seed_kb_article( title: str, body: str, tag: str, *, language: str = "ru", article_group_id: str | None = None, ) -> dict[str, str]: session = get_session() try: now = utc_now_iso() category_id = new_id("kbc") article_id = new_id("kba") resolved_group_id = article_group_id or article_id session.add( KBCategoryRow( category_id=category_id, name="WhatsApp AI", description="AI test category", created_at=now, ) ) session.add( KBArticleRow( article_id=article_id, category_id=category_id, article_group_id=resolved_group_id, language=language, title=title, body=body, tags_json=f'["{tag}"]', created_at=now, updated_at=now, ) ) session.commit() return {"article_id": article_id, "article_group_id": resolved_group_id} finally: session.close() def patch_ai_internal_calls(monkeypatch, whatsapp_client: TestClient, interaction_client: TestClient): def fake_whatsapp_request(method: str, path: str, *, payload: dict | None = None) -> dict: response = whatsapp_client.request(method, path, json=payload, headers=admin_headers()) response.raise_for_status() return response.json() def fake_interaction_request(method: str, path: str, *, payload: dict | None = None) -> dict: response = interaction_client.request(method, path, json=payload, headers=admin_headers()) response.raise_for_status() return response.json() monkeypatch.setattr(ai_module, "_whatsapp_request", fake_whatsapp_request) monkeypatch.setattr(ai_module, "_interaction_request", fake_interaction_request) def create_inbound_thread(whatsapp_client: TestClient, chat_id: str, text: str) -> dict: response = whatsapp_client.post( "/integrations/whatsapp/webhook", json={ "chat_id": chat_id, "text": text, "external_message_id": f"{chat_id}-ext-1", "whatsapp_user_id": f"user-{chat_id}", "phone_number": "+77010000000", "display_name": f"Lead {chat_id}", }, ) assert response.status_code == 200 return response.json() def fetch_ai_summary(whatsapp_client: TestClient, thread_id: str): return whatsapp_client.get( f"/integrations/whatsapp/threads/{thread_id}/ai-summary", headers=admin_headers(), ) def test_ai_whatsapp_always_reply_mode_answers_without_handoff(monkeypatch): monkeypatch.setenv("AI_WHATSAPP_ENABLED", "1") monkeypatch.setenv("AI_WHATSAPP_ALWAYS_REPLY", "1") monkeypatch.setenv("AI_PROVIDER", "stub") monkeypatch.setattr(whatsapp_module, "_ai_enqueue_request", lambda thread_id, trigger_message_id: None) monkeypatch.setattr(whatsapp_module, "_start_whatsapp_reply_delivery", lambda message_id: None) whatsapp_client = TestClient(whatsapp_app) interaction_client = TestClient(interaction_app) ai_client = TestClient(ai_app) patch_ai_internal_calls(monkeypatch, whatsapp_client, interaction_client) seed_kb_article("WhatsApp greeting", "Здравствуйте! Поможем через WhatsApp.", "greeting") created = create_inbound_thread(whatsapp_client, "wa_ai_greeting", "Привет") enqueue = ai_client.post( f"/ai/whatsapp/threads/{created['thread_id']}/enqueue", headers=admin_headers(), json={"trigger_message_id": created["message_id"]}, ) assert enqueue.status_code == 200 assert enqueue.json()["status"] == "done" thread = whatsapp_client.get( f"/integrations/whatsapp/threads/{created['thread_id']}", headers=admin_headers(), ) assert thread.status_code == 200 assert thread.json()["ai_state"] == "active" messages = whatsapp_client.get( f"/integrations/whatsapp/threads/{created['thread_id']}/messages", headers=admin_headers(), ) assert messages.status_code == 200 assert messages.json()[-1]["author_type"] == "ai" assert messages.json()[-1]["text"] summary = fetch_ai_summary(whatsapp_client, created["thread_id"]) assert summary.status_code == 200 assert summary.json() is None def test_ai_whatsapp_relaxed_kb_search_answers_phrase_query(monkeypatch): monkeypatch.setenv("AI_WHATSAPP_ENABLED", "1") monkeypatch.setenv("AI_PROVIDER", "stub") monkeypatch.setattr(whatsapp_module, "_ai_enqueue_request", lambda thread_id, trigger_message_id: None) monkeypatch.setattr(whatsapp_module, "_start_whatsapp_reply_delivery", lambda message_id: None) whatsapp_client = TestClient(whatsapp_app) interaction_client = TestClient(interaction_app) ai_client = TestClient(ai_app) patch_ai_internal_calls(monkeypatch, whatsapp_client, interaction_client) seed_kb_article( "Tariff warelaxx", "Tariff warelaxx activates after the request is confirmed.", "warelaxx", ) created = create_inbound_thread( whatsapp_client, "wa_ai_relaxed_kb", _u(r"\u0425\u043e\u0447\u0443 \u0443\u0437\u043d\u0430\u0442\u044c \u0442\u0430\u0440\u0438\u0444 warelaxx"), ) enqueue = ai_client.post( f"/ai/whatsapp/threads/{created['thread_id']}/enqueue", headers=admin_headers(), json={"trigger_message_id": created["message_id"]}, ) assert enqueue.status_code == 200 assert enqueue.json()["status"] == "done" messages = whatsapp_client.get( f"/integrations/whatsapp/threads/{created['thread_id']}/messages", headers=admin_headers(), ) assert messages.status_code == 200 assert messages.json()[-1]["author_type"] == "ai" assert "warelaxx" in messages.json()[-1]["text"].lower() def test_ai_whatsapp_handoff_creates_summary(monkeypatch): monkeypatch.setenv("AI_WHATSAPP_ENABLED", "1") monkeypatch.delenv("AI_WHATSAPP_ALWAYS_REPLY", raising=False) monkeypatch.setenv("AI_PROVIDER", "stub") monkeypatch.setattr(whatsapp_module, "_ai_enqueue_request", lambda thread_id, trigger_message_id: None) monkeypatch.setattr(whatsapp_module, "_start_whatsapp_reply_delivery", lambda message_id: None) whatsapp_client = TestClient(whatsapp_app) interaction_client = TestClient(interaction_app) ai_client = TestClient(ai_app) patch_ai_internal_calls(monkeypatch, whatsapp_client, interaction_client) created = create_inbound_thread(whatsapp_client, "wa_ai_handoff", "Соедините с оператором") enqueue = ai_client.post( f"/ai/whatsapp/threads/{created['thread_id']}/enqueue", headers=admin_headers(), json={"trigger_message_id": created["message_id"]}, ) assert enqueue.status_code == 200 assert enqueue.json()["status"] == "handoff_required" thread = whatsapp_client.get( f"/integrations/whatsapp/threads/{created['thread_id']}", headers=admin_headers(), ) assert thread.status_code == 200 assert thread.json()["ai_state"] == "handoff_required" assert thread.json()["claimed_by_user"] is None summary = fetch_ai_summary(whatsapp_client, created["thread_id"]) assert summary.status_code == 200 payload = summary.json() assert payload["thread_id"] == created["thread_id"] assert payload["status_tone"] == "handoff" assert payload["customer_request_text"] == created["text"] assert payload["handoff_reason"] session = get_session() try: thread_row = session.execute( select(WhatsAppThreadRow).where(WhatsAppThreadRow.thread_id == created["thread_id"]) ).scalar_one() ai_session = session.execute( select(AISessionRow).where(AISessionRow.session_id == thread_row.ai_session_id) ).scalar_one() assert thread_row.ai_state == "handoff_required" assert ai_session.status == "handoff_required" finally: session.close()