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_short_first_chunk_preserves_all_words(self):
        text = 'Je peux te répondre rapidement, puis poursuivre avec les détails pendant la lecture du début.'
        chunks = list(speech_chunks(text, limit=80, first_limit=48))
        self.assertEqual(chunks[0], 'Je peux te répondre rapidement,')
        self.assertEqual(' '.join(chunks), text)
        self.assertLessEqual(len(chunks[0]), 48)
        self.assertTrue(all(len(chunk) <= 80 for chunk in chunks))

    async def test_next_synthesis_overlaps_playback_and_is_cancelled(self):
        second_started = asyncio.Event()
        second_cancelled = asyncio.Event()
        first_played = asyncio.Event()
        requests = []

        class Response:
            async def __aenter__(self):
                return self

            async def __aexit__(self, *args):
                pass

            def raise_for_status(self):
                pass

            async def aiter_bytes(self):
                if len(requests) == 2:
                    second_started.set()
                    try:
                        await asyncio.Event().wait()
                    finally:
                        second_cancelled.set()
                yield b'wav'

        def stream(*args, **kwargs):
            requests.append(kwargs['json']['input'])
            return Response()

        async def play(*args):
            first_played.set()
            await second_started.wait()
            await asyncio.Event().wait()

        http = AsyncMock()
        http.__aenter__.return_value.stream = Mock(side_effect=stream)
        with patch('voice_tts.httpx.AsyncClient', return_value=http), patch('voice_tts.play_wav', side_effect=play):
            task = asyncio.create_task(speak(Mock(), 'Bonjour. Voici la suite.', 'http://tts'))
            try:
                await asyncio.wait_for(second_started.wait(), 1)
                self.assertTrue(first_played.is_set())
                self.assertEqual(requests, ['Bonjour.', 'Voici la suite.'])
            finally:
                task.cancel()
                with self.assertRaises(asyncio.CancelledError):
                    await task
            self.assertTrue(second_cancelled.is_set())

    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()