98 lines
3.5 KiB
Python
98 lines
3.5 KiB
Python
import unittest
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
from typing import Any
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
from zoneinfo import ZoneInfo
|
|
|
|
from telegram.constants import ParseMode
|
|
|
|
from assistant_bot.handlers import private_voice
|
|
from assistant_bot.services import SERVICES_KEY, ApplicationServices
|
|
from assistant_bot.telegram_utils import render_markdown_chunks
|
|
|
|
|
|
class VoiceHandlerTests(unittest.IsolatedAsyncioTestCase):
|
|
@staticmethod
|
|
def make_update_and_context(duration: int, transcript: str = "Напомни позвонить"):
|
|
status_message = SimpleNamespace(edit_text=AsyncMock())
|
|
message = SimpleNamespace(
|
|
chat_id=100,
|
|
voice=SimpleNamespace(duration=duration, file_id="voice-file-id"),
|
|
reply_text=AsyncMock(return_value=status_message),
|
|
)
|
|
update: Any = SimpleNamespace(message=message)
|
|
telegram_file = SimpleNamespace(download_to_drive=AsyncMock())
|
|
bot = SimpleNamespace(
|
|
get_file=AsyncMock(return_value=telegram_file),
|
|
send_chat_action=AsyncMock(),
|
|
)
|
|
recognizer = MagicMock()
|
|
recognizer.transcribe.return_value = transcript
|
|
context: Any = SimpleNamespace(
|
|
bot=bot,
|
|
application=SimpleNamespace(
|
|
bot_data={
|
|
SERVICES_KEY: ApplicationServices(
|
|
storage=SimpleNamespace(),
|
|
ai_client=SimpleNamespace(),
|
|
timezone=ZoneInfo("UTC"),
|
|
speech_recognizer=recognizer,
|
|
voice_max_duration_seconds=120,
|
|
assistant_password="secret",
|
|
),
|
|
}
|
|
),
|
|
)
|
|
return update, context, status_message, recognizer, telegram_file
|
|
|
|
async def test_rejects_voice_message_over_duration_limit(self) -> None:
|
|
update, context, _status, recognizer, _telegram_file = (
|
|
self.make_update_and_context(duration=121)
|
|
)
|
|
|
|
await private_voice(update, context)
|
|
|
|
rendered = render_markdown_chunks(
|
|
"🎙️ **Голосовое сообщение слишком длинное.** Максимум: `120` сек."
|
|
)[0][0]
|
|
update.message.reply_text.assert_awaited_once_with(
|
|
rendered,
|
|
parse_mode=ParseMode.MARKDOWN_V2,
|
|
)
|
|
recognizer.transcribe.assert_not_called()
|
|
|
|
async def test_transcribes_voice_and_passes_text_to_agent(self) -> None:
|
|
update, context, status, recognizer, telegram_file = (
|
|
self.make_update_and_context(duration=10)
|
|
)
|
|
|
|
with patch(
|
|
"assistant_bot.handlers.run_agent_prompt",
|
|
new_callable=AsyncMock,
|
|
) as run_agent:
|
|
await private_voice(update, context)
|
|
|
|
recognizer.transcribe.assert_called_once()
|
|
temporary_path = Path(recognizer.transcribe.call_args.args[0])
|
|
self.assertFalse(temporary_path.exists())
|
|
telegram_file.download_to_drive.assert_awaited_once_with(
|
|
custom_path=temporary_path
|
|
)
|
|
rendered = render_markdown_chunks(
|
|
"🎙️ **Распознано:** Напомни позвонить"
|
|
)[0][0]
|
|
status.edit_text.assert_awaited_once_with(
|
|
rendered,
|
|
parse_mode=ParseMode.MARKDOWN_V2,
|
|
)
|
|
run_agent.assert_awaited_once_with(
|
|
update,
|
|
context,
|
|
"Напомни позвонить",
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|