Mikhail Putilovskij
d5ab527f5d
- sdk/mock.py: stream_message was async def (coroutine), must be async generator with yield — caused TypeError on every user message - topic_events.py: on_topic_created now skips bot-created topics (from_user.id == bot.id); cmd_new already registers them under the correct human user_id - commands.py: cmd_archive now sends "Чат архивирован." confirmation - test_topic_events.py: add bot=SimpleNamespace(id=BOT_ID) to fixture
74 lines
2.5 KiB
Python
74 lines
2.5 KiB
Python
from __future__ import annotations
|
|
|
|
import importlib
|
|
from types import SimpleNamespace
|
|
from unittest.mock import AsyncMock
|
|
|
|
import pytest
|
|
|
|
|
|
@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
|
|
|
|
|
|
BOT_ID = 9999 # distinct from any test user_id
|
|
|
|
|
|
def make_service_message(*, user_id=1, thread_id=42, topic_name="Мой чат"):
|
|
m = SimpleNamespace()
|
|
m.message_thread_id = thread_id
|
|
m.from_user = SimpleNamespace(id=user_id, full_name="Alice")
|
|
m.chat = SimpleNamespace(id=user_id)
|
|
m.forum_topic_created = SimpleNamespace(name=topic_name)
|
|
m.forum_topic_edited = SimpleNamespace(name="Новое имя")
|
|
m.forum_topic_closed = SimpleNamespace()
|
|
m.answer = AsyncMock()
|
|
m.bot = SimpleNamespace(id=BOT_ID)
|
|
return m
|
|
|
|
|
|
async def test_on_topic_created_registers_chat(fresh_db, monkeypatch):
|
|
import adapter.telegram.handlers.topic_events as mod
|
|
importlib.reload(mod)
|
|
msg = make_service_message(user_id=5, thread_id=99, topic_name="Мой чат")
|
|
await mod.on_topic_created(msg)
|
|
chat = fresh_db.get_chat(5, 99)
|
|
assert chat is not None
|
|
assert chat["chat_name"] == "Мой чат"
|
|
|
|
|
|
async def test_on_topic_edited_renames_chat(fresh_db, monkeypatch):
|
|
import adapter.telegram.handlers.topic_events as mod
|
|
importlib.reload(mod)
|
|
fresh_db.create_chat(5, 99, "Старое имя")
|
|
msg = make_service_message(user_id=5, thread_id=99)
|
|
await mod.on_topic_edited(msg)
|
|
assert fresh_db.get_chat(5, 99)["chat_name"] == "Новое имя"
|
|
|
|
|
|
async def test_on_topic_edited_unknown_chat_is_noop(fresh_db):
|
|
import adapter.telegram.handlers.topic_events as mod
|
|
importlib.reload(mod)
|
|
msg = make_service_message(user_id=5, thread_id=999)
|
|
await mod.on_topic_edited(msg) # should not raise
|
|
|
|
|
|
async def test_on_topic_closed_archives_chat(fresh_db):
|
|
import adapter.telegram.handlers.topic_events as mod
|
|
importlib.reload(mod)
|
|
fresh_db.create_chat(5, 99, "Чат #1")
|
|
msg = make_service_message(user_id=5, thread_id=99)
|
|
await mod.on_topic_closed(msg)
|
|
assert fresh_db.get_chat(5, 99)["archived_at"] is not None
|
|
|
|
|
|
async def test_on_topic_closed_unknown_chat_is_noop(fresh_db):
|
|
import adapter.telegram.handlers.topic_events as mod
|
|
importlib.reload(mod)
|
|
msg = make_service_message(user_id=5, thread_id=999)
|
|
await mod.on_topic_closed(msg) # should not raise
|