PTSRaiAgent/tests/test_telegram.py

"""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