feat(fleet): add board observation center with live sources and maps

Add adaptive per-vehicle layouts, full-panel dragging, source controls and automatic preview recovery for RealSense, Insta360 X4 and XGRIDS K1 observation views. Integrate cached Cesium map layers through the private NAS gateway link, retain bounded sensor command receipts, and document validation and operational handoff.
This commit is contained in:
DCCONSTRUCTIONS
2026-09-10 22:12:05 +03:00
parent 5f75afb720
commit be58d589e2
48 changed files with 1575 additions and 83 deletions
+60
View File
@@ -0,0 +1,60 @@
import importlib.util
import plistlib
from pathlib import Path
from unittest.mock import Mock
import pytest
SCRIPT = Path(__file__).parents[1] / "scripts/manage_map_gateway_link.py"
spec = importlib.util.spec_from_file_location("map_gateway_link", SCRIPT)
assert spec and spec.loader
link = importlib.util.module_from_spec(spec)
spec.loader.exec_module(link)
def test_link_is_local_only_and_cannot_execute_remote_commands(tmp_path):
value = plistlib.loads(link.payload("operator@nas.example", tmp_path))
argv = value["ProgramArguments"]
assert argv[-3:] == ["-L", "127.0.0.1:18103:127.0.0.1:18103", "operator@nas.example"]
for flag in ("-N", "-T", "StrictHostKeyChecking=yes", "BatchMode=yes",
"ExitOnForwardFailure=yes", "ForwardAgent=no", "PermitLocalCommand=no"):
assert flag in argv
assert "-R" not in argv and "-D" not in argv
assert value["KeepAlive"] and value["ThrottleInterval"] == 30
for target in ("-oProxyCommand=evil", "operator@host;touch x", "operator@host\ncommand"):
with pytest.raises(ValueError):
link.payload(target, tmp_path)
def test_stale_plan_cannot_change_runtime(monkeypatch, tmp_path):
monkeypatch.setattr(link.Path, "home", lambda: tmp_path)
run = Mock()
monkeypatch.setattr(link.subprocess, "run", run)
monkeypatch.setattr("sys.argv", [str(SCRIPT), "apply", "--target", "operator@nas.example",
"--expected-current-sha256", "absent",
"--expected-desired-sha256", "stale"])
with pytest.raises(ValueError, match="plan changed"):
link.main()
run.assert_not_called()
assert not (tmp_path / "Library").exists()
def test_failed_first_install_removes_candidate_agent(monkeypatch, tmp_path):
monkeypatch.setattr(link.Path, "home", lambda: tmp_path)
runtime = tmp_path / "Library/Logs/NODEDC/MissionCore/map-gateway-link"
expected = link.digest(link.payload("operator@nas.example", runtime))
monkeypatch.setattr("sys.argv", [str(SCRIPT), "apply", "--target", "operator@nas.example",
"--expected-current-sha256", "absent",
"--expected-desired-sha256", expected])
monkeypatch.setattr(link, "loaded", lambda: False)
unload = Mock()
monkeypatch.setattr(link, "unload", unload)
socket = Mock()
socket.__enter__ = Mock(return_value=Mock())
socket.__exit__ = Mock(return_value=False)
monkeypatch.setattr(link.socket, "socket", lambda: socket)
monkeypatch.setattr(link.subprocess, "run", Mock(side_effect=RuntimeError("bootstrap failed")))
with pytest.raises(RuntimeError, match="bootstrap failed"):
link.main()
unload.assert_called_once()
assert not (tmp_path / "Library/LaunchAgents" / f"{link.LABEL}.plist").exists()