tests/test_progress.py - master - artyuhovs/AIFeedBackTrainingBot

tests/test_progress.py
170 строк6 KB
Codex
Добавить прозрачный медиаконвейер в версии 0.8.1
19 июл 2026, 20:24
19 июл 2026, 20:249c5c8db
from __future__ import annotations

import asyncio

from aiogram.exceptions import TelegramRetryAfter

from app.bot.progress import TelegramProgressReporter


class FakeStatusMessage:
    def __init__(self) -> None:
        self.text = ""
        self.edits: list[str] = []

    async def edit_text(self, text: str) -> None:
        self.text = text
        self.edits.append(text)


class RetryOnceStatusMessage(FakeStatusMessage):
    def __init__(self) -> None:
        super().__init__()
        self.attempts = 0

    async def edit_text(self, text: str) -> None:
        self.attempts += 1
        if self.attempts == 1:
            raise TelegramRetryAfter(method=None, message="retry later", retry_after=3)
        await super().edit_text(text)


class FakeSender:
    def __init__(self, status_message: FakeStatusMessage | None = None) -> None:
        self.status_message = status_message or FakeStatusMessage()
        self.sent: list[str] = []

    async def __call__(self, text: str) -> FakeStatusMessage:
        self.sent.append(text)
        self.status_message.text = text
        return self.status_message


def test_progress_reporter_marks_completed_steps() -> None:
    async def run() -> tuple[FakeSender, FakeStatusMessage]:
        sender = FakeSender()
        reporter = TelegramProgressReporter(sender, min_edit_interval_seconds=0)

        await reporter("Кружок получен. Скачиваю файл.")
        await reporter("Файл скачан. Готовлю аудио для распознавания.")
        await reporter.complete("Готово. Расшифровка отправлена в ответ на ваше сообщение.")

        return sender, sender.status_message

    sender, status_message = asyncio.run(run())

    assert sender.sent == ["Кружок получен. Скачиваю файл."]
    assert status_message.text == (
        "✔️ Кружок получен. Скачиваю файл.\n"
        "✔️ Файл скачан. Готовлю аудио для распознавания.\n"
        "✔️ Готово. Расшифровка отправлена в ответ на ваше сообщение."
    )
    assert len(status_message.edits) == 2


def test_progress_reporter_replaces_queue_position_without_appending_steps() -> None:
    async def run() -> FakeStatusMessage:
        sender = FakeSender()
        reporter = TelegramProgressReporter(sender, min_edit_interval_seconds=0, max_edits=0)

        await reporter.update_current("Кружок получен. Место в очереди на скачивание: 3.")
        await reporter.update_current("Кружок получен. Место в очереди на скачивание: 2.")
        await reporter.update_current("Кружок получен. Скачиваю файл.")

        return sender.status_message

    status_message = asyncio.run(run())

    assert status_message.text == "Кружок получен. Скачиваю файл."
    assert len(status_message.edits) == 2


def test_queue_position_updates_do_not_consume_regular_edit_budget() -> None:
    async def run() -> FakeStatusMessage:
        sender = FakeSender()
        reporter = TelegramProgressReporter(sender, min_edit_interval_seconds=0, max_edits=8)

        for position in range(12, 0, -1):
            await reporter.update_current(f"Место в очереди на распознавание: {position}.")

        await reporter("Распознаю речь. Это может занять несколько минут.")
        return sender.status_message

    status_message = asyncio.run(run())

    assert status_message.text.endswith("Распознаю речь. Это может занять несколько минут.")


def test_progress_reporter_appends_unmarked_multiline_detail() -> None:
    async def run() -> FakeStatusMessage:
        sender = FakeSender()
        reporter = TelegramProgressReporter(sender, min_edit_interval_seconds=0)

        await reporter("Кружок получен.")
        await reporter("Распознавание завершено.")
        await reporter.complete(
            "Готово. Ответ сохранён в учебном задании.",
            detail_text="Распознанный ответ: Первая строка.\n\nВторая строка.",
        )

        return sender.status_message

    status_message = asyncio.run(run())

    assert status_message.text == (
        "✔️ Кружок получен.\n"
        "✔️ Распознавание завершено.\n"
        "✔️ Готово. Ответ сохранён в учебном задании.\n\n"
        "Распознанный ответ: Первая строка.\n\nВторая строка."
    )


def test_progress_reporter_batches_fast_updates_into_one_final_edit() -> None:
    clock = {"now": 100.0}

    async def fake_sleep(delay: float) -> None:
        clock["now"] += delay

    async def run() -> FakeStatusMessage:
        sender = FakeSender()
        reporter = TelegramProgressReporter(
            sender,
            min_edit_interval_seconds=10,
            sleep=fake_sleep,
            monotonic=lambda: clock["now"],
        )

        await reporter("Первый шаг.")
        await reporter("Второй шаг.")
        await reporter("Третий шаг.")
        await reporter.complete("Готово.")
        await asyncio.sleep(0)

        return sender.status_message

    status_message = asyncio.run(run())

    assert status_message.edits == ["✔️ Первый шаг.\n✔️ Второй шаг.\n✔️ Третий шаг.\n✔️ Готово."]


def test_progress_reporter_respects_retry_after_on_final_edit() -> None:
    slept: list[float] = []

    async def fake_sleep(delay: float) -> None:
        slept.append(delay)

    async def run() -> RetryOnceStatusMessage:
        status_message = RetryOnceStatusMessage()
        sender = FakeSender(status_message)
        reporter = TelegramProgressReporter(sender, min_edit_interval_seconds=0, sleep=fake_sleep)

        await reporter("Оцениваю ОС.")
        await reporter.complete("Готово.")

        return status_message

    status_message = asyncio.run(run())

    assert slept == [3.0]
    assert status_message.attempts == 2
    assert status_message.text == "✔️ Оцениваю ОС.\n✔️ Готово."