from __future__ import annotations import json import sys from io import BytesIO from pathlib import Path import pytest from k1link.web.xgrids_k1_camera import ( CAMERA_MEDIA_TYPE, XgridsK1CameraGateway, _build_ffmpeg_argv, _read_mp4_box, ) from k1link.web.xgrids_k1_facade import ( XGRIDS_K1_COMPATIBILITY_PROFILE_ID, XGRIDS_K1_PLUGIN_ID, CameraPreviewSelectRequest, CameraPreviewStopRequest, XgridsK1CompatibilityService, ) def _fake_ffmpeg(tmp_path: Path) -> Path: executable = tmp_path / "fake-ffmpeg" executable.write_text( f"#!{sys.executable}\n" "import sys, time\n" "def box(kind, payload=b''):\n" " return (8 + len(payload)).to_bytes(4, 'big') + kind + payload\n" "payload = (box(b'ftyp', b'isom') + box(b'moov') + " "box(b'moof') + box(b'mdat', b'frame'))\n" "sys.stdout.buffer.write(payload)\n" "sys.stdout.buffer.flush()\n" "time.sleep(10)\n", encoding="utf-8", ) executable.chmod(0o700) return executable def _gateway( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> XgridsK1CameraGateway: monkeypatch.setenv("MISSIONCORE_FFMPEG_BINARY", str(_fake_ffmpeg(tmp_path))) return XgridsK1CameraGateway(tmp_path, XGRIDS_K1_PLUGIN_ID) def test_camera_selection_is_exclusive_and_hides_device_transport( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: gateway = _gateway(tmp_path, monkeypatch) try: left = gateway.select("sensor.camera.left", "192.168.8.52") assert left["active_source_id"] == "sensor.camera.left" assert left["generation"] == 1 assert left["activation"]["max_active"] == 1 assert left["delivery"]["kind"] == "mse-fmp4-websocket" assert left["delivery"]["media_type"] == CAMERA_MEDIA_TYPE serialized = json.dumps(left) assert "192.168.8.52" not in serialized assert "rtsp://" not in serialized assert "chn_left_main" not in serialized right = gateway.select("sensor.camera.right", "192.168.8.52") assert right["active_source_id"] == "sensor.camera.right" assert right["generation"] == 2 assert right["delivery"]["url"].endswith("/camera-preview/2") with pytest.raises(ValueError, match="generation"): gateway.stop(1) assert gateway.snapshot()["active_source_id"] == "sensor.camera.right" stopped = gateway.stop(2) assert stopped["phase"] == "idle" assert stopped["delivery"] is None finally: gateway.close() def test_camera_ffmpeg_command_is_allowlisted_copy_remux() -> None: argv = _build_ffmpeg_argv( Path("/trusted/ffmpeg"), "10.0.0.24", "sensor.camera.left", ) assert argv[0] == "/trusted/ffmpeg" assert argv[argv.index("-i") + 1] == ( "rtsp://10.0.0.24:8554/live/chn_left_main" ) assert argv[argv.index("-c:v") + 1] == "copy" assert argv[argv.index("-allowed_media_types") + 1] == "video" assert argv[argv.index("-flush_packets") + 1] == "1" assert "-c:v" in argv assert ";" not in " ".join(argv) with pytest.raises(ValueError, match="private IPv4"): _build_ffmpeg_argv( Path("/trusted/ffmpeg"), "example.com", "sensor.camera.left", ) def test_iso_bmff_reader_preserves_complete_boxes() -> None: def box(kind: bytes, payload: bytes = b"") -> bytes: return (8 + len(payload)).to_bytes(4, "big") + kind + payload stream = BytesIO(box(b"ftyp", b"isom") + box(b"moov") + box(b"moof")) assert _read_mp4_box(stream) == (b"ftyp", box(b"ftyp", b"isom")) assert _read_mp4_box(stream) == (b"moov", box(b"moov")) assert _read_mp4_box(stream) == (b"moof", box(b"moof")) oversized = (9 * 1024 * 1024).to_bytes(4, "big") + b"mdat" with pytest.raises(ValueError, match="unbounded"): _read_mp4_box(BytesIO(oversized)) def test_gateway_emits_init_and_complete_media_segments_without_transcoding( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: gateway = _gateway(tmp_path, monkeypatch) try: state = gateway.select("sensor.camera.right", "192.168.1.20") lease = gateway.open_delivery(state["generation"]) init_segment = lease.segments.get(timeout=3) media_segment = lease.segments.get(timeout=3) assert init_segment is not None and init_segment[0] == "init" assert b"ftyp" in init_segment[1] and b"moov" in init_segment[1] assert media_segment is not None and media_segment[0] == "media" assert b"moof" in media_segment[1] and b"mdat" in media_segment[1] gateway.mark_streaming(lease) assert gateway.snapshot()["phase"] == "streaming" gateway.release_delivery(lease, client_closed=True) assert gateway.snapshot()["phase"] == "selected" finally: gateway.close() def test_service_publishes_two_dynamic_camera_rows_and_stale_stop_is_safe( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: monkeypatch.setenv("MISSIONCORE_FFMPEG_BINARY", str(_fake_ffmpeg(tmp_path))) service = XgridsK1CompatibilityService(tmp_path) try: with service._lock: service._k1_ip = "192.168.1.20" service._device_id = "device-k1-test" service._device_session_id = "device-session-test" service._device_session_opened_at = "2026-07-16T20:00:00Z" service._compatibility_attestation = { "firmware_version": "3.0.2", "topology": "direct-lan", "basis": "operator-attested", "observed_at": "2026-07-16T20:00:00Z", } state = service.select_camera_preview( CameraPreviewSelectRequest( source_id="sensor.camera.left", device_session_id="device-session-test", ) ) cameras = [ stream for stream in state["sensor_catalog"]["streams"] if stream.get("semantic_channel_id") == "camera.preview.live" ] assert len(cameras) == 2 assert sum(bool(stream["activation"]["selected"]) for stream in cameras) == 1 assert sum(stream["delivery"] is not None for stream in cameras) == 1 assert state["compatibility"]["profile_id"] == XGRIDS_K1_COMPATIBILITY_PROFILE_ID camera_contract = json.dumps( {"camera_preview": state["camera_preview"], "streams": cameras} ) assert "rtsp://" not in camera_contract assert "192.168.1.20" not in camera_contract generation = state["camera_preview"]["generation"] stopped = service.stop_camera_preview( CameraPreviewStopRequest( device_session_id="device-session-test", generation=generation, ) ) assert stopped["camera_preview"]["phase"] == "idle" with pytest.raises(ValueError, match="device-сессия"): service.select_camera_preview( CameraPreviewSelectRequest( source_id="sensor.camera.right", device_session_id="stale-session", ) ) finally: service.close()