Store Support Bot in Python: Source Code

1from dataclasses import replace2from pathlib import Path3from types import SimpleNamespace45from store_assistant.assistant import StoreAssistant6from store_assistant.bot import SLOW_DOWN, TOO_LONG, SupportBot7from store_assistant.config import Settings8from store_assistant.memory import ConversationMemory9from store_assistant.tools import ToolExecutor1011from .conftest import FakeClient, completion, tool_call1213SETTINGS = Settings(14    devup_api_key="test",15    devup_base_url="https://api.example.test/v1",16    model="test-model",17    telegram_bot_token="123456:TEST",18    admin_chat_id=999,19    db_path=Path("unused.db"),20    catalog_path=Path("unused.json"),21    history_messages=12,22    rate_limit_per_minute=2,23    max_input_chars=50,24    max_tool_rounds=4,25    max_output_tokens=256,26    request_timeout_s=10,27)282930class FakeMessage:31    def __init__(self, text: str) -> None:32        self.text = text33        self.replies: list[str] = []3435    async def reply_text(self, text: str, **_: object) -> None:36        self.replies.append(text)373839class FakeBot:40    def __init__(self) -> None:41        self.sent: list[dict] = []42        self.actions = 04344    async def send_chat_action(self, **_: object) -> None:45        self.actions += 14647    async def send_message(self, **kwargs: object) -> None:48        self.sent.append(kwargs)495051def make_update(text: str, chat_id: int = 10, username: str = "yacine"):52    message = FakeMessage(text)53    update = SimpleNamespace(54        effective_message=message,55        effective_chat=SimpleNamespace(id=chat_id),56        effective_user=SimpleNamespace(id=chat_id, username=username),57    )58    return update, message596061def make_bot(store, responses, settings=SETTINGS):62    client = FakeClient(responses)63    assistant = StoreAssistant(client, "test-model", ToolExecutor(store), ConversationMemory(12), "SYSTEM")64    return SupportBot(settings, assistant, "Nour Shop"), client656667def test_application_registers_handlers(store) -> None:68    bot, _ = make_bot(store, [])69    app = bot.build_application()70    assert sum(len(group) for group in app.handlers.values()) == 6717273async def test_text_message_gets_reply(store) -> None:74    bot, _ = make_bot(store, [completion("Salam Yacine!")])75    update, message = make_update("salam")76    context = SimpleNamespace(bot=FakeBot())77    await bot.on_text(update, context)78    assert message.replies == ["Salam Yacine!"]798081async def test_id_command(store) -> None:82    bot, _ = make_bot(store, [])83    update, message = make_update("/id", chat_id=555)84    await bot.on_id(update, SimpleNamespace(bot=FakeBot()))85    assert message.replies == ["Chat ID: 555"]868788async def test_long_message_is_rejected_without_api_call(store) -> None:89    bot, client = make_bot(store, [])90    update, message = make_update("x" * 51)91    await bot.on_text(update, SimpleNamespace(bot=FakeBot()))92    assert message.replies == [TOO_LONG]93    assert client.requests == []949596async def test_rate_limit(store) -> None:97    bot, client = make_bot(store, [completion("1"), completion("2")])98    context = SimpleNamespace(bot=FakeBot())99    for _ in range(3):100        update, message = make_update("salam")101        await bot.on_text(update, context)102    assert message.replies == [SLOW_DOWN.format(seconds=60)]103    assert len(client.requests) == 2104105106async def test_ticket_notifies_admin(store) -> None:107    bot, _ = make_bot(108        store,109        [110            completion(tool_calls=[tool_call("t", "request_human_agent", {"reason": "complaint"})]),111            completion("Un conseiller va vous répondre."),112        ],113    )114    fake_bot = FakeBot()115    update, _ = make_update("<b>je veux parler à quelqu'un</b>", username="sara")116    await bot.on_text(update, SimpleNamespace(bot=fake_bot))117    assert len(fake_bot.sent) == 1118    notice = fake_bot.sent[0]119    assert notice["chat_id"] == 999120    assert "@sara" in notice["text"]121    assert "&lt;b&gt;" in notice["text"]  # customer text is escaped before HTML formatting122123124async def test_ticket_without_admin_chat(store) -> None:125    settings = replace(SETTINGS, admin_chat_id=None)126    bot, _ = make_bot(127        store,128        [completion(tool_calls=[tool_call("t", "request_human_agent", {"reason": "x"})]), completion("ok")],129        settings,130    )131    fake_bot = FakeBot()132    update, message = make_update("agent please")133    await bot.on_text(update, SimpleNamespace(bot=fake_bot))134    assert fake_bot.sent == []135    assert message.replies == ["ok"]