import asyncio
import unittest
from unittest.mock import AsyncMock, patch

from pipecat.frames.frames import (
    BotBackchannelFrame, BotStoppedSpeakingFrame, CancelFrame,
    InterruptionFrame, LLMFullResponseStartFrame, TTSSpeakFrame,
    UserStartedSpeakingFrame, UserStoppedSpeakingFrame,
    VADUserStartedSpeakingFrame, VADUserStoppedSpeakingFrame,
)
from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
from schemas import TurnConfig
from services.pipecat.backchannel import BotBackchannelProcessor


class BotBackchannelTests(unittest.IsolatedAsyncioTestCase):
    async def asyncSetUp(self):
        self.allowed = True
        self.frames = []
        self.processor = BotBackchannelProcessor(delay_secs=0.01, can_speak=lambda: self.allowed)
        self.processor.create_task = asyncio.create_task
        async def cancel(task):
            task.cancel()
            try:
                await task
            except asyncio.CancelledError:
                pass
        self.processor.cancel_task = cancel
        async def push(frame, direction=FrameDirection.DOWNSTREAM):
            self.frames.append(frame)
        self.processor.push_frame = push
        self.patch = patch.object(FrameProcessor, "process_frame", AsyncMock())
        self.patch.start()

    async def asyncTearDown(self):
        await self.processor.cleanup()
        self.patch.stop()

    async def send(self, frame):
        await self.processor.process_frame(frame, FrameDirection.DOWNSTREAM)

    async def start_turn(self):
        await self.send(VADUserStartedSpeakingFrame())
        await self.send(UserStartedSpeakingFrame())
        await self.send(InterruptionFrame())

    def acknowledgments(self):
        return [frame for frame in self.frames if isinstance(frame, BotBackchannelFrame)]

    async def test_long_turn_once_and_excludes_context(self):
        await self.start_turn()
        await asyncio.sleep(0.03)
        self.assertEqual(len(self.acknowledgments()), 1)
        self.assertFalse(self.acknowledgments()[0].append_to_context)
        self.assertFalse(self.acknowledgments()[0].interruptible)
        await self.send(BotStoppedSpeakingFrame())
        await self.send(VADUserStoppedSpeakingFrame())
        await self.send(VADUserStartedSpeakingFrame())
        await asyncio.sleep(0.03)
        self.assertEqual(len(self.acknowledgments()), 1)

    async def test_short_turn_does_not_acknowledge(self):
        await self.start_turn()
        await self.send(UserStoppedSpeakingFrame())
        await asyncio.sleep(0.03)
        self.assertFalse(self.acknowledgments())

    async def test_silence_cancels_pending(self):
        await self.start_turn()
        await self.send(VADUserStoppedSpeakingFrame())
        await asyncio.sleep(0.03)
        self.assertFalse(self.acknowledgments())

    async def test_formal_reply_cancels_pending(self):
        await self.start_turn()
        await self.send(LLMFullResponseStartFrame())
        await asyncio.sleep(0.03)
        self.assertFalse(self.acknowledgments())

    async def test_greeting_reserves_output(self):
        await self.send(TTSSpeakFrame("欢迎"))
        await self.send(UserStartedSpeakingFrame())
        await self.send(VADUserStartedSpeakingFrame())
        await asyncio.sleep(0.03)
        self.assertFalse(self.acknowledgments())

    async def test_closing_or_muted_blocks_timer(self):
        await self.start_turn()
        self.allowed = False
        await asyncio.sleep(0.03)
        self.assertFalse(self.acknowledgments())

    async def test_cancel_closes_controller(self):
        await self.start_turn()
        await self.send(CancelFrame())
        await asyncio.sleep(0.03)
        self.assertFalse(self.acknowledgments())
        self.assertIsNone(self.processor._timer)

    async def test_cooldown_applies_between_turns(self):
        await self.start_turn()
        await asyncio.sleep(0.03)
        await self.send(UserStoppedSpeakingFrame())
        await self.send(BotStoppedSpeakingFrame())
        await self.start_turn()
        await asyncio.sleep(0.03)
        self.assertEqual(len(self.acknowledgments()), 1)

    def test_old_configuration_defaults_off_and_alias_roundtrip(self):
        self.assertFalse(TurnConfig().bot_backchannel.enabled)
        config = TurnConfig.model_validate({"botBackchannel": {"enabled": True, "delaySecs": 5}})
        self.assertEqual(config.bot_backchannel.delay_secs, 5)


class BackchannelPipelineTest(unittest.IsolatedAsyncioTestCase):
    async def test_native_pipeline_delivers_backchannel_without_assistant_context(self):
        """Exercise real frame queues and the project's pre-TTS aggregator."""
        from pipecat.frames.frames import EndFrame
        from pipecat.pipeline.pipeline import Pipeline
        from pipecat.pipeline.worker import PipelineWorker
        from pipecat.processors.aggregators.llm_context import LLMContext
        from services.pipecat.processors import PassthroughLLMAssistantAggregator
        from services.pipecat.pipeline_runner import run_media_worker

        received = asyncio.Event()
        context = LLMContext([])
        class Sink(FrameProcessor):
            async def process_frame(self, frame, direction):
                await super().process_frame(frame, direction)
                if isinstance(frame, BotBackchannelFrame):
                    received.set()
                await self.push_frame(frame, direction)

        controller = BotBackchannelProcessor(delay_secs=0.01, can_speak=lambda: True)
        worker = PipelineWorker(Pipeline([
            PassthroughLLMAssistantAggregator(context), controller, Sink(),
        ]), enable_rtvi=False)
        @worker.event_handler("on_pipeline_started")
        async def start(_worker, _frame):
            for frame in (VADUserStartedSpeakingFrame(), UserStartedSpeakingFrame(), InterruptionFrame()):
                await worker.queue_frame(frame)

        run = asyncio.create_task(run_media_worker(worker, None))
        try:
            await asyncio.wait_for(received.wait(), 3)
            self.assertEqual(context.messages, [])
            await worker.queue_frame(EndFrame())
            await asyncio.wait_for(run, 3)
            self.assertIsNone(controller._timer)
        finally:
            if not run.done():
                run.cancel()
                try:
                    await run
                except asyncio.CancelledError:
                    pass