You cannot select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
450 lines
20 KiB
Python
450 lines
20 KiB
Python
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('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('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.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()
|