vee
179 lines
6.9 KiB
Python
179 lines
6.9 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import tempfile
|
|
import threading
|
|
import urllib.error
|
|
import urllib.request
|
|
from http.server import ThreadingHTTPServer
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from cognitive_agent.application.calibration.service import CalibrationService
|
|
from cognitive_agent.auth import AuthConfig, AuthRuntime
|
|
from cognitive_agent.server import REPOSITORY_ROOT, SeatDemoHandler
|
|
|
|
|
|
def _auth_disabled() -> AuthRuntime:
|
|
return AuthRuntime(
|
|
AuthConfig(
|
|
enabled=False,
|
|
username="",
|
|
password_hash="",
|
|
session_secret=b"",
|
|
session_ttl_seconds=3600,
|
|
login_attempts=5,
|
|
login_window_seconds=300,
|
|
requests_per_minute=120,
|
|
trust_proxy_headers=False,
|
|
)
|
|
)
|
|
|
|
|
|
class TestSelfCalibrationApi:
|
|
@classmethod
|
|
def setup_class(cls) -> None:
|
|
cls.temporary_directory = tempfile.TemporaryDirectory()
|
|
cls.calibration = CalibrationService(
|
|
Path(cls.temporary_directory.name),
|
|
repository_root=REPOSITORY_ROOT,
|
|
)
|
|
cls.calibration.start()
|
|
cls.original_calibration = SeatDemoHandler.calibration
|
|
cls.original_auth = SeatDemoHandler.auth
|
|
SeatDemoHandler.calibration = cls.calibration
|
|
SeatDemoHandler.auth = _auth_disabled()
|
|
cls.server = ThreadingHTTPServer(("127.0.0.1", 0), SeatDemoHandler)
|
|
cls.thread = threading.Thread(target=cls.server.serve_forever, daemon=True)
|
|
cls.thread.start()
|
|
cls.base_url = f"http://127.0.0.1:{cls.server.server_port}"
|
|
|
|
@classmethod
|
|
def teardown_class(cls) -> None:
|
|
cls.server.shutdown()
|
|
cls.server.server_close()
|
|
cls.thread.join(timeout=2)
|
|
cls.calibration.close()
|
|
SeatDemoHandler.calibration = cls.original_calibration
|
|
SeatDemoHandler.auth = cls.original_auth
|
|
cls.temporary_directory.cleanup()
|
|
|
|
def request(
|
|
self,
|
|
path: str,
|
|
*,
|
|
method: str = "GET",
|
|
payload: dict[str, object] | None = None,
|
|
role: str | None = "admin",
|
|
) -> urllib.request.Request:
|
|
headers = {"Accept": "application/json"}
|
|
data = None
|
|
if role is not None:
|
|
headers["X-Cognitive-Agent-Role"] = role
|
|
if payload is not None:
|
|
headers["Content-Type"] = "application/json"
|
|
data = json.dumps(payload).encode("utf-8")
|
|
return urllib.request.Request(
|
|
f"{self.base_url}{path}",
|
|
method=method,
|
|
headers=headers,
|
|
data=data,
|
|
)
|
|
|
|
def test_admin_gate_protects_calibration_api(self) -> None:
|
|
with pytest.raises(urllib.error.HTTPError) as caught:
|
|
urllib.request.urlopen(
|
|
self.request("/api/calibration/state", role="user"),
|
|
timeout=3,
|
|
)
|
|
assert caught.value.code == 403
|
|
body = json.loads(caught.value.read().decode("utf-8"))
|
|
assert body["error"] == "admin_required"
|
|
|
|
def test_simulation_connect_move_and_stop(self) -> None:
|
|
with urllib.request.urlopen(
|
|
self.request(
|
|
"/api/calibration/can/connect",
|
|
method="POST",
|
|
payload={"can": {"mode": "simulation"}},
|
|
),
|
|
timeout=3,
|
|
) as response:
|
|
connected = json.loads(response.read().decode("utf-8"))["state"]
|
|
assert connected["transport"]["state"] == "online"
|
|
assert [axis["range"] for axis in connected["axes"]] == [480, 647, 917, 373]
|
|
|
|
with urllib.request.urlopen(
|
|
self.request(
|
|
"/api/calibration/move",
|
|
method="POST",
|
|
payload={"axis": 2, "permille": 650},
|
|
),
|
|
timeout=3,
|
|
) as response:
|
|
moving = json.loads(response.read().decode("utf-8"))["state"]
|
|
assert moving["controller"]["armed"]
|
|
|
|
with urllib.request.urlopen(
|
|
self.request("/api/calibration/stop", method="POST", payload={}),
|
|
timeout=3,
|
|
) as response:
|
|
stopped = json.loads(response.read().decode("utf-8"))["state"]
|
|
assert not stopped["controller"]["armed"]
|
|
|
|
def test_frontend_and_original_model_are_served_by_existing_service(self) -> None:
|
|
with urllib.request.urlopen(f"{self.base_url}/", timeout=3) as response:
|
|
html = response.read().decode("utf-8")
|
|
with urllib.request.urlopen(
|
|
f"{self.base_url}/self-calibration.js?v=calibration-6",
|
|
timeout=3,
|
|
) as response:
|
|
script = response.read().decode("utf-8")
|
|
with urllib.request.urlopen(
|
|
f"{self.base_url}/self-calibration.css?v=calibration-10",
|
|
timeout=3,
|
|
) as response:
|
|
styles = response.read().decode("utf-8")
|
|
with urllib.request.urlopen(
|
|
f"{self.base_url}/console-v2.js?v=desktop-refactor-31",
|
|
timeout=3,
|
|
) as response:
|
|
console_script = response.read().decode("utf-8")
|
|
with urllib.request.urlopen(
|
|
f"{self.base_url}/assets/models/n7_seat.glb",
|
|
timeout=3,
|
|
) as response:
|
|
model = response.read()
|
|
content_type = response.headers["Content-Type"]
|
|
content_security_policy = response.headers["Content-Security-Policy"]
|
|
|
|
assert 'href="/self-calibration.css?v=calibration-10"' in html
|
|
assert 'src="/self-calibration.js?v=calibration-6"' in html
|
|
assert 'data-calibration-page="live"' in script
|
|
assert 'data-calibration-view="live"' in script
|
|
assert 'data-calibration-page="hall"' not in script
|
|
assert 'data-calibration-page="stereo"' not in script
|
|
assert "座椅与双目联合标定" in script
|
|
assert 'class="v2-module-heading"' in script
|
|
assert 'SELF-CALIBRATION' in script
|
|
assert 'padding: var(--v2-module-title-block) 0 0' in styles
|
|
assert 'margin-inline: var(--v2-module-title-inline)' in styles
|
|
assert "实时座椅姿态" in script
|
|
assert "四轴实时控制" in script
|
|
assert "ZED 实时画面" in script
|
|
assert "双目位姿数据" not in script
|
|
assert 'class="v2-calibration-panel v2-calibration-mqtt-panel"' not in script
|
|
assert "grid-template-columns: minmax(420px, 1.06fr) minmax(380px, 0.94fr)" in styles
|
|
assert "标定仿真台架" in script
|
|
assert "X-Cognitive-Agent-Role" in script
|
|
assert "--v2-sidebar: #dae3f5" in styles
|
|
assert "--v2-sidebar-deep: #b6c7ea" in styles
|
|
assert "--v2-accent-strong: #1e386b" in styles
|
|
assert '["self-calibration", "self-repair"].includes(module)' in console_script
|
|
assert 'setRole("admin", { allowModuleTransition: true })' in console_script
|
|
assert len(model) == 845_992
|
|
assert content_type.startswith("model/gltf-binary")
|
|
assert "img-src 'self' data: blob:;" in content_security_policy
|
|
assert "connect-src 'self' blob:;" in content_security_policy
|