Mikhail Putilovskij
8901e60f6a
- commands.py: try/except TelegramBadRequest around all Bot API calls (#2); /new handles "topics limit" with user-friendly message (#4) - start.py: isolate _check_and_prune_stale_topics with try/except Exception (#3) - message.py: asyncio.timeout(30) around stream_message; handle TimeoutError (#6) - db.py: add idx_chats_user_id index in init_db() (#7) - settings.py: remove dead active_chat_id variable (#8) - tests: add test_message.py (stream error/success); add 2 tests in test_commands.py (topics limit, /archive in General topic)
102 lines
3.6 KiB
Python
102 lines
3.6 KiB
Python
from __future__ import annotations
|
|
|
|
import importlib
|
|
from types import SimpleNamespace
|
|
from unittest.mock import AsyncMock, MagicMock
|
|
|
|
import pytest
|
|
from aiogram.exceptions import TelegramBadRequest
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def fresh_db(tmp_path, monkeypatch):
|
|
monkeypatch.setenv("DB_PATH", str(tmp_path / "test.db"))
|
|
import adapter.telegram.db as db_mod
|
|
importlib.reload(db_mod)
|
|
db_mod.init_db()
|
|
return db_mod
|
|
|
|
|
|
def make_message(*, user_id=1, thread_id=42, chat_id=100, text="/new"):
|
|
m = SimpleNamespace()
|
|
m.from_user = SimpleNamespace(id=user_id, full_name="Alice")
|
|
m.message_thread_id = thread_id
|
|
m.chat = SimpleNamespace(id=chat_id)
|
|
m.text = text
|
|
m.answer = AsyncMock()
|
|
m.reply = AsyncMock()
|
|
m.bot = MagicMock()
|
|
m.bot.create_forum_topic = AsyncMock(
|
|
return_value=SimpleNamespace(message_thread_id=200)
|
|
)
|
|
m.bot.close_forum_topic = AsyncMock()
|
|
m.bot.edit_forum_topic = AsyncMock()
|
|
m.bot.send_message = AsyncMock()
|
|
return m
|
|
|
|
|
|
async def test_cmd_new_creates_topic(fresh_db, monkeypatch):
|
|
import adapter.telegram.handlers.commands as mod
|
|
importlib.reload(mod)
|
|
fresh_db.create_chat(1, 42, "Чат #1")
|
|
msg = make_message(user_id=1, thread_id=42, chat_id=100)
|
|
await mod.cmd_new(msg)
|
|
msg.bot.create_forum_topic.assert_called_once()
|
|
call_kwargs = str(msg.bot.create_forum_topic.call_args)
|
|
assert "Чат #2" in call_kwargs
|
|
assert fresh_db.get_chat(1, 200) is not None
|
|
|
|
|
|
async def test_cmd_archive_closes_and_archives(fresh_db, monkeypatch):
|
|
import adapter.telegram.handlers.commands as mod
|
|
importlib.reload(mod)
|
|
fresh_db.create_chat(1, 42, "Чат #1")
|
|
msg = make_message(user_id=1, thread_id=42, chat_id=100)
|
|
await mod.cmd_archive(msg)
|
|
msg.bot.close_forum_topic.assert_called_once_with(chat_id=100, message_thread_id=42)
|
|
assert fresh_db.get_chat(1, 42)["archived_at"] is not None
|
|
|
|
|
|
async def test_cmd_archive_unknown_topic_replies_error(fresh_db, monkeypatch):
|
|
import adapter.telegram.handlers.commands as mod
|
|
importlib.reload(mod)
|
|
msg = make_message(user_id=1, thread_id=999, chat_id=100)
|
|
await mod.cmd_archive(msg)
|
|
msg.answer.assert_called_once()
|
|
|
|
|
|
async def test_cmd_rename_updates_db_and_topic(fresh_db, monkeypatch):
|
|
import adapter.telegram.handlers.commands as mod
|
|
importlib.reload(mod)
|
|
fresh_db.create_chat(1, 42, "Чат #1")
|
|
msg = make_message(user_id=1, thread_id=42, chat_id=100, text="/rename Работа")
|
|
await mod.cmd_rename(msg)
|
|
msg.bot.edit_forum_topic.assert_called_once_with(
|
|
chat_id=100, message_thread_id=42, name="Работа"
|
|
)
|
|
assert fresh_db.get_chat(1, 42)["chat_name"] == "Работа"
|
|
|
|
|
|
async def test_cmd_new_topics_limit(fresh_db):
|
|
"""When Telegram returns topics limit error, user gets a friendly message."""
|
|
import adapter.telegram.handlers.commands as mod
|
|
importlib.reload(mod)
|
|
msg = make_message(user_id=1, thread_id=42, chat_id=100)
|
|
msg.bot.create_forum_topic = AsyncMock(
|
|
side_effect=TelegramBadRequest(method=MagicMock(), message="topics limit exceeded")
|
|
)
|
|
await mod.cmd_new(msg)
|
|
msg.answer.assert_called_once()
|
|
assert "лимит" in msg.answer.call_args[0][0]
|
|
# No chat should be created
|
|
assert fresh_db.count_active_chats(1) == 0
|
|
|
|
|
|
async def test_cmd_archive_general_topic(fresh_db):
|
|
"""/archive in General topic (thread_id=None) replies with 'not found'."""
|
|
import adapter.telegram.handlers.commands as mod
|
|
importlib.reload(mod)
|
|
msg = make_message(user_id=1, thread_id=None, chat_id=100)
|
|
await mod.cmd_archive(msg)
|
|
msg.answer.assert_called_once()
|
|
msg.bot.close_forum_topic.assert_not_called()
|