from dataclasses import replace from pathlib import Path from types import SimpleNamespace from store_assistant.assistant import StoreAssistant from store_assistant.bot import SLOW_DOWN, TOO_LONG, SupportBot from store_assistant.config import Settings from store_assistant.memory import ConversationMemory from store_assistant.tools import ToolExecutor from .conftest import FakeClient, completion, tool_call SETTINGS = Settings( devup_api_key="test", devup_base_url="https://api.example.test/v1", model="test-model", telegram_bot_token="123456:TEST", admin_chat_id=999, db_path=Path("unused.db"), catalog_path=Path("unused.json"), history_messages=12, rate_limit_per_minute=2, max_input_chars=50, max_tool_rounds=4, max_output_tokens=256, request_timeout_s=10, ) class FakeMessage: def __init__(self, text: str) -> None: self.text = text self.replies: list[str] = [] async def reply_text(self, text: str, **_: object) -> None: self.replies.append(text) class FakeBot: def __init__(self) -> None: self.sent: list[dict] = [] self.actions = 0 async def send_chat_action(self, **_: object) -> None: self.actions += 1 async def send_message(self, **kwargs: object) -> None: self.sent.append(kwargs) def make_update(text: str, chat_id: int = 10, username: str = "yacine"): message = FakeMessage(text) update = SimpleNamespace( effective_message=message, effective_chat=SimpleNamespace(id=chat_id), effective_user=SimpleNamespace(id=chat_id, username=username), ) return update, message def make_bot(store, responses, settings=SETTINGS): client = FakeClient(responses) assistant = StoreAssistant(client, "test-model", ToolExecutor(store), ConversationMemory(12), "SYSTEM") return SupportBot(settings, assistant, "Nour Shop"), client def test_application_registers_handlers(store) -> None: bot, _ = make_bot(store, []) app = bot.build_application() assert sum(len(group) for group in app.handlers.values()) == 6 async def test_text_message_gets_reply(store) -> None: bot, _ = make_bot(store, [completion("Salam Yacine!")]) update, message = make_update("salam") context = SimpleNamespace(bot=FakeBot()) await bot.on_text(update, context) assert message.replies == ["Salam Yacine!"] async def test_id_command(store) -> None: bot, _ = make_bot(store, []) update, message = make_update("/id", chat_id=555) await bot.on_id(update, SimpleNamespace(bot=FakeBot())) assert message.replies == ["Chat ID: 555"] async def test_long_message_is_rejected_without_api_call(store) -> None: bot, client = make_bot(store, []) update, message = make_update("x" * 51) await bot.on_text(update, SimpleNamespace(bot=FakeBot())) assert message.replies == [TOO_LONG] assert client.requests == [] async def test_rate_limit(store) -> None: bot, client = make_bot(store, [completion("1"), completion("2")]) context = SimpleNamespace(bot=FakeBot()) for _ in range(3): update, message = make_update("salam") await bot.on_text(update, context) assert message.replies == [SLOW_DOWN.format(seconds=60)] assert len(client.requests) == 2 async def test_ticket_notifies_admin(store) -> None: bot, _ = make_bot( store, [ completion(tool_calls=[tool_call("t", "request_human_agent", {"reason": "complaint"})]), completion("Un conseiller va vous répondre."), ], ) fake_bot = FakeBot() update, _ = make_update("je veux parler à quelqu'un", username="sara") await bot.on_text(update, SimpleNamespace(bot=fake_bot)) assert len(fake_bot.sent) == 1 notice = fake_bot.sent[0] assert notice["chat_id"] == 999 assert "@sara" in notice["text"] assert "<b>" in notice["text"] # customer text is escaped before HTML formatting async def test_ticket_without_admin_chat(store) -> None: settings = replace(SETTINGS, admin_chat_id=None) bot, _ = make_bot( store, [completion(tool_calls=[tool_call("t", "request_human_agent", {"reason": "x"})]), completion("ok")], settings, ) fake_bot = FakeBot() update, message = make_update("agent please") await bot.on_text(update, SimpleNamespace(bot=fake_bot)) assert fake_bot.sent == [] assert message.replies == ["ok"]