120 lines
3.3 KiB
Python
120 lines
3.3 KiB
Python
"""Tests for steward.bot.telegram (handler logic)."""
|
|
|
|
from unittest.mock import AsyncMock, MagicMock
|
|
|
|
import pytest
|
|
from telegram import Message, Update, User
|
|
from telegram.ext import CallbackContext
|
|
|
|
from steward.bot.telegram import (
|
|
_history,
|
|
_is_allowed,
|
|
clear_handler,
|
|
message_handler,
|
|
start_handler,
|
|
)
|
|
from steward.config import Settings
|
|
from steward.llm.client import LLMClient
|
|
|
|
|
|
def _make_settings(**kwargs) -> Settings:
|
|
defaults = dict(
|
|
telegram_bot_token="test-token",
|
|
openai_api_key="test-key",
|
|
)
|
|
defaults.update(kwargs)
|
|
return Settings(**defaults)
|
|
|
|
|
|
def _make_update(user_id: int = 12345, text: str = "hello") -> Update:
|
|
user = MagicMock(spec=User)
|
|
user.id = user_id
|
|
|
|
message = MagicMock(spec=Message)
|
|
message.text = text
|
|
message.reply_text = AsyncMock()
|
|
|
|
update = MagicMock(spec=Update)
|
|
update.effective_user = user
|
|
update.message = message
|
|
return update
|
|
|
|
|
|
def _make_context(settings: Settings, llm: LLMClient | None = None) -> CallbackContext: # type: ignore[type-arg]
|
|
ctx = MagicMock(spec=CallbackContext)
|
|
ctx.bot_data = {"settings": settings, "llm": llm}
|
|
return ctx
|
|
|
|
|
|
class TestIsAllowed:
|
|
def test_no_allowlist_allows_everyone(self):
|
|
settings = _make_settings(telegram_allowed_user_ids=[])
|
|
assert _is_allowed(999, settings) is True
|
|
|
|
def test_allowlist_accepts_known_user(self):
|
|
settings = _make_settings(telegram_allowed_user_ids=[1, 2, 3])
|
|
assert _is_allowed(2, settings) is True
|
|
|
|
def test_allowlist_rejects_unknown_user(self):
|
|
settings = _make_settings(telegram_allowed_user_ids=[1, 2, 3])
|
|
assert _is_allowed(999, settings) is False
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_start_handler_replies(monkeypatch):
|
|
settings = _make_settings()
|
|
update = _make_update()
|
|
ctx = _make_context(settings)
|
|
|
|
await start_handler(update, ctx)
|
|
|
|
update.message.reply_text.assert_awaited_once()
|
|
text = update.message.reply_text.call_args.args[0]
|
|
assert "Steward" in text
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_start_handler_ignores_disallowed_user():
|
|
settings = _make_settings(telegram_allowed_user_ids=[9999])
|
|
update = _make_update(user_id=1111)
|
|
ctx = _make_context(settings)
|
|
|
|
await start_handler(update, ctx)
|
|
|
|
update.message.reply_text.assert_not_awaited()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_clear_handler_clears_history():
|
|
settings = _make_settings()
|
|
user_id = 42
|
|
_history[user_id] = [{"role": "user", "content": "old msg"}]
|
|
|
|
update = _make_update(user_id=user_id)
|
|
ctx = _make_context(settings)
|
|
|
|
await clear_handler(update, ctx)
|
|
|
|
assert _history[user_id] == []
|
|
update.message.reply_text.assert_awaited_once()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_message_handler_calls_llm_and_replies():
|
|
settings = _make_settings()
|
|
mock_llm = MagicMock(spec=LLMClient)
|
|
mock_llm.chat = AsyncMock(return_value="LLM response")
|
|
|
|
update = _make_update(user_id=77, text="What is the weather?")
|
|
ctx = _make_context(settings, llm=mock_llm)
|
|
|
|
_history[77].clear()
|
|
await message_handler(update, ctx)
|
|
|
|
mock_llm.chat.assert_awaited_once()
|
|
update.message.reply_text.assert_awaited_once()
|
|
# History should now contain the user/assistant turn
|
|
assert len(_history[77]) == 2
|
|
assert _history[77][0]["role"] == "user"
|
|
assert _history[77][1]["role"] == "assistant"
|