ElebobaandClaude Opus 5.5
bfb0c9e7bb
CI / ruff · pytest (pull_request) Successful in 2m45s
Refs #108 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
350 lines
13 KiB
Python
350 lines
13 KiB
Python
"""Telegram channel (#11): deep-link binding, routing to the engine, rendering, outbox — no network, SQLite DBs."""
|
|
|
|
import logging
|
|
from collections.abc import AsyncGenerator, AsyncIterator
|
|
from datetime import UTC, datetime
|
|
from typing import Any
|
|
|
|
import pytest
|
|
from aiogram import Bot, Dispatcher
|
|
from aiogram.client.session.base import BaseSession
|
|
from aiogram.methods import SendMessage, TelegramMethod
|
|
from aiogram.types import CallbackQuery, Chat, Update, User
|
|
from aiogram.types import Message as TgMessage
|
|
|
|
from app.channels import telegram
|
|
from app.channels.telegram import (
|
|
CALLBACK_DATA_MAX,
|
|
ChatDirectory,
|
|
TelegramChannel,
|
|
TelegramOutbox,
|
|
load_texts,
|
|
render_markup,
|
|
split_text,
|
|
)
|
|
from app.config import get_settings
|
|
from app.db import identity_repo, session
|
|
from app.db.bootstrap import init_db
|
|
from app.dialog.contracts import Button, Event, Reply
|
|
from app.integrations.base import MemoryOutbox, VerifiedPerson, integrations
|
|
|
|
T0 = datetime(2026, 10, 3, 9, 0, tzinfo=UTC)
|
|
TOKEN = "a" * 32
|
|
TOKEN2 = "b" * 32
|
|
CHAT = 1001
|
|
SECRET_TEXT = "меня зовут Андрей, мне плохо"
|
|
|
|
|
|
class FakeSender:
|
|
def __init__(self) -> None:
|
|
self.sent: list[tuple[int, str, object]] = []
|
|
|
|
async def send_message(
|
|
self, chat_id: int, text: str, parse_mode: str | None = None, reply_markup: object = None
|
|
) -> object:
|
|
self.sent.append((chat_id, text, reply_markup))
|
|
return None
|
|
|
|
@property
|
|
def texts(self) -> list[str]:
|
|
return [t for _, t, _ in self.sent]
|
|
|
|
|
|
class RecordingEngine:
|
|
def __init__(self, fail: bool = False) -> None:
|
|
self.calls: list[tuple[str, Event]] = []
|
|
self.fail = fail
|
|
|
|
async def handle(self, token: str, event: Event) -> list[Reply]:
|
|
self.calls.append((token, event))
|
|
if self.fail:
|
|
raise RuntimeError("boom")
|
|
return [Reply(text=f"ok:{event.type}", buttons=[[Button(label="Да", data="yes")]])]
|
|
|
|
|
|
class FakeIdentity:
|
|
def __init__(self, codes: dict[str, str]) -> None:
|
|
self.codes = codes
|
|
|
|
async def verify(self, auth_code: str) -> VerifiedPerson | None:
|
|
token = self.codes.get(auth_code)
|
|
return VerifiedPerson(token=token, doctor_id="doc-1") if token else None
|
|
|
|
|
|
@pytest.fixture
|
|
async def dbs() -> AsyncIterator[None]:
|
|
await session.configure(identity_url="sqlite+aiosqlite:///:memory:", clinical_url="sqlite+aiosqlite:///:memory:")
|
|
await init_db()
|
|
async with session.get_identity_session() as s:
|
|
for token, snils in ((TOKEN, "000-000-000 01"), (TOKEN2, "000-000-000 02")):
|
|
await identity_repo.create_patient(
|
|
s, token=token, display_name="Синтетика", snils=snils, clinic_id="c1", doctor_id="doc-1", now=T0
|
|
)
|
|
yield
|
|
await session.configure()
|
|
|
|
|
|
@pytest.fixture
|
|
def engine(monkeypatch: pytest.MonkeyPatch) -> RecordingEngine:
|
|
eng = RecordingEngine()
|
|
monkeypatch.setattr(telegram, "get_engine", lambda: eng)
|
|
return eng
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def identity(monkeypatch: pytest.MonkeyPatch) -> FakeIdentity:
|
|
fake = FakeIdentity({"DEMO-0001": TOKEN, "DEMO-0002": TOKEN2, "GHOST": "c" * 32})
|
|
monkeypatch.setattr(integrations, "identity", fake)
|
|
return fake
|
|
|
|
|
|
# ----------------------------------------------------------------- rendering
|
|
|
|
|
|
def test_render_markup_rows_and_empty() -> None:
|
|
reply = Reply(
|
|
text="?", buttons=[[Button(label="1", data="a"), Button(label="2", data="b")], [Button(label="3", data="c")]]
|
|
)
|
|
markup = render_markup(reply)
|
|
assert markup is not None
|
|
assert [[b.callback_data for b in row] for row in markup.inline_keyboard] == [["a", "b"], ["c"]]
|
|
assert render_markup(Reply(text="x")) is None
|
|
|
|
|
|
def test_render_markup_drops_too_long_data() -> None:
|
|
reply = Reply(
|
|
text="?", buttons=[[Button(label="ok", data="x"), Button(label="long", data="я" * CALLBACK_DATA_MAX)]]
|
|
)
|
|
markup = render_markup(reply)
|
|
assert markup is not None
|
|
assert [b.callback_data for b in markup.inline_keyboard[0]] == ["x"]
|
|
|
|
|
|
def test_split_text_respects_limit_and_lines() -> None:
|
|
text = "\n".join(f"строка {i}" for i in range(50))
|
|
parts = split_text(text, limit=100)
|
|
assert all(len(p) <= 100 for p in parts)
|
|
assert "\n".join(parts).split("\n") == text.split("\n")
|
|
assert split_text("коротко") == ["коротко"]
|
|
assert all(len(p) <= 10 for p in split_text("x" * 35, limit=10))
|
|
|
|
|
|
def test_texts_load() -> None:
|
|
texts = load_texts()
|
|
assert texts.not_connected and texts.code_invalid and texts.error and texts.unsupported
|
|
|
|
|
|
# ----------------------------------------------------------------- channel
|
|
|
|
|
|
async def test_start_with_code_binds_chat_and_calls_engine(dbs: None, engine: RecordingEngine) -> None:
|
|
sender = FakeSender()
|
|
channel = TelegramChannel(sender)
|
|
await channel.on_start(CHAT, "DEMO-0001")
|
|
|
|
assert engine.calls == [(TOKEN, Event(type="start", params={"code": "DEMO-0001"}))]
|
|
assert sender.sent[0][1] == "ok:start"
|
|
assert sender.sent[0][2] is not None # buttons rendered
|
|
async with session.get_identity_session() as s:
|
|
patient = await identity_repo.get_patient(s, TOKEN)
|
|
assert patient is not None and patient.tg_chat_id == CHAT
|
|
|
|
|
|
async def test_invalid_code_does_not_reach_engine(dbs: None, engine: RecordingEngine) -> None:
|
|
sender = FakeSender()
|
|
await TelegramChannel(sender).on_start(CHAT, "WRONG")
|
|
assert engine.calls == []
|
|
assert sender.texts == [load_texts().code_invalid]
|
|
|
|
|
|
async def test_verified_token_without_identity_record(dbs: None, engine: RecordingEngine) -> None:
|
|
sender = FakeSender()
|
|
await TelegramChannel(sender).on_start(CHAT, "GHOST")
|
|
assert engine.calls == []
|
|
assert sender.texts == [load_texts().code_invalid]
|
|
|
|
|
|
async def test_unbound_chat_gets_not_connected(
|
|
dbs: None, engine: RecordingEngine, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
monkeypatch.setattr(get_settings(), "demo_clock", False)
|
|
sender = FakeSender()
|
|
channel = TelegramChannel(sender)
|
|
await channel.on_start(CHAT, None)
|
|
await channel.on_text(CHAT, "привет")
|
|
await channel.on_button(CHAT, "yes")
|
|
assert engine.calls == []
|
|
assert sender.texts == [load_texts().not_connected] * 3
|
|
|
|
|
|
async def test_bound_chat_routes_text_button_and_plain_start(dbs: None, engine: RecordingEngine) -> None:
|
|
sender = FakeSender()
|
|
channel = TelegramChannel(sender)
|
|
await channel.on_start(CHAT, "DEMO-0001")
|
|
# Fresh directory: binding must come from the identity DB, not the cache.
|
|
channel = TelegramChannel(sender, ChatDirectory())
|
|
await channel.on_text(CHAT, "плохо спал")
|
|
await channel.on_button(CHAT, "yes")
|
|
await channel.on_start(CHAT, None)
|
|
await channel.on_text(CHAT, None) # sticker / photo
|
|
assert [e for _, e in engine.calls[1:]] == [
|
|
Event(type="text", text="плохо спал"),
|
|
Event(type="button", button="yes"),
|
|
Event(type="start"),
|
|
]
|
|
assert all(tok == TOKEN for tok, _ in engine.calls)
|
|
assert sender.texts[-1] == load_texts().unsupported
|
|
|
|
|
|
async def test_rebinding_chat_to_another_patient(dbs: None, engine: RecordingEngine) -> None:
|
|
channel = TelegramChannel(FakeSender())
|
|
await channel.on_start(CHAT, "DEMO-0001")
|
|
await channel.on_start(CHAT, "DEMO-0002")
|
|
await channel.on_text(CHAT, "привет")
|
|
assert engine.calls[-1][0] == TOKEN2
|
|
assert await channel.directory.chat_for(TOKEN) is None
|
|
async with session.get_identity_session() as s:
|
|
first = await identity_repo.get_patient(s, TOKEN)
|
|
assert first is not None and first.tg_chat_id is None
|
|
|
|
|
|
async def test_engine_failure_sends_error_without_text_in_logs(
|
|
dbs: None, monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture
|
|
) -> None:
|
|
eng = RecordingEngine()
|
|
monkeypatch.setattr(telegram, "get_engine", lambda: eng)
|
|
sender = FakeSender()
|
|
channel = TelegramChannel(sender)
|
|
await channel.on_start(CHAT, "DEMO-0001")
|
|
eng.fail = True
|
|
with caplog.at_level(logging.DEBUG, logger="app.channels.telegram"):
|
|
await channel.on_text(CHAT, SECRET_TEXT)
|
|
assert sender.texts[-1] == load_texts().error
|
|
logged = "\n".join(r.getMessage() for r in caplog.records)
|
|
assert SECRET_TEXT not in logged and str(CHAT) not in logged and TOKEN not in logged
|
|
|
|
|
|
async def test_crisis_text_sent_verbatim(dbs: None, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
crisis = "Позвоните 8-800-2000-122 *сейчас* <b>не</b> _одни_"
|
|
|
|
class CrisisEngine:
|
|
async def handle(self, token: str, event: Event) -> list[Reply]:
|
|
return [Reply(text=crisis, kind="crisis")]
|
|
|
|
monkeypatch.setattr(telegram, "get_engine", CrisisEngine)
|
|
sender = FakeSender()
|
|
await TelegramChannel(sender).on_start(CHAT, "DEMO-0001")
|
|
assert sender.texts == [crisis]
|
|
|
|
|
|
# ----------------------------------------------------------------- outbox
|
|
|
|
|
|
async def test_outbox_pushes_to_bound_chat_and_falls_back(dbs: None, engine: RecordingEngine) -> None:
|
|
sender = FakeSender()
|
|
channel = TelegramChannel(sender)
|
|
await channel.on_start(CHAT, "DEMO-0001")
|
|
sender.sent.clear()
|
|
|
|
fallback = MemoryOutbox()
|
|
outbox = TelegramOutbox(sender, ChatDirectory(), fallback=fallback)
|
|
await outbox.push(TOKEN, [Reply(text="Время принять сертралин")])
|
|
await outbox.push(TOKEN2, [Reply(text="Как вы сегодня?")])
|
|
|
|
assert sender.sent == [(CHAT, "Время принять сертралин", None)]
|
|
assert [r.text for r in fallback.drain(TOKEN2)] == ["Как вы сегодня?"]
|
|
|
|
|
|
async def test_start_without_code_on_demo_stand_binds_a_demo_copy(
|
|
dbs: None, engine: RecordingEngine, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
monkeypatch.setattr(get_settings(), "demo_clock", True)
|
|
sender = FakeSender()
|
|
channel = TelegramChannel(sender)
|
|
await channel.on_start(CHAT, None)
|
|
token = await channel.directory.token_for(CHAT)
|
|
assert token is not None and token.startswith("tok-andrey-")
|
|
assert [c[1].type for c in engine.calls] == ["start"]
|
|
|
|
|
|
async def test_start_is_noop_without_token(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
monkeypatch.setattr(telegram, "get_settings", lambda: type("S", (), {"tg_bot_token": ""})())
|
|
before = integrations.outbox
|
|
await telegram.start()
|
|
assert telegram._task is None
|
|
assert integrations.outbox is before
|
|
await telegram.stop()
|
|
|
|
|
|
async def test_start_stop_swaps_outbox(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
monkeypatch.setattr(
|
|
telegram, "get_settings", lambda: type("S", (), {"tg_bot_token": "123:fake", "tg_proxy_url": ""})()
|
|
)
|
|
|
|
async def fake_polling(self: object, *bots: object, **kwargs: object) -> None:
|
|
import asyncio
|
|
|
|
await asyncio.Event().wait()
|
|
|
|
monkeypatch.setattr(telegram.Dispatcher, "start_polling", fake_polling)
|
|
before = integrations.outbox
|
|
await telegram.start()
|
|
assert isinstance(integrations.outbox, TelegramOutbox)
|
|
await telegram.stop()
|
|
assert integrations.outbox is before
|
|
assert telegram._task is None
|
|
|
|
|
|
# ----------------------------------------------------------------- aiogram wiring (fake HTTP session)
|
|
|
|
|
|
class FakeSession(BaseSession):
|
|
def __init__(self) -> None:
|
|
super().__init__()
|
|
self.calls: list[TelegramMethod[Any]] = []
|
|
|
|
async def make_request(self, bot: Bot, method: TelegramMethod[Any], timeout: int | None = None) -> Any: # noqa: ASYNC109 — aiogram signature
|
|
self.calls.append(method)
|
|
if isinstance(method, SendMessage):
|
|
return TgMessage(message_id=1, date=T0, chat=Chat(id=method.chat_id, type="private"), text=method.text)
|
|
return True
|
|
|
|
async def close(self) -> None:
|
|
pass
|
|
|
|
async def stream_content(self, *args: Any, **kwargs: Any) -> AsyncGenerator[bytes, None]:
|
|
yield b""
|
|
|
|
|
|
def _update(update_id: int, **kwargs: Any) -> Update:
|
|
return Update(update_id=update_id, **kwargs)
|
|
|
|
|
|
async def test_router_handles_real_updates(dbs: None, engine: RecordingEngine) -> None:
|
|
fake = FakeSession()
|
|
bot = Bot("123:fake", session=fake)
|
|
channel = TelegramChannel(bot)
|
|
dp = Dispatcher()
|
|
dp.include_router(telegram.build_router(channel))
|
|
chat = Chat(id=CHAT, type="private")
|
|
user = User(id=CHAT, is_bot=False, first_name="Синтетика")
|
|
|
|
start = TgMessage(message_id=10, date=T0, chat=chat, from_user=user, text="/start DEMO-0001")
|
|
await dp.feed_update(bot, _update(1, message=start))
|
|
text = TgMessage(message_id=11, date=T0, chat=chat, from_user=user, text="плохо спал")
|
|
await dp.feed_update(bot, _update(2, message=text))
|
|
shown = TgMessage(message_id=12, date=T0, chat=chat, text="?")
|
|
query = CallbackQuery(id="q1", from_user=user, chat_instance="ci", message=shown, data="yes")
|
|
await dp.feed_update(bot, _update(3, callback_query=query))
|
|
|
|
assert [e for _, e in engine.calls] == [
|
|
Event(type="start", params={"code": "DEMO-0001"}),
|
|
Event(type="text", text="плохо спал"),
|
|
Event(type="button", button="yes"),
|
|
]
|
|
kinds = [type(m).__name__ for m in fake.calls]
|
|
assert kinds.count("SendMessage") == 3
|
|
assert "SendChatAction" in kinds and "AnswerCallbackQuery" in kinds and "EditMessageReplyMarkup" in kinds
|
|
sent = [m for m in fake.calls if isinstance(m, SendMessage)]
|
|
assert all(m.parse_mode is None for m in sent) # crisis scripts must reach the patient verbatim
|