import os
import stat
import tempfile
import unittest
from pathlib import Path

import yaml
from pydantic import ValidationError

from app.source_adapters.rtsp import (
    RtspSourceAdapter,
    RtspSourceAdapterError,
    RtspSourceOptions,
)


class _Node:
    def __init__(self):
        self.properties = None

    def set(self, properties):
        self.properties = dict(properties)


class _Pipeline:
    def __init__(self):
        self.nodes = {}

    def __getitem__(self, name):
        return self.nodes.setdefault(name, _Node())


class _Flow:
    def __init__(self):
        self.capture_args = None
        self.probe = None

    def batch_capture(self, path, **kwargs):
        self.capture_args = (path, kwargs)
        return self

    def attach(self, probe):
        self.probe = probe
        return self


class RtspSourceOptionsTest(unittest.TestCase):
    def test_defaults_and_nvidia_mapping(self):
        options = RtspSourceOptions()
        self.assertEqual("tcp", options.transport)
        self.assertEqual(1000, options.latency_ms)
        self.assertTrue(options.disable_audio)
        self.assertEqual("none", options.reconnect_policy)
        self.assertEqual(60, options.first_frame_timeout_s)
        self.assertEqual(
            {
                "type": 2,
                "file-loop": False,
                "select-rtp-protocol": 4,
                "latency": 1000,
                "disable-audio": True,
                "gpu-id": 0,
                "init-rtsp-reconnect-interval": 0,
                "rtsp-reconnect-interval": 0,
                "rtsp-reconnect-attempts": 0,
            },
            options.nvidia_properties(0),
        )

    def test_auto_transport_and_validation(self):
        self.assertEqual(
            7,
            RtspSourceOptions(transport="auto").nvidia_properties(0)[
                "select-rtp-protocol"
            ],
        )
        with self.assertRaises(ValidationError):
            RtspSourceOptions(transport="udp")
        with self.assertRaises(ValidationError):
            RtspSourceOptions(latency_ms=60_001)
        with self.assertRaises(ValidationError):
            RtspSourceOptions(first_frame_timeout_s=4)
        with self.assertRaises(ValidationError):
            RtspSourceOptions(reconnect_policy="automatic")


class RtspSourceAdapterTest(unittest.TestCase):
    def _adapter(self, root, uri="rtsp://camera.invalid/live"):
        return RtspSourceAdapter(uri, config_root=root)

    def test_rtp_setup_uses_capture_probe_and_exact_source(self):
        from unittest.mock import Mock
        from app.start_diagnostics import StartDiagnostics
        with tempfile.TemporaryDirectory() as root:
            with self._adapter(root).prepare('det', 1) as prepared:
                flow = _Flow()
                flow.attach = Mock(return_value=flow)
                diagnostic = Mock(wraps=StartDiagnostics('det'))
                prepared.build_capture(flow, _Pipeline(), rtp_time=True, diagnostic=diagnostic)
                flow.attach.assert_called_once_with('contextra_rtp_time', 'rtp-time',
                    properties={'source-node': prepared.source_node})

    def test_prepare_writes_official_private_source_config(self):
        with tempfile.TemporaryDirectory() as root:
            config_root = Path(root) / "sources"
            prepared = self._adapter(config_root).prepare("det", 3)
            document = yaml.safe_load(prepared.config_path.read_text(encoding="utf-8"))

            self.assertEqual(0o700, stat.S_IMODE(config_root.stat().st_mode))
            self.assertEqual(0o600, stat.S_IMODE(prepared.config_path.stat().st_mode))
            self.assertEqual("nvurisrcbin", document["source-config"]["source-bin"])
            self.assertFalse(document["source-config"]["properties"]["file-loop"])
            self.assertEqual(4, document["source-config"]["properties"]["select-rtp-protocol"])
            self.assertEqual(prepared.source_node, document["source-list"][0]["sensor-id"])
            self.assertEqual("rtsp:det:g3", prepared.stage_identity)

            prepared.cleanup()
            prepared.cleanup()
            self.assertFalse(prepared.config_path.exists())

    def test_each_role_generation_gets_a_collision_free_path(self):
        with tempfile.TemporaryDirectory() as root:
            adapter = self._adapter(Path(root) / "sources")
            first = adapter.prepare("det", 1)
            second = adapter.prepare("det", 1)
            try:
                self.assertNotEqual(first.config_path, second.config_path)
                self.assertTrue(first.config_path.exists())
                self.assertTrue(second.config_path.exists())
            finally:
                first.cleanup()
                second.cleanup()

    def test_build_capture_applies_properties_omitted_by_flow(self):
        with tempfile.TemporaryDirectory() as root:
            prepared = self._adapter(Path(root) / "sources").prepare("det", 1)
            flow = _Flow()
            pipeline = _Pipeline()
            probe = object()
            try:
                result = prepared.build_capture(flow, pipeline, decoded_probe=probe)
                self.assertIs(result, flow)
                self.assertEqual(str(prepared.config_path), flow.capture_args[0])
                self.assertEqual(
                    {"gpu_id": 0, "live_source": True, "file_loop": False},
                    flow.capture_args[1],
                )
                self.assertEqual(
                    prepared.nvidia_properties,
                    pipeline[prepared.source_node].properties,
                )
                self.assertIs(probe, flow.probe)
                prepared.build_capture(flow, pipeline, frame_size=(576, 576))
                self.assertEqual(576, flow.capture_args[1]['width'])
                self.assertEqual(576, flow.capture_args[1]['height'])
            finally:
                prepared.cleanup()

    def test_uri_and_path_traversal_are_fail_closed_without_secret_echo(self):
        with tempfile.TemporaryDirectory() as root:
            with self.assertRaises(RtspSourceAdapterError) as invalid_uri:
                self._adapter(Path(root) / "sources", uri="not-a-source")
            self.assertNotIn("not-a-source", str(invalid_uri.exception))

            protected_uri = "rtsp://camera.invalid/private-path"
            adapter = self._adapter(Path(root) / "sources", protected_uri)
            self.assertNotIn(protected_uri, repr(adapter))
            with self.assertRaises(RtspSourceAdapterError):
                adapter.prepare("../det", 1)

    def test_relative_or_symlink_root_is_rejected(self):
        with self.assertRaises(RtspSourceAdapterError):
            self._adapter("relative/sources").prepare("det", 1)
        with tempfile.TemporaryDirectory() as root:
            target = Path(root) / "target"
            target.mkdir()
            link = Path(root) / "sources"
            os.symlink(target, link)
            with self.assertRaises(RtspSourceAdapterError):
                self._adapter(link).prepare("det", 1)


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