import unittest
import os
import sys
import types
from io import BytesIO
from pathlib import Path
from unittest.mock import patch

sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
import ui_server
from ui_server import HTML, INPUT_EXTENSIONS, MAX_JSON_BODY_BYTES, active_video_name, archive_files, archive_path, camera_bridge_source, cleanup_empty_recordings, content_disposition, copy_exact, h264_cache_path, input_upload_path, latest_file, model_upload_path, normalize_control, parse_perf, parse_quality, parse_range_header, parse_source, read_json_file, resolve_model_path, tail_text, udp_probe_path


class UIServerTests(unittest.TestCase):
    def test_camera_bridge_source_contains_requested_capture_mode(self):
        source = camera_bridge_source(
            "http://host.docker.internal:8091/stream.mjpg",
            0,
            1920,
            1080,
            30,
        )
        self.assertIn("index=0", source)
        self.assertIn("width=1920", source)
        self.assertIn("height=1080", source)
        self.assertIn("fps=30", source)

    def test_html_has_separate_settings_and_log_drawer(self):
        self.assertIn('id="settingsTab"', HTML)
        self.assertIn('data-tab="settings"', HTML)
        self.assertIn('id="logDrawer"', HTML)
        self.assertIn('value="guidance_v1"', HTML)
        self.assertIn('id="errorObjectId"', HTML)
        self.assertIn('id="errorHost"', HTML)
        self.assertIn('placeholder="192.168.1.10"', HTML)
        self.assertIn('src = \'/stream.mjpg?t=\'', HTML)
        self.assertIn('if (streamRunning && !wasRunning) reconnectFrame();', HTML)
        self.assertIn('<img id="frame"', HTML)
        self.assertIn('<img id="frameFallback"', HTML)
        self.assertEqual(HTML.count('id="startRun"'), 1)

    def test_model_panel_and_themes_are_available(self):
        self.assertIn('data-tab="model"', HTML)
        self.assertIn('id="modelTab"', HTML)
        self.assertIn('id="modelSelect"', HTML)
        self.assertNotIn('id="modelInspect"', HTML)
        self.assertNotIn('id="architectureList"', HTML)
        self.assertIn('id="modelNetron"', HTML)
        self.assertIn('id="netronFrame"', HTML)
        self.assertIn('/api/model/netron', HTML)
        self.assertIn('id="modelConf"', HTML)
        self.assertIn('value="light"', HTML)
        self.assertIn('value="amber"', HTML)
        self.assertIn('value="midnight"', HTML)
        self.assertIn('value="forest"', HTML)
        self.assertIn('value="rose"', HTML)
        self.assertIn('id="accentSelect"', HTML)
        self.assertIn('id="accentCustom"', HTML)
        self.assertIn('id="accentIntensity"', HTML)
        self.assertIn('value="violet"', HTML)
        self.assertIn('value="lime"', HTML)
        for value in ("graphite", "solarized", "ocean", "neon", "copper", "mono", "indigo", "teal", "gold", "sky", "coral"):
            self.assertIn(f'value="{value}"', HTML)
        self.assertIn('fpv-accent-intensity', HTML)
        self.assertIn('@keyframes page-enter', HTML)
        self.assertIn('@keyframes form-field-in', HTML)
        self.assertIn('showFormBlock', HTML)
        self.assertIn('@keyframes form-group-out', HTML)
        self.assertIn('data-section="source"', HTML)
        self.assertIn('initCollapsibleSections', HTML)
        self.assertIn('configResetLayout', HTML)
        self.assertIn('configRestoreToggle', HTML)
        self.assertIn('configWorkspace', HTML)
        self.assertIn('configDeletedToggle', HTML)
        self.assertIn('configDeletedList', HTML)
        self.assertIn('configTemplateSelect', HTML)
        self.assertIn('configCreateBlock', HTML)
        self.assertIn('config-canvas', HTML)
        self.assertIn('setConfigBlockState', HTML)
        self.assertIn('createCustomConfigBlock', HTML)
        self.assertIn('duplicateConfigBlock', HTML)
        self.assertIn('focusConfigBlock', HTML)
        self.assertIn('startConfigBlockDrag', HTML)
        self.assertIn('configDragBlocked', HTML)
        self.assertIn('removeConfigBlockForever', HTML)
        self.assertIn('positionNewConfigBlock', HTML)
        self.assertIn('finishConfigDrag', HTML)
        self.assertIn('fpv-config-layout-v1', HTML)
        self.assertIn('config-trash', HTML)
        self.assertNotIn('block-actions', HTML)
        self.assertNotIn('block-resize', HTML)
        self.assertIn('config-block-deleted', HTML)
        self.assertNotIn('⠿', HTML)
        self.assertIn('Compact form layer', HTML)
        self.assertIn('prefers-reduced-motion', HTML)
        self.assertIn('color-mix', HTML)

    def test_model_paths_are_confined_to_pt_files(self):
        self.assertIsNotNone(resolve_model_path("best.pt"))
        self.assertIsNone(resolve_model_path("../secret.pt"))
        self.assertIsNone(resolve_model_path("best.onnx"))
        self.assertIsNone(model_upload_path("../evil.pt"))
        self.assertIsNone(model_upload_path("weights.onnx"))

    def test_model_inference_settings_are_clamped(self):
        state = normalize_control({
            "device": -99,
            "use_half": False,
            "conf": 4,
            "img_size_roi": 1,
            "img_size_full": 99999,
            "max_det": 9999,
        })
        self.assertEqual(state["device"], -1)
        self.assertFalse(state["use_half"])
        self.assertEqual(state["conf"], 1.0)
        self.assertEqual(state["img_size_roi"], 128)
        self.assertEqual(state["img_size_full"], 4096)
        self.assertEqual(state["max_det"], 300)

    def test_netron_starts_only_for_confined_model(self):
        calls = []
        fake_netron = types.SimpleNamespace(
            status=lambda address: False,
            stop=lambda address: calls.append(("stop", address)),
            start=lambda path, address, browse: calls.append(("start", path, address, browse)),
        )
        previous = dict(ui_server.NETRON_STATE)
        try:
            with patch.dict(sys.modules, {"netron": fake_netron}):
                payload = ui_server.ensure_netron("best.pt")
            self.assertEqual(payload["url"], "/netron/")
            self.assertEqual(calls[0][0], "stop")
            self.assertEqual(calls[1][0], "start")
            self.assertEqual(calls[1][3], False)
        finally:
            ui_server.NETRON_STATE.clear()
            ui_server.NETRON_STATE.update(previous)

    def test_udp_source_modes_and_formats_are_available(self):
        state = normalize_control({"source_mode": "udp_dump", "file_path": "camera.ts"})
        self.assertEqual(state["source_mode"], "udp_dump")
        for mode in ("udp_mik_live", "udp_delimited_live", "udp_custom_live", "udp_delimited_file"):
            self.assertEqual(normalize_control({"source_mode": mode})["source_mode"], mode)
        self.assertIn(".ts", INPUT_EXTENSIONS)
        self.assertIn(".h264", INPUT_EXTENSIONS)
        self.assertIn(".udp", INPUT_EXTENSIONS)
        self.assertIn(".avi", INPUT_EXTENSIONS)
        self.assertIn("", INPUT_EXTENSIONS)
        self.assertNotIn(".pcap", INPUT_EXTENSIONS)
        self.assertIn('<option value="udp_mik_live">', HTML)
        self.assertIn('<option value="udp_delimited_live">', HTML)
        self.assertIn('<option value="udp_custom_live">', HTML)
        self.assertIn('<option value="udp_dump">', HTML)
        self.assertIn('<option value="udp_delimited_file">', HTML)
        self.assertIn('<option value="512x640">', HTML)
        self.assertIn('id="separatorByte"', HTML)
        self.assertIn('id="frameEncoding"', HTML)
        self.assertIn('id="probeUdp"', HTML)
        self.assertIn('id="udpProbeDetails"', HTML)
        self.assertIn('id="packetPreset"', HTML)
        self.assertIn('<option value="auto">Автоопределение</option>', HTML)
        self.assertIn('id="packetConstructorFields"', HTML)
        self.assertIn('id="packetByteMap"', HTML)
        self.assertIn('id="packetFieldList"', HTML)
        self.assertIn('id="addPacketField"', HTML)
        self.assertIn('data-action="remove"', HTML)
        self.assertIn('id="packetHeaderSize"', HTML)
        self.assertIn("postJson('/api/udp-probe'", HTML)
        self.assertIn('Адрес привязки (обычно 0.0.0.0)', HTML)

    def test_custom_udp_packet_schema_is_normalized(self):
        state = normalize_control({
            "source_mode": "udp_custom_live",
            "packet_preset": "custom",
            "packet_schema": {
                "assembly": "datagram",
                "header_size": 12,
                "flags_offset": 99,
                "start_mask": "0x80",
            },
        })
        self.assertEqual(state["packet_preset"], "custom")
        self.assertEqual(state["packet_schema"]["assembly"], "datagram")
        self.assertEqual(state["packet_schema"]["header_size"], 12)
        self.assertEqual(state["packet_schema"]["flags_offset"], -1)
        self.assertEqual(state["packet_schema"]["start_mask"], 0x80)

    def test_visual_packet_layout_controls_parser_offsets(self):
        state = normalize_control({
            "source_mode": "udp_custom_live",
            "packet_preset": "custom",
            "packet_layout": [
                {"role": "skip", "size": 2, "label": "magic"},
                {"role": "field", "size": 2, "label": "temperature"},
                {"role": "flags", "size": 1, "label": "flags"},
                {"role": "sequence", "size": 3, "label": "frame"},
            ],
        })
        self.assertEqual(state["packet_schema"]["header_size"], 8)
        self.assertEqual(state["packet_schema"]["flags_offset"], 4)
        self.assertEqual(state["packet_schema"]["sequence_offset"], 5)
        self.assertEqual(state["packet_schema"]["sequence_size"], 3)
        self.assertEqual(
            state["packet_schema"]["read_fields"],
            [{"name": "temperature", "offset": 2, "size": 2}],
        )

    def test_custom_frame_size_and_fps_are_supported(self):
        state = normalize_control({"quality": "1536x864", "fps": 47})
        self.assertEqual(state["quality"], "1536x864")
        self.assertEqual(state["fps"], 47)
        self.assertEqual(parse_quality("9000x2"), (8192, 16))
        self.assertIn('id="frameWidth"', HTML)
        self.assertIn('id="frameHeight"', HTML)
        self.assertIn('id="fps" type="number"', HTML)
        self.assertIn('<option value="custom">Произвольный</option>', HTML)
        self.assertIn('udp_raw_gray16_40404', HTML)

    def test_large_upload_uses_streaming_ui_with_progress(self):
        self.assertIn("new XMLHttpRequest()", HTML)
        self.assertIn('id="uploadProgress"', HTML)
        self.assertIn('id="cancelUpload"', HTML)
        self.assertNotIn('accept=".mp4,.avi,.mov,.mkv,.m4v"', HTML)
        self.assertNotIn("new FormData()", HTML)

    def test_copy_exact_streams_only_requested_bytes(self):
        source = BytesIO(b"abcdefgh")
        target = BytesIO()
        self.assertEqual(copy_exact(source, target, 6, chunk_size=2), 6)
        self.assertEqual(target.getvalue(), b"abcdef")
        self.assertEqual(source.read(), b"gh")

    def test_upload_path_accepts_extensionless_dump_and_rejects_traversal(self):
        self.assertEqual(input_upload_path("1785156788883112336").name, "1785156788883112336")
        self.assertEqual(input_upload_path("camera.vendor-format").name, "camera.vendor-format")
        self.assertIsNone(input_upload_path("../dump"))
        self.assertIsNone(input_upload_path(r"..\dump"))

    def test_input_list_keeps_unknown_video_and_treats_transport_stream_as_video(self):
        from tempfile import TemporaryDirectory

        old_input = ui_server.INPUT_DIR
        try:
            with TemporaryDirectory() as tmp:
                root = Path(tmp)
                (root / "clip.ts").write_bytes(b"ts")
                (root / "clip.vendor").write_bytes(b"video")
                (root / "extensionless-video").write_bytes(b"\x1aE\xdf\xa3video")
                mik_header = (
                    (59004).to_bytes(2, "little")
                    + (8).to_bytes(2, "little")
                    + bytes((0, 2, 7, 0))
                    + (256).to_bytes(4, "little")
                )
                (root / "extensionless-mik").write_bytes(mik_header)
                (root / "camera.udp").write_bytes(b"dump")
                (root / ".partial.upload").write_bytes(b"partial")
                ui_server.INPUT_DIR = root

                rows = {row["name"]: row["kind"] for row in ui_server.input_video_files()}

                self.assertEqual(rows["clip.ts"], "video")
                self.assertEqual(rows["clip.vendor"], "video")
                self.assertEqual(rows["extensionless-video"], "video")
                self.assertEqual(rows["extensionless-mik"], "udp_dump")
                self.assertEqual(rows["camera.udp"], "udp_dump")
                self.assertNotIn(".partial.upload", rows)
        finally:
            ui_server.INPUT_DIR = old_input

    def test_tail_text_returns_last_lines(self):
        from tempfile import TemporaryDirectory
        from pathlib import Path

        with TemporaryDirectory() as tmp:
            log = Path(tmp) / "main.log"
            log.write_text("one\ntwo\nthree\n", encoding="utf-8")
            self.assertEqual(tail_text(log, 2), "two\nthree")

    def test_read_json_file_returns_empty_dict_for_missing_file(self):
        from tempfile import TemporaryDirectory
        from pathlib import Path

        with TemporaryDirectory() as tmp:
            self.assertEqual(read_json_file(Path(tmp) / "missing.json"), {})

    def test_latest_file_returns_newest_match(self):
        from tempfile import TemporaryDirectory
        from pathlib import Path
        import os

        with TemporaryDirectory() as tmp:
            root = Path(tmp)
            old = root / "out_1.mp4"
            new = root / "out_2.mp4"
            old.write_text("old", encoding="utf-8")
            new.write_text("new", encoding="utf-8")
            os.utime(old, (1, 1))
            os.utime(new, (2, 2))
            self.assertEqual(latest_file(root, "*.mp4"), new)

    def test_archive_path_rejects_path_traversal(self):
        from tempfile import TemporaryDirectory
        from pathlib import Path

        with TemporaryDirectory() as tmp:
            root = Path(tmp)
            self.assertIsNone(archive_path(root, "../evil.mp4"))

    def test_archive_path_accepts_url_encoded_name(self):
        from tempfile import TemporaryDirectory
        from pathlib import Path

        with TemporaryDirectory() as tmp:
            root = Path(tmp)
            self.assertEqual(archive_path(root, "clip%201.mp4"), root / "clip 1.mp4")

    def test_archive_path_rejects_symlink_outside_archive(self):
        from tempfile import TemporaryDirectory

        with TemporaryDirectory() as tmp:
            root = Path(tmp) / "archive"
            outside = Path(tmp) / "secret.mp4"
            root.mkdir()
            outside.write_bytes(b"secret")
            link = root / "public.mp4"
            try:
                os.symlink(outside, link)
            except (OSError, NotImplementedError):
                self.skipTest("symlinks unavailable")
            self.assertIsNone(archive_path(root, "public.mp4"))

    def test_json_body_limit_rejects_oversized_request(self):
        handler = ui_server.Handler.__new__(ui_server.Handler)
        handler.headers = {"Content-Length": str(MAX_JSON_BODY_BYTES + 1)}
        with self.assertRaisesRegex(ValueError, "too large"):
            handler.read_json_body()

    def test_udp_probe_path_accepts_capture_and_rejects_traversal(self):
        from tempfile import TemporaryDirectory

        with TemporaryDirectory() as tmp:
            root = Path(tmp)
            self.assertEqual(udp_probe_path(root, "capture.udp"), root / "capture.udp")
            self.assertEqual(udp_probe_path(root, "capture.json"), root / "capture.json")
            self.assertIsNone(udp_probe_path(root, "../capture.udp"))
            self.assertIsNone(udp_probe_path(root, "capture.mp4"))

    def test_archive_files_lists_mp4_metadata(self):
        from tempfile import TemporaryDirectory
        from pathlib import Path
        import os

        with TemporaryDirectory() as tmp:
            root = Path(tmp)
            video = root / "clip.mp4"
            video.write_bytes(b"1234")
            os.utime(video, (3, 3))

            self.assertEqual(archive_files(root)[0]["name"], "clip.mp4")
            self.assertEqual(archive_files(root)[0]["size"], 4)

    def test_cleanup_empty_recordings_removes_only_broken_mp4(self):
        from tempfile import TemporaryDirectory

        with TemporaryDirectory() as tmp:
            root = Path(tmp)
            broken = root / "broken.mp4"
            valid = root / "valid.mp4"
            broken.write_bytes(b"x" * 44)
            valid.write_bytes(b"x" * 1024)

            self.assertEqual(cleanup_empty_recordings(root), 1)
            self.assertFalse(broken.exists())
            self.assertTrue(valid.exists())

    def test_h264_cache_path_tracks_source_version(self):
        from tempfile import TemporaryDirectory
        from pathlib import Path

        with TemporaryDirectory() as tmp:
            root = Path(tmp)
            video = root / "clip.mp4"
            video.write_bytes(b"1234")

            cache = h264_cache_path(video, root / ".downloads")
            self.assertEqual(cache.parent, root / ".downloads")
            self.assertTrue(cache.name.endswith(".h264.mp4"))
            self.assertIn("clip", cache.name)
            self.assertIn("4", cache.name)

    def test_content_disposition_keeps_utf8_filename(self):
        header = content_disposition("тест.mp4")
        self.assertIn("attachment;", header)
        self.assertIn("filename*=", header)
        self.assertIn("%D1%82%D0%B5%D1%81%D1%82.mp4", header)

    def test_active_video_name_uses_marker_only_when_file_exists(self):
        from tempfile import TemporaryDirectory
        from pathlib import Path

        with TemporaryDirectory() as tmp:
            root = Path(tmp)
            marker = root / ".active_video"
            marker.write_text("clip.mp4", encoding="utf-8")
            self.assertIsNone(active_video_name(root, marker))
            (root / "clip.mp4").write_bytes(b"1234")
            self.assertEqual(active_video_name(root, marker), "clip.mp4")

    def test_control_payload_clears_stale_active_marker(self):
        from tempfile import TemporaryDirectory

        old_out = ui_server.OUT_DIR
        old_marker = ui_server.ACTIVE_VIDEO_PATH
        old_process = ui_server.CONTROL_PROCESS
        try:
            with TemporaryDirectory() as tmp:
                root = Path(tmp)
                marker = root / ".active_video"
                (root / "stale.mp4").write_bytes(b"1234")
                marker.write_text("stale.mp4", encoding="utf-8")
                ui_server.OUT_DIR = root
                ui_server.ACTIVE_VIDEO_PATH = marker
                ui_server.CONTROL_PROCESS = None

                payload = ui_server.control_payload()

                self.assertFalse(payload["running"])
                self.assertEqual(payload["active_video"], "")
                self.assertFalse(marker.exists())
        finally:
            ui_server.OUT_DIR = old_out
            ui_server.ACTIVE_VIDEO_PATH = old_marker
            ui_server.CONTROL_PROCESS = old_process

    def test_parse_range_header_supports_suffix_range(self):
        self.assertEqual(parse_range_header("bytes=-4", 10), (6, 9))

    def test_parse_range_header_rejects_out_of_range(self):
        self.assertIsNone(parse_range_header("bytes=20-30", 10))

    def test_parse_source_reads_opened_source_log(self):
        log = "[entrypoint] source=0 backend=json\nOpened source: 0 (camera)\n"
        self.assertEqual(parse_source(log), "0 (camera)")

    def test_parse_perf_reads_realtime_pass_counters(self):
        log = "[perf] fps~58.0 iter p50=3.0 p95=28.0 | yolo p50=25.0 p95=34.0 skip=4 pass=120 analysisEvery=2\n"
        perf = parse_perf(log)
        self.assertEqual(perf["fps"], "58.0")
        self.assertEqual(perf["skip"], "4")
        self.assertEqual(perf["pass"], "120")
        self.assertEqual(perf["analysis_every"], "2")


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