import os import sys import unittest from pathlib import Path from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker ROOT = Path(__file__).resolve().parents[1] os.environ["JWT_SECRET"] = "test-secret-value-with-enough-length-12345" os.environ.setdefault("DB_NAME", ":memory:.db") sys.path.insert(0, str(ROOT)) from app.api import telegram as telegram_api from app.core.database import Base from app.core.models import SystemConfig, Tenant from app.services import telegram_service class TelegramWebhookTextTests(unittest.IsolatedAsyncioTestCase): async def asyncSetUp(self): self.engine = create_engine("sqlite:///:memory:") Base.metadata.create_all(self.engine) self.db = sessionmaker(bind=self.engine)() self.db.add_all([ Tenant(id=1, company_code="test", name="Test"), SystemConfig(tenant_id=1, key_name="telegram_chat_id", key_value="798233405"), SystemConfig(tenant_id=1, key_name="telegram_webhook_secret", key_value="test-secret", is_secret=True), ]) self.db.commit() async def asyncTearDown(self): self.db.close() async def test_authorized_text_message_uses_shared_ai_handler(self): sent_messages = [] original_get_config = telegram_api.get_telegram_config original_send = telegram_api.send_telegram_message original_handler = getattr(telegram_api, "handle_ai_chat_message", None) async def fake_handler(db, message, context=None, tenant_id=None): self.assertEqual("help me", message) self.assertEqual([], context) return telegram_api.ChatResponse( reply="Database-backed answer", action="query_result", source="database", tool_name="query_revenue", result={"ok": True}, ) async def fake_send(chat_id, text, **kwargs): sent_messages.append((chat_id, text)) return {"success": True, "error": None, "status_code": 200} telegram_api.get_telegram_config = lambda: {"bot_token": "token", "chat_id": "798233405"} telegram_api.send_telegram_message = fake_send telegram_api.handle_ai_chat_message = fake_handler try: result = await telegram_api.telegram_webhook( { "message": { "message_id": 1, "chat": {"id": 798233405}, "text": "help me", } }, x_telegram_bot_api_secret_token="test-secret", db=self.db, ) finally: telegram_api.get_telegram_config = original_get_config telegram_api.send_telegram_message = original_send if original_handler is None: delattr(telegram_api, "handle_ai_chat_message") else: telegram_api.handle_ai_chat_message = original_handler self.assertEqual({"ok": True, "handled": "ai_chat", "action": "query_result"}, result) self.assertEqual("798233405", sent_messages[0][0]) self.assertEqual("Database-backed answer", sent_messages[0][1]) async def test_authorized_text_message_reuses_chat_context_until_new(self): sent_messages = [] seen_contexts = [] original_get_config = telegram_api.get_telegram_config original_send = telegram_api.send_telegram_message original_handler = getattr(telegram_api, "handle_ai_chat_message", None) if hasattr(telegram_api, "_TELEGRAM_CHAT_SESSIONS"): telegram_api._TELEGRAM_CHAT_SESSIONS.clear() async def fake_handler(db, message, context=None, tenant_id=None): seen_contexts.append((message, list(context or []))) return telegram_api.ChatResponse(reply=f"reply:{message}") async def fake_send(chat_id, text, **kwargs): sent_messages.append((chat_id, text)) return {"success": True, "error": None, "status_code": 200} telegram_api.get_telegram_config = lambda: {"bot_token": "token", "chat_id": "798233405"} telegram_api.send_telegram_message = fake_send telegram_api.handle_ai_chat_message = fake_handler try: first = await telegram_api.telegram_webhook( {"message": {"message_id": 1, "chat": {"id": 798233405}, "text": "first"}}, x_telegram_bot_api_secret_token="test-secret", db=self.db, ) second = await telegram_api.telegram_webhook( {"message": {"message_id": 2, "chat": {"id": 798233405}, "text": "second"}}, x_telegram_bot_api_secret_token="test-secret", db=self.db, ) reset = await telegram_api.telegram_webhook( {"message": {"message_id": 3, "chat": {"id": 798233405}, "text": "/new"}}, x_telegram_bot_api_secret_token="test-secret", db=self.db, ) third = await telegram_api.telegram_webhook( {"message": {"message_id": 4, "chat": {"id": 798233405}, "text": "third"}}, x_telegram_bot_api_secret_token="test-secret", db=self.db, ) finally: telegram_api.get_telegram_config = original_get_config telegram_api.send_telegram_message = original_send if original_handler is None: delattr(telegram_api, "handle_ai_chat_message") else: telegram_api.handle_ai_chat_message = original_handler if hasattr(telegram_api, "_TELEGRAM_CHAT_SESSIONS"): telegram_api._TELEGRAM_CHAT_SESSIONS.clear() self.assertEqual({"ok": True, "handled": "ai_chat", "action": None}, first) self.assertEqual({"ok": True, "handled": "ai_chat", "action": None}, second) self.assertEqual({"ok": True, "handled": "new_chat"}, reset) self.assertEqual({"ok": True, "handled": "ai_chat", "action": None}, third) self.assertEqual([], seen_contexts[0][1]) self.assertEqual( [ {"role": "user", "content": "first"}, {"role": "assistant", "content": "reply:first"}, ], seen_contexts[1][1], ) self.assertEqual([], seen_contexts[2][1]) self.assertIn("/new", sent_messages[2][1]) async def test_stop_command_pauses_conversation_until_new(self): sent_messages = [] handler_calls = [] original_get_config = telegram_api.get_telegram_config original_send = telegram_api.send_telegram_message original_handler = getattr(telegram_api, "handle_ai_chat_message", None) if hasattr(telegram_api, "_TELEGRAM_CHAT_SESSIONS"): telegram_api._TELEGRAM_CHAT_SESSIONS.clear() async def fake_handler(db, message, context=None, tenant_id=None): handler_calls.append(message) return telegram_api.ChatResponse(reply=f"reply:{message}") async def fake_send(chat_id, text, **kwargs): sent_messages.append((chat_id, text)) return {"success": True, "error": None, "status_code": 200} telegram_api.get_telegram_config = lambda: {"bot_token": "token", "chat_id": "798233405"} telegram_api.send_telegram_message = fake_send telegram_api.handle_ai_chat_message = fake_handler try: stop_result = await telegram_api.telegram_webhook( {"message": {"message_id": 1, "chat": {"id": 798233405}, "text": "/stop"}}, x_telegram_bot_api_secret_token="test-secret", db=self.db, ) blocked_result = await telegram_api.telegram_webhook( {"message": {"message_id": 2, "chat": {"id": 798233405}, "text": "still there?"}}, x_telegram_bot_api_secret_token="test-secret", db=self.db, ) restart_result = await telegram_api.telegram_webhook( {"message": {"message_id": 3, "chat": {"id": 798233405}, "text": "/new"}}, x_telegram_bot_api_secret_token="test-secret", db=self.db, ) resumed_result = await telegram_api.telegram_webhook( {"message": {"message_id": 4, "chat": {"id": 798233405}, "text": "back again"}}, x_telegram_bot_api_secret_token="test-secret", db=self.db, ) finally: telegram_api.get_telegram_config = original_get_config telegram_api.send_telegram_message = original_send if original_handler is None: delattr(telegram_api, "handle_ai_chat_message") else: telegram_api.handle_ai_chat_message = original_handler if hasattr(telegram_api, "_TELEGRAM_CHAT_SESSIONS"): telegram_api._TELEGRAM_CHAT_SESSIONS.clear() self.assertEqual({"ok": True, "handled": "stop_chat"}, stop_result) self.assertEqual({"ok": True, "handled": "chat_stopped"}, blocked_result) self.assertEqual({"ok": True, "handled": "new_chat"}, restart_result) self.assertEqual({"ok": True, "handled": "ai_chat", "action": None}, resumed_result) self.assertEqual(["back again"], handler_calls) self.assertIn("/new", sent_messages[0][1]) self.assertIn("/new", sent_messages[1][1]) async def test_webhook_info_masks_token_and_exposes_status(self): original_config = telegram_service.get_telegram_config original_reader = telegram_service._read_telegram_json telegram_service.get_telegram_config = lambda *args, **kwargs: {"bot_token": "123456:SECRET", "chat_id": "798233405"} telegram_service._read_telegram_json = lambda url: { "ok": True, "result": { "url": "https://example.test/api/telegram/webhook", "pending_update_count": 2, "last_error_message": "Wrong response from webhook", }, } try: result = await telegram_service.get_telegram_webhook_info() finally: telegram_service.get_telegram_config = original_config telegram_service._read_telegram_json = original_reader self.assertTrue(result["success"]) self.assertEqual("https://example.test/api/telegram/webhook", result["webhook"]["url"]) self.assertEqual(2, result["webhook"]["pending_update_count"]) self.assertEqual("Wrong response from webhook", result["webhook"]["last_error_message"]) async def test_set_webhook_rejects_non_https_url(self): result = await telegram_service.set_telegram_webhook("http://example.test/api/telegram/webhook") self.assertFalse(result["success"]) self.assertIn("HTTPS", result["error"]) if __name__ == "__main__": unittest.main()