satoko-chatbot/test_voice_tts.py

import asyncio
import ast
import unittest
import wave
from io import BytesIO
from pathlib import Path
from types import SimpleNamespace
from unittest.mock import AsyncMock, Mock, patch

import discord
from voice_tts import WavSource, speech_chunks, play_wav, speak
from voice_stt import STTSink


class AudioTests(unittest.IsolatedAsyncioTestCase):
    def test_speech_removes_unicode_sequences_and_custom_aliases(self):
        text = 'Salut :hap: 😊 ! Ça va ❤️ 👍🏽 👩‍💻 🇫🇷 1️⃣ :desu: <:minou:123> <a:eh:456> ?'
        self.assertEqual(list(speech_chunks(text)), ['Salut !', 'Ça va ?'])

    def test_speech_preserves_french_numbers_and_times(self):
        text = 'Écoute : rendez-vous à 12:30:45, pour 42 € et 20 %.'
        self.assertEqual(list(speech_chunks(text)), [text])

    async def test_emoji_only_response_is_silent(self):
        voice = Mock()
        with patch('voice_tts.httpx.AsyncClient') as factory:
            factory.return_value = AsyncMock()
            await speak(voice, '😊 :hap: 👩‍💻', 'http://tts')
            factory.return_value.__aenter__.return_value.stream.assert_not_called()
        voice.play.assert_not_called()

    def test_chunks_are_bounded_and_remove_discord_markup(self):
        chunks = list(speech_chunks('Bonjour ! [Lis ceci](https://example.org) <:hap:123> ' + 'mot ' * 200))
        self.assertTrue(all(0 < len(c) <= 240 for c in chunks))
        self.assertNotIn('https', ' '.join(chunks))
        self.assertNotIn('<:hap', ' '.join(chunks))
        self.assertEqual(chunks[0], 'Bonjour !')

    def wav(self):
        buf = BytesIO()
        with wave.open(buf, 'wb') as wav:
            wav.setnchannels(1)
            wav.setsampwidth(2)
            wav.setframerate(24000)
            wav.writeframes(b'\0\1' * 2400)
        return buf.getvalue()

    def test_discord_frames_are_stereo_48khz_20ms(self):
        source = WavSource(self.wav())
        frames = []
        while frame := source.read():
            frames.append(frame)
        self.assertEqual(len(frames), 5)
        self.assertTrue(all(len(frame) == 3840 for frame in frames))
        source.cleanup()

    async def test_cancel_stops_playback(self):
        voice = SimpleNamespace(play=Mock(), stop=Mock())
        task = asyncio.create_task(play_wav(voice, self.wav()))
        await asyncio.sleep(0)
        task.cancel()
        with self.assertRaises(asyncio.CancelledError):
            await task
        voice.stop.assert_called_once()

    async def test_batch_statuses_track_success_failure_and_duplicate_phrases(self):
        statuses = AsyncMock()
        reply = AsyncMock(side_effect=[None, RuntimeError('LLM unavailable')])
        sink = STTSink(reply, 'http://unused', on_batch_status=statuses)
        try:
            sink.pending = [('Alice', 'Bonjour'), ('Alice', 'Bonjour')]
            sink.pending_receipts = ['message-1', 'message-2']
            sink.pending_event.set()
            await asyncio.sleep(0.1)
            self.assertEqual(statuses.call_args_list[0].args, (['message-1', 'message-2'], 'processing'))
            self.assertEqual(statuses.call_args_list[1].args, (['message-1', 'message-2'], 'done'))
            sink.pending = [('Alice', 'Autre phrase')]
            sink.pending_receipts = ['message-3']
            sink.pending_event.set()
            await asyncio.sleep(0.1)
            self.assertEqual(statuses.call_args_list[-1].args, (['message-3'], 'error'))
        finally:
            await sink.close()

    async def test_audio_mode_does_not_publish_conversation(self):
        tree = ast.parse(Path(__file__).with_name('chatbot.py').read_text(encoding='utf-8'))
        function = next(n for n in tree.body if isinstance(n, ast.AsyncFunctionDef)
                        and n.name == 'handle_voice_command')
        http = AsyncMock()
        http.__aenter__.return_value.get.return_value = Mock()
        sink = SimpleNamespace(close=AsyncMock(), receiver_stopped=Mock(), log_diagnostic=Mock())
        factory = Mock(return_value=sink)
        voice = SimpleNamespace(sink=sink, listen=Mock(), disconnect=AsyncMock(),
                                is_connected=lambda: True, is_listening=lambda: True)
        channel = SimpleNamespace(connect=AsyncMock(return_value=voice))
        voice.channel = channel
        message = SimpleNamespace(content='!join audio',
            guild=SimpleNamespace(voice_client=None, me=SimpleNamespace(voice=None)),
            author=SimpleNamespace(voice=SimpleNamespace(channel=channel)),
            channel=SimpleNamespace(send=AsyncMock()))
        speak = AsyncMock()
        ns = dict(VOICE_ENABLED=True, STT_URL='http://stt/transcribe', TTS_URL='http://tts/v1/audio/speech',
                  TTS_VOICE='ff_siwis', voice_command_lock=asyncio.Lock(),
                  httpx=SimpleNamespace(AsyncClient=Mock(return_value=http)),
                  voice_recv=SimpleNamespace(VoiceRecvClient=object), STTSink=factory,
                  call_openai_api=AsyncMock(return_value=SimpleNamespace(choices=[
                      SimpleNamespace(message=SimpleNamespace(content='Bonjour Alice.'))])),
                  filter_message=lambda x: x, speak=speak, discord=discord, logger=Mock())
        exec(compile(ast.Module(body=[function], type_ignores=[]), '<commands>', 'exec'), ns)
        await ns['handle_voice_command'](message)
        self.assertIsNone(factory.call_args.kwargs['on_transcription'])
        message.channel.send.reset_mock()
        await factory.call_args.args[0]([(SimpleNamespace(display_name='Alice'), 'Salut')])
        speak.assert_awaited_once_with(voice, 'Bonjour Alice.', ns['TTS_URL'], 'ff_siwis')
        message.channel.send.assert_not_awaited()