"""Offline transport construction and pipeline/event fixtures, with no worker run."""

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

import aiohttp
from pipecat.processors.frame_processor import FrameProcessor
from pipecat.transports.lemonslice.transport import LemonSliceNewSessionRequest, LemonSliceTransport
from pipecat.workers.runner import WorkerRunner
from pipecat_lemonslice import AvatarConfig, build_avatar_pipeline


CONFIG = AvatarConfig(
    image_url="https://example.com/avatar.jpg",
    daily_room_url="https://example.daily.co/offline-room",
    api_key="offline-provider-placeholder",
    daily_token="offline-room-token-placeholder",
)


def stages():
    return {name: FrameProcessor(name=name) for name in (
        "stt", "user_aggregator", "llm", "tts", "assistant_aggregator"
    )}


class TransportFixture:
    def __init__(self, **kwargs):
        self.kwargs = kwargs
        self.events = {}
        self.input_stage = FrameProcessor(name="avatar_input")
        self.output_stage = FrameProcessor(name="avatar_output")

    def input(self):
        return self.input_stage

    def output(self):
        return self.output_stage

    def event_handler(self, name):
        def register(callback):
            self.events[name] = callback
            return callback
        return register


class Tests(unittest.IsolatedAsyncioTestCase):
    async def asyncSetUp(self):
        self.guard = patch("socket.socket.connect", side_effect=AssertionError("Network forbidden"))
        self.guard.start()
        self.request_guard = patch.object(
            aiohttp.ClientSession, "_request", side_effect=AssertionError("HTTP forbidden")
        )
        self.request_guard.start()

    async def asyncTearDown(self):
        self.request_guard.stop()
        self.guard.stop()

    async def test_installed_sdk_contract_and_transport_construction(self):
        required = {"agent_image_url", "daily_room_url", "daily_token", "idle_timeout"}
        self.assertTrue(required <= LemonSliceNewSessionRequest.model_fields.keys())
        self.assertIn("session_request", inspect.signature(LemonSliceTransport).parameters)
        self.assertTrue(inspect.iscoroutinefunction(WorkerRunner.add_workers))
        async with aiohttp.ClientSession() as http:
            transport, worker = build_avatar_pipeline(http, CONFIG, **stages())
            self.assertIsInstance(transport, LemonSliceTransport)
            self.assertIsNotNone(worker)
            self.assertIsNotNone(transport.input())
            self.assertIsNotNone(transport.output())
        # No setup(), WorkerRunner.run(), room, audio device, or provider request.

    async def test_pipeline_order_uses_existing_voice_processors(self):
        captured = []
        from pipecat.pipeline.pipeline import Pipeline

        def capture(processors):
            captured.extend(processors)
            return Pipeline(processors)

        voices = stages()
        async with aiohttp.ClientSession() as http:
            with patch("pipecat_lemonslice.Pipeline", side_effect=capture):
                transport, _worker = build_avatar_pipeline(
                    http, CONFIG, **voices, transport_factory=TransportFixture
                )
        self.assertEqual(captured, [
            transport.input_stage, voices["stt"], voices["user_aggregator"],
            voices["llm"], voices["tts"], transport.output_stage, voices["assistant_aggregator"],
        ])
        request = transport.kwargs["session_request"]
        self.assertEqual(request.daily_room_url, CONFIG.daily_room_url)
        self.assertEqual(request.idle_timeout, 60)
        self.assertFalse(transport.kwargs["params"].microphone_out_enabled)

    async def test_human_and_avatar_disconnect_cancel_worker(self):
        async with aiohttp.ClientSession() as http:
            transport, worker = build_avatar_pipeline(
                http, CONFIG, **stages(), transport_factory=TransportFixture
            )
            with patch.object(worker, "cancel", new_callable=AsyncMock) as cancel:
                await transport.events["on_client_disconnected"](transport, {"id": "human"})
                await transport.events["on_avatar_disconnected"](transport, {"id": "avatar"}, "offline")
                self.assertEqual(cancel.await_count, 2)

    async def test_keys_redacted_and_bad_configuration_rejected(self):
        self.assertNotIn(CONFIG.api_key, repr(CONFIG))
        self.assertNotIn(CONFIG.daily_token, repr(CONFIG))
        with self.assertRaises(ValueError):
            AvatarConfig(CONFIG.image_url, "http://example.com/room", "x", "x")
        with self.assertRaises(ValueError):
            AvatarConfig(CONFIG.image_url, CONFIG.daily_room_url, "", "x")


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