"""Несетевые тесты V3. Локальные HTTP/RTC peers не обращаются к OpenAI."""
import asyncio
import importlib.util
import json
from pathlib import Path
import struct
import subprocess
import sys
import tempfile
import types
import unittest
from unittest.mock import patch
import wave

import aiohttp
from aiohttp import web
from aiortc import RTCPeerConnection, RTCConfiguration

ROOT = Path(__file__).resolve().parent
spec = importlib.util.spec_from_file_location("voice", ROOT / "voice-console.py")
voice = importlib.util.module_from_spec(spec)
spec.loader.exec_module(voice)


def args(root):
    return types.SimpleNamespace(auth_file=root / "auth.json", input_device=None, output_device=None,
                                 output=root / "answer.wav", voice="cove", model=None, timeout=2, max_seconds=2)


def wav(path, rate=48000, channels=1, width=2):
    with wave.open(str(path), "wb") as audio:
        audio.setparams((channels, width, rate, 0, "NONE", "not compressed"))
        audio.writeframes(b"\x01" * (rate // 4 * channels * width))


class ContractTests(unittest.TestCase):
    def test_v3_body_has_only_native_session_fields(self):
        body = voice.call_body("v=0\r\n", "cove")
        self.assertEqual(set(body), {"sdp", "session"})
        self.assertEqual(body["session"]["audio"], {"output": {"voice": "cove"}})
        self.assertEqual(body["session"]["delegation"], {"type": "client"})
        self.assertNotIn("type", body["session"])
        self.assertEqual(body["session"]["model"], "gpt-live-1-codex")
        self.assertEqual(voice.call_body("sdp", "sol", "snapshot")["session"]["model"], "snapshot")
        with self.assertRaises(voice.VoiceError):
            voice.call_body("sdp", "marin")  # GA V2 voice — не каталог V3.

    def test_location_is_id_not_redirect_target(self):
        self.assertEqual(voice.call_id_from_location("/v1/live/rtc_valid-1"), "rtc_valid-1")
        for location in [None, "", "/", "..", "/v1/live/%2fsecret", "/v1/live/name with space"]:
            with self.subTest(location=location), self.assertRaises(voice.VoiceError):
                voice.call_id_from_location(location)

    def test_auth_read_only_and_rejects_injection(self):
        with tempfile.TemporaryDirectory() as folder:
            path = Path(folder) / "auth.json"
            for auth in [{}, {"auth_mode": "apikey"}, {"auth_mode": "chatgpt", "tokens": {"access_token": "0000"}},
                         {"auth_mode": "chatgpt", "tokens": {"access_token": "[скрыто]"}},
                         {"auth_mode": "chatgpt", "tokens": {"access_token": "[скрыто]", "account_id": "bad\nheader"}}]:
                path.write_text(json.dumps(auth))
                with self.assertRaises(voice.VoiceError):
                    voice.read_auth(path)
            path.write_text(json.dumps({"auth_mode": "chatgpt", "tokens": {"access_token": "[скрыто]", "account_id": "own"}}))
            original = path.read_bytes()
            headers = voice.read_auth(path)
            self.assertEqual(headers["Authorization"], "Bearer synthetic")
            self.assertEqual(path.read_bytes(), original)
            self.assertNotIn("Cookie", headers)
            self.assertNotIn("X-OAI-IS", headers)
            self.assertEqual(headers["OpenAI-Alpha"], "quicksilver=v2")
            self.assertEqual(headers["originator"], "codex_voice_isolated")

    def test_file_wrong_rate_channels_width_and_truncation(self):
        with tempfile.TemporaryDirectory() as folder:
            path = Path(folder) / "input.wav"
            wav(path)
            self.assertEqual(len(voice.read_wav(path)), 24000)
            for options in [{"rate": 24000}, {"channels": 2}, {"width": 1}, {"width": 3}]:
                wav(path, **options)
                with self.assertRaises(voice.VoiceError):
                    voice.read_wav(path)
            wav(path)
            path.write_bytes(path.read_bytes()[:-5])
            with self.assertRaises(voice.VoiceError):
                voice.read_wav(path)
            path.write_bytes(b"not-a-WAV")
            with self.assertRaises(wave.Error):
                voice.read_wav(path)

    def test_caption_v3_not_ga_or_nested_content(self):
        self.assertEqual(voice.caption({"type": "input_transcript.added", "item": {"text": "Привет"}}), ("user", False, "Привет"))
        self.assertEqual(voice.caption({"type": "turn.done", "turn": {"role": "assistant", "transcript": "Да"}}), ("assistant", True, "Да"))
        for event in [{"type": "response.audio_transcript.delta", "delta": "wrong"},
                      {"type": "input_transcript.added", "item": {"content": "wrong"}},
                      {"type": "turn.done", "turn": {"role": "tool", "transcript": "command"}}]:
            self.assertIsNone(voice.caption(event))

    def test_output_private_complete_no_overwrite_and_invalid_pcm(self):
        with tempfile.TemporaryDirectory() as folder:
            path = Path(folder) / "output.wav"
            for data in [b"", b"\x01"]:
                with self.assertRaises(voice.VoiceError):
                    voice.save_wav(path, data)
            voice.save_wav(path, b"\x01\x02" * 960)
            self.assertEqual(path.stat().st_mode & 0o777, 0o600)
            original = path.read_bytes()
            with self.assertRaises(FileExistsError):
                voice.save_wav(path, b"\x01\x02")
            self.assertEqual(path.read_bytes(), original)
            with wave.open(str(path)) as audio:
                self.assertEqual((audio.getframerate(), audio.getnchannels(), audio.getsampwidth(), audio.getnframes()), (48000, 1, 2, 960))

    def test_copied_script_dry_run_no_auth_or_fork(self):
        with tempfile.TemporaryDirectory() as folder:
            path = Path(folder) / "voice-console.py"
            path.write_bytes((ROOT / path.name).read_bytes())
            result = subprocess.run([sys.executable, str(path), "--dry-run", "--auth-file", str(Path(folder) / "missing")],
                                    cwd=folder, capture_output=True, text=True, timeout=5)
            self.assertEqual(result.returncode, 0, result.stderr)
            data = json.loads(result.stdout)
            for key in ["network", "credentials_read", "microphone", "codex_process"]:
                self.assertFalse(data[key])


class AsyncTests(unittest.IsolatedAsyncioTestCase):
    async def test_device_abort_failure_still_closes_peer_and_http(self):
        class BrokenCapture:
            closed = False
            def abort(self): raise OSError("synthetic device failure")
            def close(self): self.closed = True
        with tempfile.TemporaryDirectory() as folder:
            session = voice.VoiceSession(args(Path(folder)), b"\0\0" * 960, emit=lambda _: None)
            capture = BrokenCapture()
            session.track.capture = capture
            session.http = aiohttp.ClientSession()
            await session.close()
            self.assertTrue(capture.closed and session.http.closed)
            self.assertEqual(session.pc.connectionState, "closed")
            self.assertEqual(session.track.readyState, "ended")

    async def test_assistant_greeting_does_not_complete_file_before_user_turn(self):
        class FakeWS:
            events = [{"type": "turn.done", "turn": {"role": "assistant", "transcript": "Здравствуйте"}}]
            def __aiter__(self): return self
            async def __anext__(self):
                if not self.events: raise StopAsyncIteration
                return types.SimpleNamespace(type=aiohttp.WSMsgType.TEXT, data=json.dumps(self.events.pop(0)))
        with tempfile.TemporaryDirectory() as folder:
            session = voice.VoiceSession(args(Path(folder)), b"\0\0" * 960, emit=lambda _: None)
            session.ws = FakeWS()
            await session.receive_events()
            self.assertFalse(session.assistant_done.is_set())
            session.ws = None
            await session.close()

    async def test_websocket_redirect_refused_before_second_request(self):
        seen = []
        async def redirect(request):
            seen.append("first")
            return web.Response(status=302, headers={"Location": "/second"})
        async def second(request):
            seen.append("second")
            return web.Response(status=403)
        app = web.Application()
        app.router.add_get("/first", redirect)
        app.router.add_get("/second", second)
        runner = web.AppRunner(app)
        await runner.setup()
        site = web.TCPSite(runner, "127.0.0.1", 0)
        await site.start()
        port = site._server.sockets[0].getsockname()[1]
        try:
            async with aiohttp.ClientSession(headers={"Authorization": "[скрыто]"}, trace_configs=[voice.no_redirect_trace()]) as client:
                with self.assertRaisesRegex(voice.VoiceError, "перенаправление"):
                    await client.ws_connect(f"http://127.0.0.1:{port}/first")
            self.assertEqual(seen, ["first"])
        finally:
            await runner.cleanup()

    async def test_mute_flushes_backlog_and_old_callback(self):
        track = voice.InputTrack(b"\x01\x02" * 1000)
        track.unmute()
        track.enqueue(b"\x03\x04" * 960, track.epoch)
        old_epoch = track.epoch
        track.mute()
        track.enqueue(b"\x03\x04" * 960, old_epoch)
        self.assertFalse(track.queue)
        self.assertFalse(track.pending)
        self.assertEqual(track.next_pcm(), voice.SILENCE)
        track.unmute()
        self.assertEqual(track.next_pcm()[:2], b"\x01\x02")
        track.stop()

    async def test_microphone_aborted_and_closed_on_mute(self):
        class Capture:
            aborted = closed = False
            def abort(self): self.aborted = True
            def close(self): self.closed = True
        track = voice.InputTrack()
        capture = Capture()
        track.capture = capture
        track.muted = False
        track.mute()
        self.assertTrue(capture.aborted and capture.closed)
        self.assertIsNone(track.capture)
        track.stop()

    async def test_queue_bounded_and_frames_aligned(self):
        track = voice.InputTrack(b"\x01\x02" * 960)
        track.unmute()
        track.enqueue(b"\x01\x02" * (960 * 20) + b"\x05", track.epoch)
        self.assertEqual(len(track.queue), 10)
        self.assertEqual(len(track.pending), 1)
        frames = [await track.recv(), await track.recv()]
        self.assertEqual([f.pts for f in frames], [0, 960])
        self.assertEqual(frames[0].sample_rate, 48000)
        self.assertEqual(frames[0].samples, 960)
        track.stop()

    async def test_outgoing_http_body_and_failure_cleanup(self):
        # Перехват на локальном HTTP endpoint: никаких настоящих credentials.
        requests = []
        async def handler(request):
            requests.append((request.method, await request.json(), dict(request.headers)))
            return web.Response(status=403, headers={"cf-mitigated": "challenge"}, text="private must stay hidden")
        app = web.Application()
        app.router.add_post("/call", handler)
        runner = web.AppRunner(app)
        await runner.setup()
        site = web.TCPSite(runner, "127.0.0.1", 0)
        await site.start()
        port = site._server.sockets[0].getsockname()[1]
        try:
            with tempfile.TemporaryDirectory() as folder:
                root = Path(folder)
                (root / "auth.json").write_text(json.dumps({"auth_mode": "chatgpt", "tokens": {"access_token": "[скрыто]"}}))
                session = voice.VoiceSession(args(root), b"\x01\x02" * 960, emit=lambda _: None)
                with patch.object(voice, "CALL_URL", f"http://127.0.0.1:{port}/call"):
                    with self.assertRaisesRegex(voice.VoiceError, "HTTP 403; challenge=True"):
                        await session.start()
                self.assertTrue(session.closed and session.http.closed)
                self.assertEqual(session.pc.connectionState, "closed")
                self.assertEqual(session.track.readyState, "ended")
                self.assertEqual(requests[0][0], "POST")
                self.assertEqual(requests[0][1]["session"], voice.call_body("", "cove")["session"])
                self.assertTrue(requests[0][1]["sdp"].startswith("v=0"))
                self.assertNotIn("audio", requests[0][1])  # Нет file/PCM в signaling.
                await session.close()  # Повторное закрытие безопасно.
        finally:
            await runner.cleanup()

    async def test_delegation_closes_without_running_tools(self):
        class FakeWS:
            def __aiter__(self):
                return self
            async def __anext__(self):
                if getattr(self, "sent", False):
                    raise StopAsyncIteration
                self.sent = True
                return types.SimpleNamespace(type=aiohttp.WSMsgType.TEXT, data=json.dumps({"type": "delegation.created", "item": {"content": "do not execute"}}))
        with tempfile.TemporaryDirectory() as folder:
            session = voice.VoiceSession(args(Path(folder)), b"\0\0" * 960, emit=lambda _: None)
            session.ws = FakeWS()
            await session.receive_events()
            self.assertTrue(session.finished.is_set())
            self.assertIn("delegation", str(session.failed))
            session.ws = None
            await session.close()

    async def test_local_peer_to_peer_opus_rtp_and_decode(self):
        # Два независимых aiortc peer: реальный Opus/ICE/DTLS/SRTP без сервиса.
        sender = RTCPeerConnection(RTCConfiguration(iceServers=[]))
        receiver = RTCPeerConnection(RTCConfiguration(iceServers=[]))
        tone = b"".join(struct.pack("<h", int(8000 * __import__("math").sin(2 * __import__("math").pi * 440 * i / 48000))) for i in range(48000))
        track = voice.InputTrack(tone)
        track.unmute()
        trans = sender.addTransceiver(track, direction="sendrecv")
        trans.setCodecPreferences([c for c in voice.RTCRtpSender.getCapabilities("audio").codecs if c.mimeType == "audio/opus"])
        sender.createDataChannel("oai-events")
        decoded = []
        received = asyncio.Event()
        tasks = []
        @receiver.on("track")
        def on_track(remote):
            async def collect():
                for _ in range(8):
                    decoded.append(await remote.recv())
                received.set()
            tasks.append(asyncio.create_task(collect()))
        try:
            await sender.setLocalDescription(await sender.createOffer())
            self.assertIn("opus/48000/2", sender.localDescription.sdp)
            await receiver.setRemoteDescription(sender.localDescription)
            await receiver.setLocalDescription(await receiver.createAnswer())
            await sender.setRemoteDescription(receiver.localDescription)
            await asyncio.wait_for(received.wait(), 10)
            self.assertTrue(any(any(bytes(f.planes[0])) for f in decoded))
            self.assertTrue(all(f.sample_rate == 48000 for f in decoded))
            tx, rx = await sender.getStats(), await receiver.getStats()
            self.assertGreater(sum(getattr(s, "packetsSent", 0) for s in tx.values()), 0)
            self.assertGreater(sum(getattr(s, "packetsReceived", 0) for s in rx.values()), 0)
            track.mute()
            self.assertEqual(track.next_pcm(), voice.SILENCE)
        finally:
            track.stop()
            for task in tasks:
                task.cancel()
            await asyncio.gather(*tasks, return_exceptions=True)
            await sender.close()
            await receiver.close()
        self.assertEqual(sender.connectionState, "closed")
        self.assertEqual(receiver.connectionState, "closed")


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