"""Tests hors Discord : python -m unittest test_voice_stt."""
import asyncio
import ast
import time
import unittest
import wave
from io import BytesIO
from pathlib import Path
from types import SimpleNamespace
from unittest.mock import AsyncMock, Mock, patch

import httpx
import discord
from voice_stt import STTSink


class VoiceTests(unittest.IsolatedAsyncioTestCase):
    async def asyncSetUp(self):
        self.reply = AsyncMock()
        self.sink = STTSink(self.reply, 'http://stt/transcribe')
        self.user = SimpleNamespace(id=1, bot=False)
        self.pcm = b'\x00\x10' * (48000 * 2 // 2)  # 0,5 s stéréo

    async def asyncTearDown(self):
        await self.sink.close()

    async def test_silence_and_separate_speakers(self):
        self.sink.worker_task.cancel()
        with self.sink.lock:
            self.sink.add_pcm(self.user, bytes(len(self.pcm)), time.monotonic())
            self.assertFalse(self.sink.buffers)
            self.sink.add_pcm(self.user, self.pcm, time.monotonic() - 2)
            self.sink.add_pcm(SimpleNamespace(id=2), self.pcm, time.monotonic())
        await asyncio.sleep(0.15)
        user, pcm = self.sink.queue.get_nowait()
        self.sink.queue.task_done()
        self.assertEqual(user.id, 1)
        self.assertEqual(pcm, self.pcm)
        self.assertIn(2, self.sink.buffers)

    async def test_audio_is_bounded_and_short_noise_ignored(self):
        self.sink.worker_task.cancel()
        with self.sink.lock:
            self.sink.add_pcm(self.user, self.pcm * 50, time.monotonic())
            self.assertEqual(len(self.sink.buffers[1][1]), 48000 * 4 * 20)
            self.sink.add_pcm(SimpleNamespace(id=2), self.pcm[:1000], time.monotonic() - 2)
        await asyncio.sleep(0.15)
        self.assertEqual(self.sink.queue.qsize(), 1)
        self.assertNotIn(2, self.sink.buffers)

    async def test_dave_decrypted_before_opus(self):
        session = SimpleNamespace(ready=True, decrypt=Mock(return_value=b'opus'))
        self.sink._voice_client = SimpleNamespace(_connection=SimpleNamespace(
            dave_protocol_version=1, dave_session=session))
        data = SimpleNamespace(opus=b'encrypted', packet=SimpleNamespace(is_silence=lambda: False))
        decoder = Mock()
        decoder.decode.return_value = self.pcm
        with patch('voice_stt.discord.opus.Decoder', return_value=decoder):
            self.sink.write(self.user, data)
        session.decrypt.assert_called_once()
        decoder.decode.assert_called_once_with(b'opus', fec=False)
        self.assertEqual(bytes(self.sink.buffers[1][1]), self.pcm)

    async def test_http_recovery_and_wav_format(self):
        await self.sink.close()
        post = AsyncMock(side_effect=[
            httpx.ConnectError('offline'),
            httpx.Response(200, json={'text': 'Bonjour'}, request=httpx.Request('POST', 'http://stt')),
        ])
        http = AsyncMock()
        http.__aenter__.return_value.post = post
        with patch('voice_stt.httpx.AsyncClient', return_value=http):
            self.sink = STTSink(self.reply, 'http://stt/transcribe')
            self.sink.queue.put_nowait((self.user, self.pcm))
            await asyncio.wait_for(self.sink.queue.join(), 2)
            self.sink.queue.put_nowait((self.user, self.pcm))
            await asyncio.wait_for(self.sink.queue.join(), 2)
            self.reply.assert_awaited_once_with([(self.user, 'Bonjour')])
            payload = post.call_args.kwargs['files']['file'][1]
            with wave.open(BytesIO(payload)) as wav:
                self.assertEqual((wav.getnchannels(), wav.getsampwidth(), wav.getframerate()), (2, 2, 48000))
                self.assertEqual(wav.readframes(wav.getnframes()), self.pcm)

    async def test_transcriptions_continue_and_next_reply_groups_speakers(self):
        await self.sink.close()
        started, release, second = asyncio.Event(), asyncio.Event(), asyncio.Event()
        displayed = AsyncMock()
        batches = []

        async def reply(batch):
            batches.append(batch)
            if len(batches) == 1:
                started.set()
                await release.wait()
            else:
                second.set()

        http = AsyncMock()
        http.__aenter__.return_value.post.side_effect = [
            httpx.Response(200, json={'text': text}, request=httpx.Request('POST', 'http://stt'))
            for text in ('Bonjour', 'Minecraft ?', 'Avec des mods')
        ]
        other = SimpleNamespace(id=2)
        with patch('voice_stt.httpx.AsyncClient', return_value=http):
            self.sink = STTSink(reply, 'http://stt/transcribe', on_transcription=displayed)
            self.sink.queue.put_nowait((self.user, self.pcm))
            await asyncio.wait_for(started.wait(), 2)
            self.sink.queue.put_nowait((other, self.pcm))
            self.sink.queue.put_nowait((self.user, self.pcm))
            await asyncio.wait_for(self.sink.queue.join(), 2)
            self.assertEqual(displayed.await_count, 3)
            self.assertEqual(len(batches), 1)
            with self.sink.lock:
                self.sink.add_pcm(other, self.pcm, time.monotonic())
                self.assertIn(2, self.sink.buffers)
            release.set()
            await asyncio.wait_for(second.wait(), 4)
            self.assertEqual(batches[1], [(other, 'Minecraft ?'), (self.user, 'Avec des mods')])
            self.assertFalse(self.sink.pending)

    async def test_reply_waits_for_global_silence(self):
        with self.sink.lock:
            self.sink.last_voice = time.monotonic()
        self.sink.pending.append((self.user, 'Bonjour'))
        self.sink.pending_event.set()
        await asyncio.sleep(0.2)
        self.reply.assert_not_awaited()
        with self.sink.lock:
            self.sink.last_voice = time.monotonic() - 2.1
        await asyncio.sleep(0.2)
        self.reply.assert_awaited_once_with([(self.user, 'Bonjour')])

    async def test_close_discards_pending_audio(self):
        self.sink.queue.put_nowait((self.user, self.pcm))
        await self.sink.close()
        self.assertTrue(self.sink.closed)
        self.assertTrue(self.sink.queue.empty())
        self.assertTrue(self.sink.flush_task.done())
        self.assertTrue(self.sink.worker_task.done())
        self.assertTrue(self.sink.reply_task.done())
        self.reply.assert_not_awaited()

    async def test_diagnostics_distinguish_ignored_audio_and_dave_failure(self):
        packet = SimpleNamespace(is_silence=lambda: False)
        data = SimpleNamespace(opus=b'PRIVATE_AUDIO', packet=packet)
        session = SimpleNamespace(ready=False, decrypt=Mock(side_effect=ValueError('PRIVATE_KEY')))
        self.sink._voice_client = SimpleNamespace(_connection=SimpleNamespace(
            dave_protocol_version=1, dave_session=session))
        self.sink.on_rtcp_packet(None, None)
        self.sink.write(None, data)
        self.sink.write(self.user, data)
        session.ready = True
        with self.assertLogs('voice_stt', level='INFO') as logs:
            self.sink.write(self.user, data)
            self.sink.log_diagnostic()
        self.assertEqual(self.sink.stats['audio'], 3)
        for reason in ('rtcp', 'unknown_user', 'dave_not_ready', 'DAVE_errors'):
            self.assertEqual(self.sink.stats[reason], 1)
        self.assertEqual(self.sink.stats['pcm'], 0)
        self.assertNotIn('PRIVATE_AUDIO', str(logs.output))
        self.assertNotIn('PRIVATE_KEY', str(logs.output))

    async def test_diagnostics_without_any_audio_and_final_report(self):
        with self.assertLogs('voice_stt', level='INFO') as logs:
            self.sink.log_diagnostic()
            await self.sink.close()
            await self.sink.close()
        self.assertIn("'audio': 0", str(logs.output))
        self.assertEqual(sum('arrêt' in line for line in logs.output), 1)


class CommandTests(unittest.IsolatedAsyncioTestCase):
    async def test_voice_limits_history_and_disables_thinking_only_for_voice(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 == 'call_openai_api')
        response = SimpleNamespace(choices=[SimpleNamespace(message=SimpleNamespace(
            tool_calls=None, content='Bonjour'))])
        create = AsyncMock(return_value=response)
        history = [{'role': 'user', 'content': str(i)} for i in range(10)]
        namespace = dict(transform_emote=lambda text, output: text, PERSONALITY_PROMPT='Personnalité',
                         conversation_history=history, search_web=AsyncMock(return_value=None),
                         WEB_SEARCH_ENABLED=True, MODEL='test', TEMPERATURE=1, logger=Mock(),
                         openai_client=SimpleNamespace(chat=SimpleNamespace(
                             completions=SimpleNamespace(create=create))),
                         add_to_conversation_history=AsyncMock())
        exec(compile(ast.Module(body=[function], type_ignores=[]), '<api>', 'exec'), namespace)
        await namespace['call_openai_api']('Bonjour', 'Alice', voice=True)
        args = create.call_args.kwargs
        self.assertEqual(args['messages'][1:-1], history[-5:])
        self.assertEqual(args['extra_body']['options']['think'], False)
        self.assertIn('une à trois phrases', args['messages'][0]['content'])
        self.assertIn('search_web', args['messages'][0]['content'])
        await namespace['call_openai_api']('Bonjour', 'Alice')
        args = create.call_args.kwargs
        self.assertEqual(args['messages'][1:-1], history)
        self.assertNotIn('extra_body', args)

    async def test_status_connected_and_disconnected(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_status')
        namespace = dict(BOT_VERSION='test', MODEL='gemma', discord=discord)
        exec(compile(ast.Module(body=[function], type_ignores=[]), '<status>', 'exec'), namespace)
        message = SimpleNamespace(guild=None, channel=SimpleNamespace(send=AsyncMock()))
        await namespace['handle_status'](message)
        self.assertIn('déconnecté', message.channel.send.call_args.args[0])
        sink = SimpleNamespace(responding=True, queue=asyncio.Queue(), pending=[1, 2],
                               last_stt_seconds=0.5, last_reply_seconds=None)
        message.guild = SimpleNamespace(voice_client=SimpleNamespace(sink=sink, is_connected=lambda: True))
        await namespace['handle_status'](message)
        text = message.channel.send.call_args.args[0]
        self.assertIn('transcriptions en attente : 2', text)
        self.assertIn('0.5 s', text)
        self.assertIn('pas encore mesuré', text)

    async def test_join_replies_in_command_channel_and_quit_closes_sink(self):
        # Charger uniquement la fonction : aucun token ni connexion Discord.
        tree = ast.parse(Path(__file__).with_name('chatbot.py').read_text(encoding='utf-8'))
        function = next(node for node in tree.body if isinstance(node, ast.AsyncFunctionDef)
                        and node.name == 'handle_voice_command')
        http = AsyncMock()
        http.__aenter__.return_value.get.return_value = Mock()
        sink = SimpleNamespace(close=AsyncMock(), receiver_stopped=Mock(), log_diagnostic=Mock())
        make_sink = 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', guild=SimpleNamespace(voice_client=None, me=SimpleNamespace(voice=None)),
            author=SimpleNamespace(voice=SimpleNamespace(channel=channel)),
            channel=SimpleNamespace(send=AsyncMock(), typing=AsyncMock()),
        )
        message.channel.typing = Mock(return_value=AsyncMock())
        call_api = AsyncMock(return_value=SimpleNamespace(
            choices=[SimpleNamespace(message=SimpleNamespace(content='Réponse'))]))
        namespace = dict(
            VOICE_ENABLED=True, STT_URL='http://stt/transcribe',
            voice_command_lock=asyncio.Lock(), httpx=SimpleNamespace(AsyncClient=Mock(return_value=http)),
            voice_recv=SimpleNamespace(VoiceRecvClient=object), STTSink=make_sink,
            call_openai_api=call_api, transform_emote=lambda text, output: text,
            filter_message=lambda text: text, split_message=lambda text, **kwargs: [text],
            discord=discord, logger=Mock(),
        )
        exec(compile(ast.Module(body=[function], type_ignores=[]), '<commands>', 'exec'), namespace)
        await namespace['handle_voice_command'](message)
        voice.listen.assert_called_once_with(sink, after=sink.receiver_stopped)
        callback = make_sink.call_args.args[0]
        await make_sink.call_args.kwargs['on_transcription'](SimpleNamespace(display_name='Alice'), 'Bonjour')
        await callback([(SimpleNamespace(display_name='Alice'), 'Bonjour')])
        call_api.assert_awaited_once_with('Alice : Bonjour', 'Conversation vocale', voice=True)
        self.assertEqual(message.channel.send.call_args.args[0], 'Réponse')
        quote = message.channel.send.call_args_list[-2]
        self.assertEqual(quote.args[0], '> 🎙 Alice : Bonjour')
        self.assertEqual(quote.kwargs['allowed_mentions'].to_dict(), {'parse': []})
        message.content = '!quit'
        message.guild.voice_client = voice
        await namespace['handle_voice_command'](message)
        sink.close.assert_awaited_once()
        voice.disconnect.assert_awaited_once_with(force=True)


if __name__ == '__main__':
    unittest.main()