Miayui
c812a1a595
Integrate the remote firmware workbench, OTA delivery, and timed mode schedules while preserving the local N7 scene and runtime changes.
105 lines
3.6 KiB
Python
105 lines
3.6 KiB
Python
"""Exercise real pointer/click dispatch on the complete console in Chromium."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import re
|
|
import shutil
|
|
import subprocess
|
|
import threading
|
|
from http import HTTPStatus
|
|
from http.server import ThreadingHTTPServer
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
from urllib.parse import urlparse
|
|
|
|
import pytest
|
|
|
|
from cognitive_agent.application.seat_service import SeatDemoService
|
|
from cognitive_agent.auth import AuthConfig, AuthRuntime
|
|
from cognitive_agent.server import REPOSITORY_ROOT, SeatDemoHandler
|
|
|
|
|
|
def test_mode_rail_real_browser_interactions(tmp_path: Path) -> None:
|
|
node = shutil.which("node")
|
|
if not node:
|
|
pytest.skip("Node.js is required for the browser interaction regression")
|
|
env = dict(os.environ)
|
|
bundled_modules = Path(node).resolve().parent.parent / "node_modules"
|
|
if bundled_modules.is_dir():
|
|
env["NODE_PATH"] = os.pathsep.join(
|
|
item for item in (str(bundled_modules), env.get("NODE_PATH")) if item
|
|
)
|
|
available = subprocess.run(
|
|
[node, "-e", "require.resolve('playwright')"],
|
|
env=env,
|
|
capture_output=True,
|
|
encoding="utf-8",
|
|
check=False,
|
|
)
|
|
if available.returncode:
|
|
pytest.skip("Playwright is required for the real browser regression")
|
|
with patch.dict(
|
|
os.environ,
|
|
{
|
|
"COGNITIVE_AGENT_HARDWARE_EXECUTION_ENABLED": "false",
|
|
"COGNITIVE_AGENT_VEHICLE_CAN_ENABLED": "false",
|
|
"COGNITIVE_AGENT_WIRED_PASSENGER_CAN_ENABLED": "false",
|
|
"COGNITIVE_AGENT_VEHICLE_MQTT_ENABLED": "false",
|
|
"COGNITIVE_AGENT_EXECUTION_MODE": "dry_run",
|
|
},
|
|
):
|
|
service = SeatDemoService(REPOSITORY_ROOT)
|
|
service.artifact_root = tmp_path / "artifacts"
|
|
|
|
class BrowserHandler(SeatDemoHandler):
|
|
auth = AuthRuntime(AuthConfig(False, "", "", b"", 3600, 5, 300, 120, False))
|
|
|
|
def do_GET(self) -> None:
|
|
path = urlparse(self.path).path
|
|
if path == "/api/runtime":
|
|
runtime = self.service.runtime_info()
|
|
runtime["execution_mode"] = "dry_run"
|
|
self._json(runtime)
|
|
return
|
|
if path in {"/api/hardware/seat-state", "/api/hardware/vehicle-state"}:
|
|
self._json({"status": "unavailable", "hardware_execution_allowed": False})
|
|
return
|
|
super().do_GET()
|
|
|
|
def do_POST(self) -> None:
|
|
path = urlparse(self.path).path
|
|
if re.fullmatch(r"/api/mode-profiles/[a-z0-9_-]+/execute", path):
|
|
super().do_POST()
|
|
return
|
|
# Assistant requests are intercepted; unexpected writes fail closed.
|
|
self._json({"error": "browser_test_write_forbidden"}, HTTPStatus.FORBIDDEN)
|
|
|
|
def log_message(self, *_args: object) -> None:
|
|
pass
|
|
|
|
BrowserHandler.service = service
|
|
server = ThreadingHTTPServer(("127.0.0.1", 0), BrowserHandler)
|
|
thread = threading.Thread(target=server.serve_forever, daemon=True)
|
|
thread.start()
|
|
try:
|
|
completed = subprocess.run(
|
|
[
|
|
node,
|
|
str(Path(__file__).with_name("mode_rail_browser.cjs")),
|
|
f"http://127.0.0.1:{server.server_port}",
|
|
str(tmp_path),
|
|
],
|
|
env=env,
|
|
capture_output=True,
|
|
encoding="utf-8",
|
|
timeout=120,
|
|
check=False,
|
|
)
|
|
assert completed.returncode == 0, completed.stdout + completed.stderr
|
|
finally:
|
|
server.shutdown()
|
|
server.server_close()
|
|
thread.join(timeout=3)
|
|
service.close()
|