"""Real host RTP/RTCP parsers; RTSP server lifecycle is explicitly a test double."""
import importlib.util
from pathlib import Path
import sys
import threading
import types
import unittest
from unittest.mock import patch

import gi
gi.require_version("Gst", "1.0")
gi.require_version("GstRtp", "1.0")
gi.require_version("GstRtsp", "1.0")
from gi.repository import Gst, GstRtp, GstRtsp, GLib
from tools import rtsp_test_time

Gst.init(None)


def load_worker():
    repository = types.ModuleType("gi.repository")
    repository.Gst, repository.GstRtp = Gst, GstRtp
    repository.GstRtsp, repository.GLib = GstRtsp, GLib
    repository.GstRtspServer = types.SimpleNamespace(RTSPMedia=object, RTSPMediaFactory=object)
    repository.GstBase = types.SimpleNamespace()
    path = Path(__file__).parents[1] / "tools/rtsp_test_worker.py"
    spec = importlib.util.spec_from_file_location("rtsp_worker_under_test", path)
    module = importlib.util.module_from_spec(spec)
    with patch.object(gi, "require_version"), patch.dict(sys.modules, {"gi.repository": repository, "rtsp_test_time": rtsp_test_time}):
        spec.loader.exec_module(module)
    return module


WORKER = load_worker()


class WorkerBoundaryTests(unittest.TestCase):
    def test_pipe_rejects_seek_without_changing_query_format(self):
        query = Gst.Query.new_seeking(Gst.Format.TIME)
        query.set_seeking(Gst.Format.TIME, True, 0, 1000000000)
        info = types.SimpleNamespace(get_query=lambda:query)
        self.assertEqual(WORKER.nonseekable_query(None, info), Gst.PadProbeReturn.HANDLED)
        self.assertEqual(tuple(query.parse_seeking()), (Gst.Format.TIME, False, 0, -1))

    def test_unrelated_query_passes_through(self):
        query = Gst.Query.new_latency()
        info = types.SimpleNamespace(get_query=lambda:query)
        self.assertEqual(WORKER.nonseekable_query(None, info), Gst.PadProbeReturn.OK)

    def test_keeper_and_lan_urls_share_public_factory_key(self):
        factory = WORKER.Factory.__new__(WORKER.Factory)
        factory.sender = types.SimpleNamespace(args=types.SimpleNamespace(execution_id="run1"))
        _, local = GstRtsp.RTSPUrl.parse("rtsp://127.0.0.1:8554/sample01")
        _, lan = GstRtsp.RTSPUrl.parse("rtsp://192.168.10.20:18554/sample01")
        self.assertEqual(factory.do_gen_key(local), factory.do_gen_key(lan))

    def sender(self):
        sender = WORKER.Sender.__new__(WORKER.Sender)
        sender.clock = rtsp_test_time.SenderClock("execution1")
        sender.clock_lock = threading.Lock()
        sender.clock_error = None
        sender.last_packet = None
        sender.last_packet_wall = None
        return sender

    def test_official_rtp_zero_header_to_production_callback(self):
        sender = self.sender()
        buffer = GstRtp.RTPBuffer.new_allocate(4, 0, 0)
        ok, packet = GstRtp.RTPBuffer.map(buffer, Gst.MapFlags.READ | Gst.MapFlags.WRITE)
        self.assertTrue(ok)
        packet.set_timestamp(0)
        packet.set_ssrc(7)
        packet.unmap()
        pad = types.SimpleNamespace(get_current_caps=lambda:Gst.Caps.from_string("application/x-rtp,clock-rate=(int)90000"))
        info = types.SimpleNamespace(get_buffer=lambda:buffer)
        self.assertEqual(sender.on_packet(pad, info), Gst.PadProbeReturn.OK)
        self.assertEqual(sender.clock.snapshot()["latestRtpTimestamp"], 0)

    def test_official_sender_report_to_production_callback(self):
        sender = self.sender()
        sender.clock.packet(900000, 90000, 7)
        buffer = GstRtp.RTCPBuffer.new(1400)
        mapped = GstRtp.RTCPBuffer()
        self.assertTrue(GstRtp.RTCPBuffer.map(buffer, Gst.MapFlags.READ | Gst.MapFlags.WRITE, mapped))
        packet = GstRtp.RTCPPacket()
        self.assertTrue(mapped.add_packet(GstRtp.RTCPType.SR, packet))
        ntp = (rtsp_test_time.NTP_UNIX_OFFSET + 1700000000) << 32
        packet.sr_set_sender_info(7, ntp, 900000, 30, 1000)
        mapped.unmap()
        self.assertTrue(sender.on_rtcp(None, buffer, False))
        self.assertEqual(sender.clock.anchor["frame_time_anchor"]["startWallTimeMs"], 1699999990000)
        self.assertIsNone(sender.clock_error)

    def test_finish_quits_even_if_status_save_fails(self):
        sender = self.sender()
        sender.terminal = None
        from unittest.mock import Mock
        sender.loop = Mock()
        sender.report = Mock(side_effect=OSError("disk"))
        with self.assertRaises(OSError): sender.finish("FAILED", "io")
        sender.loop.quit.assert_called_once()
        sender.report.reset_mock()
        sender.finish("STOPPED")
        sender.report.assert_not_called()


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