from __future__ import annotations import json import os import tempfile import threading import unittest import urllib.error import urllib.request from http.cookiejar import CookieJar from http.server import ThreadingHTTPServer from pathlib import Path from types import SimpleNamespace from unittest.mock import patch from cognitive_agent.auth import AuthConfig, AuthRuntime, hash_password from cognitive_agent.application.seat_service import REPOSITORY_ROOT, SeatDemoService from cognitive_agent.application.mode_profile import ModeProfileRegistry from cognitive_agent.application.self_extension import ( SelfExtensionService, SimulationGateway, ) from cognitive_agent.infrastructure.llm.openai_responses import OpenAIIntentError from cognitive_agent.infrastructure.can.access import HardwareAccessController from cognitive_agent.infrastructure.can.gateway import CanGatewayError from cognitive_agent.server import SeatDemoHandler class SeatWebApiTests(unittest.TestCase): username = "test-user" password = "public-demo-test-password" @classmethod def setUpClass(cls) -> None: cls.temporary_directory = tempfile.TemporaryDirectory() # Web regression tests must never inherit a developer's live bench # enablement, especially when local debug mode intentionally skips the # per-request approval token. with patch.dict( os.environ, { "COGNITIVE_AGENT_HARDWARE_EXECUTION_ENABLED": "false", "COGNITIVE_AGENT_VEHICLE_CAN_ENABLED": "false", "COGNITIVE_AGENT_WIRED_PASSENGER_CAN_ENABLED": "false", }, clear=False, ): service = SeatDemoService(REPOSITORY_ROOT) service.artifact_root = Path(cls.temporary_directory.name) cls.service = service cls.original_service = SeatDemoHandler.service SeatDemoHandler.service = service cls.original_auth = SeatDemoHandler.auth SeatDemoHandler.auth = AuthRuntime( AuthConfig( enabled=True, username=cls.username, password_hash=hash_password( cls.password, salt=b"0123456789abcdef", iterations=100_000, ), session_secret=b"t" * 64, session_ttl_seconds=3600, login_attempts=5, login_window_seconds=300, requests_per_minute=120, trust_proxy_headers=False, ) ) 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 tearDownClass(cls) -> None: cls.server.shutdown() cls.server.server_close() cls.thread.join(timeout=2) SeatDemoHandler.service = cls.original_service SeatDemoHandler.auth = cls.original_auth cls.temporary_directory.cleanup() def setUp(self) -> None: self.original_hardware_access = self.service.hardware_access self.service.hardware_access = HardwareAccessController() self.cookie_jar = CookieJar() self.opener = urllib.request.build_opener( urllib.request.HTTPCookieProcessor(self.cookie_jar) ) request = urllib.request.Request( f"{self.base_url}/api/auth/login", method="POST", headers={"Content-Type": "application/json"}, data=json.dumps( {"username": self.username, "password": self.password} ).encode("utf-8"), ) with self.opener.open(request, timeout=3) as response: payload = json.loads(response.read().decode("utf-8")) self.assertTrue(payload["authenticated"]) def tearDown(self) -> None: self.service.hardware_access = self.original_hardware_access def test_authentication_protects_console_and_api(self) -> None: unauthenticated = urllib.request.build_opener() with unauthenticated.open(f"{self.base_url}/", timeout=3) as response: login_html = response.read().decode("utf-8") final_url = response.geturl() self.assertIn("/login", final_url) self.assertIn("体验账号登录", login_html) with unauthenticated.open(f"{self.base_url}/network.js", timeout=3) as response: network_script = response.read().decode("utf-8") self.assertIn("CognitiveAgentNetwork", network_script) with self.assertRaises(urllib.error.HTTPError) as caught: unauthenticated.open(f"{self.base_url}/api/runtime", timeout=3) self.assertEqual(caught.exception.code, 401) def test_user_mode_delete_endpoint_removes_only_user_profiles(self) -> None: original_registry = self.service.mode_profiles with tempfile.TemporaryDirectory() as directory: root = Path(directory) preset_path = root / "common" / "preset" / "office.yaml" preset_path.parent.mkdir(parents=True) preset_path.write_text( """ id: office name: 办公模式 operations: - domain: climate capability: climate.set_temperature target_scope: vehicle parameters: temperature_c: 24 """.strip() + "\n", encoding="utf-8", ) registry = ModeProfileRegistry(root) user_mode = registry.save_mode( "K歌模式", [ { "domain": "ambient_light", "capability": "ambient_light.set_power", "target_scope": "vehicle", "parameters": {"state": "on"}, } ], vehicle_id="denza_n7", ) self.service.mode_profiles = registry try: request = urllib.request.Request( f"{self.base_url}/api/mode-profiles/{user_mode.id}?vehicle_id=denza_n7", method="DELETE", ) with self.opener.open(request, timeout=3) as response: payload = json.loads(response.read().decode("utf-8")) self.assertEqual(response.status, 200) self.assertEqual(payload["deleted_mode_profile"]["name"], "K歌模式") self.assertFalse(Path(user_mode.source).exists()) protected = urllib.request.Request( f"{self.base_url}/api/mode-profiles/office?vehicle_id=denza_n7", method="DELETE", ) with self.assertRaises(urllib.error.HTTPError) as caught: self.opener.open(protected, timeout=3) self.assertEqual(caught.exception.code, 403) error = json.loads(caught.exception.read().decode("utf-8")) self.assertEqual(error["error"], "preset_mode_protected") self.assertTrue(preset_path.is_file()) finally: self.service.mode_profiles = original_registry def test_mode_card_editor_endpoint_persists_generated_can_flow(self) -> None: original_registry = self.service.mode_profiles original_artifact_root = self.service.artifact_root with tempfile.TemporaryDirectory() as directory: base = Path(directory) preset_root = base / "presets" user_root = base / "user-modes" preset = preset_root / "vehicles" / "denza_n7" / "office.yaml" preset.parent.mkdir(parents=True) preset.write_text( """ id: office name: 办公模式 scope: vehicle_ids: [denza_n7] operations: - domain: climate capability: climate.set_temperature target_scope: passenger parameters: temperature_c: 24 """.strip() + "\n", encoding="utf-8", ) request_id = "req-mode-editor-api-001" instruction = "副驾空调温度设置为24度" artifact_root = base / "runs" request_dir = artifact_root / request_id request_dir.mkdir(parents=True) result = { "request_id": request_id, "utterance": instruction, "execution_mode": "dry_run", "control_transport": "wireless_mqtt", "function_configs": { "native_climate": { "spec": { "domain": "climate", "operations": [ { "id": "passenger_temperature", "capability": "climate.set_temperature", "target": {"zone": "passenger"}, "parameters": {"temperature_c": 24}, } ], } } }, "command_plans": { "native_climate": { "spec": { "commands": [ { "message": "Native_0x1DE", "signals": { "frame_id": "0x1DE", "data_hex": "00 00 00 00 00 18 00 00", }, } ] } } }, "execution_report": {"spec": {"results": []}}, } (request_dir / "result.json").write_text( json.dumps(result, ensure_ascii=False), encoding="utf-8" ) self.service.mode_profiles = ModeProfileRegistry( preset_root, user_root=user_root ) self.service.artifact_root = artifact_root try: request = urllib.request.Request( f"{self.base_url}/api/mode-profiles/office", method="POST", headers={"Content-Type": "application/json"}, data=json.dumps( { "name": "专注办公模式", "instruction": instruction, "source_request_id": request_id, "vehicle_id": "denza_n7", }, ensure_ascii=False, ).encode("utf-8"), ) with self.opener.open(request, timeout=3) as response: payload = json.loads(response.read().decode("utf-8")) self.assertEqual(response.status, 200) self.assertEqual(payload["storage"], "persistent_yaml") self.assertEqual(payload["mode_profile"]["origin"], "customized") self.assertEqual(payload["mode_profile"]["instruction"], instruction) self.assertTrue( (user_root / "vehicles" / "denza_n7" / "user" / "office.yaml").is_file() ) finally: self.service.mode_profiles = original_registry self.service.artifact_root = original_artifact_root def test_head_unit_display_sync_tracks_tasks_and_usb_catalog(self) -> None: task_request = urllib.request.Request( f"{self.base_url}/api/requests", method="POST", headers={"Content-Type": "application/json"}, data=json.dumps( { "utterance": "副驾有点热，有点闷", "provider": "local", "conversation_id": "conv-head-unit-display-sync", "vehicle_id": "denza_n7", "execution_mode": "dry_run", "client_type": "head_unit", }, ensure_ascii=False, ).encode("utf-8"), ) with self.opener.open(task_request, timeout=3) as response: task_result = json.loads(response.read().decode("utf-8")) peripheral = { "id": "cap-usb-light-on", "device_id": "usb-light-1", "device_name": "USB 氛围灯条", "device_type": "light", "type_label": "灯光外设", "action": "开启灯带", } extension_request = urllib.request.Request( f"{self.base_url}/api/extension/display-sync", method="POST", headers={"Content-Type": "application/json"}, data=json.dumps( {"device": {"id": "usb-light-1"}, "peripherals": [peripheral]}, ensure_ascii=False, ).encode("utf-8"), ) with self.opener.open(extension_request, timeout=3) as response: extension_result = json.loads(response.read().decode("utf-8")) with self.opener.open( f"{self.base_url}/api/vehicle-display-sync", timeout=3 ) as response: snapshot = json.loads(response.read().decode("utf-8")) self.assertEqual(task_result["status"], "succeeded") self.assertTrue(extension_result["success"]) self.assertEqual(snapshot["task"]["utterance"], "副驾有点热，有点闷") self.assertEqual(snapshot["task"]["result"]["request_id"], task_result["request_id"]) self.assertEqual(snapshot["extension"]["catalog"], [peripheral]) self.assertEqual(snapshot["extension"]["last_action"]["type"], "hardware_detected") self.assertIn("extension_health", snapshot) def test_static_console_and_runtime_endpoint(self) -> None: with self.opener.open(f"{self.base_url}/", timeout=3) as response: html = response.read().decode("utf-8") with self.opener.open(f"{self.base_url}/app.js?v=dual-seat-107", timeout=3) as response: app_script = response.read().decode("utf-8") with self.opener.open( f"{self.base_url}/gesture-control.js?v=android-pad-9", timeout=3 ) as response: gesture_script = response.read().decode("utf-8") with self.opener.open( f"{self.base_url}/console-v2.js?v=task-outcome-47", timeout=3 ) as response: console_script = response.read().decode("utf-8") with self.opener.open( f"{self.base_url}/console-v2.css?v=sync-layout-41", timeout=3 ) as response: console_styles = response.read().decode("utf-8") with self.opener.open( f"{self.base_url}/self-extension.js?v=sync-layout-26", timeout=3 ) as response: extension_script = response.read().decode("utf-8") with self.opener.open( f"{self.base_url}/self-extension.css?v=sync-layout-26", timeout=3 ) as response: extension_styles = response.read().decode("utf-8").replace("\r\n", "\n") with self.opener.open( f"{self.base_url}/service-worker.js", timeout=3 ) as response: service_worker = response.read().decode("utf-8") with self.opener.open( f"{self.base_url}/api/extension/health", timeout=3 ) as response: extension_health = json.loads(response.read().decode("utf-8")) with self.opener.open(f"{self.base_url}/api/runtime", timeout=3) as response: runtime = json.loads(response.read().decode("utf-8")) self.assertIn("511工程-智能座舱", html) self.assertIn('id="seatTwinCanvas"', html) self.assertIn('data-vehicle-model="byd-tang-2024-reference"', html) self.assertIn("整车孪生界面", html) self.assertIn("车内灯光", html) self.assertIn('id="interiorLightPower"', html) self.assertIn('id="ambientBrightness"', html) self.assertIn('id="ambientColorIndex"', html) self.assertIn('data-function="massage"', html) self.assertIn("CMD 0x1DE · FB 0x404 · CH1", html) self.assertIn("CMD 0x431 · FB 0x41A · CH1", html) self.assertIn("CMD 0x431 · FB 0x388 · CH1", html) self.assertNotIn("透明车体 · 实体设备", html) self.assertIn('src="/app.js?v=dual-seat-107"', html) self.assertIn('href="/styles.css?v=dual-seat-16"', html) self.assertIn('src="/gesture-control.js?v=android-pad-9"', html) self.assertNotIn('class="vehicle-system-hud"', html) self.assertIn('src="/console-v2.js?v=task-outcome-47"', html) self.assertIn('href="/console-v2.css?v=sync-layout-41"', html) self.assertIn('', html) self.assertIn('id="v2BootScreen"', html) self.assertIn('正在加载车机控制与实时状态', html) self.assertIn('src="/self-extension.js?v=sync-layout-26"', html) self.assertIn('href="/self-extension.css?v=sync-layout-26"', html) self.assertNotIn("self-calibration", html) self.assertNotIn("self-repair", html) self.assertNotIn("self-evolution", html) self.assertNotIn("滚动查看更多", html) self.assertIn('aria-label="执行车舱控制指令"', html) self.assertNotIn('"admin"', console_script) self.assertNotIn("v2RoleSelect", console_script) self.assertIn('sidebarToggle.id = "v2SidebarToggle"', console_script) self.assertIn('id = "v2ModuleSwitcher"', console_script) self.assertIn( 'panelHeading.classList.add("v2-operation-heading", "v2-module-heading")', console_script, ) self.assertIn('kicker.textContent = "REMOTE ORCHESTRATION · LIVE SYNC"', console_script) self.assertIn('fetch("/api/vehicle-display-sync"', console_script) self.assertIn('document.body.classList.add("v2-passive-display", "v2-interactive-sync")', console_script) self.assertIn('input.readOnly = false', console_script) self.assertIn('function buildTwinDeviceInteractions()', console_script) self.assertIn('id="v2SyncLogList"', console_script) self.assertIn('data-panel-toggle="devices"', console_script) self.assertIn('data-panel-toggle="extensions"', console_script) self.assertIn('id="vehicleExtensionSync"', console_script) self.assertNotIn( 'if (ui.activeModule === "self-extension" && ui.role !== "user")', console_script, ) self.assertNotIn( 'if (ui.activeModule === "self-extension" && role !== "user"', console_script, ) self.assertIn('511工程-智能座舱', console_script) self.assertIn('网页控制链路', console_script) self.assertIn('["self-orchestration", "self-extension"]', console_script) self.assertNotIn('label: "自诊断"', console_script) self.assertNotIn('runtimeControls.id = "v2RuntimeControls"', console_script) self.assertIn('function prepareRuntimeControls(originalForm)', console_script) self.assertIn('executionField.hidden = false', console_script) self.assertIn('insertVisibleControl(executionField, "v2-sync-execution-field")', console_script) self.assertIn('insertVisibleControl(simulationFaultField, "v2-sync-simulation-fault-field")', console_script) self.assertNotIn('class="v2-profile"', console_script) self.assertNotIn('class="v2-control-switch"', console_script) self.assertNotIn('class="v2-config-stack"', console_script) self.assertNotIn('id="v2GestureModelValue"', console_script) self.assertNotIn('id="v2VehicleState"', console_script) self.assertIn('actionRail.append(cameraToggle)', console_script) self.assertNotIn("function enhanceConfiguredDropdowns()", console_script) self.assertIn("function updateModeEdgeHints()", console_script) self.assertIn('classList.toggle("v2-twin-active", definition.index === 0)', console_script) self.assertIn('label: "执行结果与硬件回读"', console_script) self.assertIn("function keepRunButtonAvailable()", console_script) self.assertIn('window.setTimeout(() => $("#v2BootScreen")?.remove(), 220)', console_script) self.assertNotIn("function requestUserSafeExecutionMode()", console_script) self.assertNotIn('dryRunButton.click()', console_script) self.assertIn('const DEFAULT_EXECUTION_MODE = "hardware"', app_script) self.assertIn('preferredExecutionMode: DEFAULT_EXECUTION_MODE', app_script) self.assertIn('selectExecutionMode(modeAvailable ? preferredMode : "dry_run", { remember: false })', app_script) self.assertIn('class="mode-option active" id="hardwareMode"', html) self.assertIn('grid-template-columns: clamp(245px, 18.5vw, 310px)', console_styles) self.assertNotIn('.v2-control-switch {', console_styles) self.assertIn('.v2-module-switcher {', console_styles) self.assertIn('grid-template-columns: repeat(2, minmax(0, 1fr))', console_styles) self.assertIn( '.console-v2 .v2-module-heading.v2-module-heading h1', console_styles, ) self.assertIn('font-size: clamp(24px, 2vw, 31px)', console_styles) self.assertIn('--v2-module-title-block: clamp(12px, 2vh, 22px)', console_styles) self.assertIn('--v2-module-title-inline: clamp(20px, 2.3vw, 34px)', console_styles) self.assertIn( 'grid-template-rows: 36px clamp(236px, 33dvh, 246px) minmax(0, 1fr)', console_styles, ) self.assertIn( 'color-mix(in srgb, var(--v2-accent-soft) 86%, white)', console_styles, ) self.assertNotIn( '.cabin-channel-legend [data-channel="ch0"]', console_styles, ) self.assertIn('margin-inline: calc(0px - var(--v2-main-pad))', console_styles) self.assertIn('background: color-mix(in srgb, var(--v2-accent) 30%, var(--v2-accent-soft))', console_styles) self.assertIn('width: clamp(48px, 45%, 72px)', console_styles) self.assertIn('font-weight: 800', console_styles) self.assertIn('white-space: nowrap', console_styles) self.assertNotIn('.v2-gesture-model-field', console_styles) self.assertIn('.v2-inline-gesture .v2-camera-toggle', console_styles) self.assertIn('.v2-inline-gesture .v2-gesture-action-rail', console_styles) self.assertIn('grid-template-rows: 42px 10px 42px', console_styles) self.assertIn('background: var(--v2-accent);', console_styles) self.assertNotIn("gestureArmButton", console_styles) self.assertIn('.v2-inline-gesture #gestureStopButton', console_styles) self.assertIn('.v2-inline-gesture .v2-gesture-result-stack', console_styles) self.assertIn('.v2-inline-gesture .v2-gesture-execution-slot', console_styles) self.assertIn('.v2-page-rail', console_styles) self.assertNotIn('.v2-dropdown-menu', console_styles) self.assertIn('.v2-edge-fade-left', console_styles) self.assertIn('.v2-edge-fade-right', console_styles) self.assertIn('.mode-group-divider', console_styles) self.assertIn('.joint-twin-shell > .cabin-band', console_styles) self.assertIn('.console-v2 .v2-page-index', console_styles) self.assertIn('.joint-twin-shell .twin-readout', console_styles) self.assertIn('.joint-seat-panel .thermal-states', console_styles) self.assertIn('.joint-twin-shell .domain-readouts', console_styles) self.assertIn('grid-template-columns: repeat(2, minmax(0, 1fr))', console_styles) self.assertNotIn('.v2-config-section', console_styles) self.assertNotIn('.console-v2 .v2-config-card {', console_styles) self.assertIn('.console-v2.v2-sidebar-collapsed .workspace', console_styles) self.assertNotIn('id="hardwareAccessField"', html) self.assertNotIn('id="hardwareClientLabel"', html) self.assertNotIn('id="requestHardwareAccess"', html) self.assertNotIn('id="releaseHardwareAccess"', html) self.assertNotIn('id="approvalField"', html) self.assertNotIn('id="approvalToken"', html) self.assertNotIn("approval_token: approvalToken", app_script) self.assertNotIn("授权并真实执行", app_script) self.assertIn('const dynamicModeOrder = ["rocking_horse", "dynamic_cradle"]', app_script) self.assertIn('divider.className = "mode-group-divider"', app_script) self.assertIn("function isHardwareExecutionReady()", app_script) self.assertIn("function hardwareExecutionIssue(", app_script) self.assertNotIn( 'if (mode === "hardware" && !hardwareExecutionReady) return false', app_script, ) self.assertIn('data-control-transport="wired_can"', html) self.assertIn('data-control-transport="wireless_mqtt"', html) self.assertIn('control_transport: state.controlTransport', app_script) self.assertIn('const transportField = $("#vehicleTransportField", originalForm)', console_script) self.assertNotIn('runtimeControls.append(transportField)', console_script) self.assertIn('insertVisibleControl(transportField, "v2-sync-transport-field")', console_script) self.assertIn('.v2-sync-execution-field .execution-mode-options', console_styles) self.assertIn('grid-template-columns: repeat(3, minmax(0, 1fr))', console_styles) self.assertNotIn('class="v2-sync-protocol"', console_script) self.assertIn('.console-v2 .v2-transport-field .vehicle-transport-options', console_styles) self.assertIn('.v2-sync-transport-field .mode-option', console_styles) self.assertIn('overflow-wrap: anywhere', console_styles) self.assertIn('button.disabled = false', app_script) self.assertIn('button.setAttribute("aria-pressed", String(selected))', app_script) self.assertNotIn( 'availableTransports.find((item) => item.state_configured)?.id', app_script, ) self.assertIn("hardwareButton.disabled = false", app_script) self.assertIn( 'hardwareButton.dataset.executionState = hardwareExecutionReady ? "available" : "blocked"', app_script, ) self.assertNotIn( 'id="hardwareMode" data-execution-mode="hardware" disabled', html, ) self.assertIn('const seat = twin.selected === "driver" ? "driver" : "passenger"', app_script) self.assertNotIn('data-mode-command="seat driver vent low"', app_script) self.assertIn("CH0 副驾座椅与 CH1 主驾座椅独立收发、独立反馈", app_script) self.assertIn("CH1 主驾实车测试不启用无闭环的座椅位置手势", app_script) self.assertIn("车端 CAN 未启用，其他领域保持虚拟执行", app_script) self.assertIn("function extractReferenceWindowPanes", app_script) self.assertIn("function collectReferenceFrontLampMeshes", app_script) self.assertIn("function emphasizeReferenceWindowPanes", app_script) self.assertIn("function configureActiveSeatDepthLayers", app_script) self.assertIn("function addTwinGlassDepthMask", app_script) self.assertIn("function applyTwinVehicleTheme", app_script) self.assertIn("leadingMaxZ", app_script) self.assertIn("function updateReferenceWindowOutlineLayers", app_script) self.assertIn("vehicle.traverse((object) => object.layers.set(1))", app_script) self.assertIn('const categoryPriority = ["body", "glass", "wheel"', app_script) self.assertIn("function styleReferencePassiveGlass", app_script) self.assertIn("function styleReferencePassiveRearExterior", app_script) self.assertIn('twinRole: "reference_passive_rear_bumper"', app_script) self.assertNotIn('"cognitive-agent:demo-role-change"', console_script) self.assertIn('resultStack.id = "v2GestureResultStack"', console_script) self.assertIn('executionSlot.id = "v2GestureExecutionSlot"', console_script) self.assertIn('statusTarget.append(requestStatus)', console_script) self.assertIn('root.id = "v2SelfExtension"', extension_script) self.assertIn('data-extension-mode="study"', extension_script) self.assertIn('data-extension-mode="control"', extension_script) self.assertIn('外设玩具能力学习', extension_script) self.assertIn('外设玩具与车端联动', extension_script) self.assertGreaterEqual(extension_script.count('class="v2-module-heading"'), 2) self.assertGreaterEqual(extension_script.count('SELF-EXTENSION'), 2) self.assertIn('id="extensionExecuteButton"', extension_script) self.assertIn('data-lucide="play"', extension_script) self.assertNotIn('id="extensionControlHealth"', extension_script) self.assertIn('data-extension-connected="${Boolean(selected)}"', extension_script) self.assertIn('红外学习器', extension_script) self.assertIn('id="extensionStudyHealthState"', extension_script) self.assertIn('id="extensionVehicleState">CHECK', extension_script) self.assertIn('id="extensionResetButton"', extension_script) self.assertNotIn('id="extensionConnectionButton"', extension_script) self.assertNotIn('id="extensionStatusDialog"', extension_script) self.assertIn('/api/extension/${path}', extension_script) self.assertIn("HEALTH_POLL_INTERVAL_MS", extension_script) self.assertIn("HEALTH_REQUEST_TIMEOUT_MS", extension_script) self.assertIn("healthState.listening", extension_script) self.assertIn("healthState.peer", extension_script) self.assertIn('healthState.message || "GSCU1B 尚未连接"', extension_script) self.assertIn('window.addEventListener("pageshow"', extension_script) self.assertIn('[data-module="self-extension"]', extension_styles) self.assertNotIn('[data-module="self-extension"] .v2-role-select', extension_styles) self.assertIn('.v2-extension-composer .primary-action {', extension_styles) self.assertNotIn('--v2-accent:', extension_styles) self.assertIn('.v2-sidebar-collapsed[data-module="self-extension"] .v2-extension-sidebar', extension_styles) self.assertIn('.v2-extension-capability span {', extension_styles) self.assertIn('.v2-extension-parameter-list > span {', extension_styles) self.assertIn('.v2-extension-vehicle-card > footer {', extension_styles) self.assertIn('.v2-extension-twin-host > .twin-band {', extension_styles) self.assertIn('.v2-extension-type-card[data-extension-connected="true"]', extension_styles) self.assertIn('.v2-extension-type-card[data-extension-connected="false"]', extension_styles) self.assertIn( '.v2-extension-control\n.v2-extension-learned-action {\n --extension-card-color: var(--v2-accent);', extension_styles, ) self.assertIn('grid-template-columns: repeat(3, minmax(0, 1fr))', extension_styles) self.assertTrue(extension_health["success"]) self.assertIn(extension_health["transport"], {"tcp", "mqtt", "simulation"}) self.assertIn("function emphasizeReferenceFrontLampMeshes", app_script) self.assertIn("function createReferenceFrontLampHighlights", app_script) self.assertIn("referenceFrontLampMaterials", app_script) self.assertIn("targets.windowPositions", app_script) self.assertIn( 'const REFERENCE_VEHICLE_ASSET_VERSION = "77a0fae7"', app_script ) self.assertIn( "const REFERENCE_VEHICLE_LOAD_TIMEOUT_MS = 30000", app_script ) self.assertIn( 'referenceVehicleAssetUrl("byd-tang-2024.fbx")', app_script ) self.assertIn("const stageHasLayout = () =>", app_script) self.assertIn("window.requestAnimationFrame(initSeatTwin)", app_script) self.assertIn('client_type: "web_console"', app_script) self.assertIn("function executionVerificationStatus(result = {})", app_script) self.assertIn('"submitted_unverified"', app_script) self.assertIn("指令已提交，等待状态确认", app_script) self.assertIn("指令已提交 · 等待状态确认", console_script) self.assertIn("CAN 指令已提交，等待车辆状态确认", console_script) self.assertIn('canvas.addEventListener("webglcontextlost"', app_script) self.assertIn('"cognitive-agent:module-change"', app_script) self.assertIn('const CACHE_VERSION = "cognitive-agent-pad-v210"', service_worker) self.assertIn('"/app.js?v=dual-seat-107"', service_worker) self.assertIn('"/assets/pets/self-extension-cat.png?v=3"', service_worker) self.assertIn(".then(() => self.skipWaiting())", service_worker) self.assertNotIn("const cached = await caches.match(request)", service_worker) self.assertNotIn("self-evolution", service_worker) self.assertNotIn("self-calibration", service_worker) self.assertNotIn("self-repair", service_worker) self.assertNotIn("n7_seat.glb", service_worker) self.assertIn("hardware_access_required", app_script) self.assertIn('button.setAttribute("aria-label", "关闭摄像头")', gesture_script) self.assertIn("function refreshCameraAvailability", gesture_script) self.assertIn("navigator.mediaDevices.enumerateDevices()", gesture_script) self.assertIn('addEventListener?.("devicechange", handleCameraDeviceChange)', gesture_script) self.assertIn('case "NotFoundError":', gesture_script) self.assertIn('case "NotAllowedError":', gesture_script) self.assertIn('case "NotReadableError":', gesture_script) self.assertIn('case "OverconstrainedError":', gesture_script) self.assertNotIn("sendSimulatedGesture", gesture_script) self.assertNotIn("gestureDebug", gesture_script) self.assertIn("HARDWARE_REQUEST_TIMEOUT_MS = 15 * 60 * 1000", app_script) self.assertIn( "requestTimeoutForExecutionMode(state.executionMode)", app_script ) self.assertIn('error.code === "hardware_busy"', app_script) self.assertIn("BENCH BUSY", app_script) self.assertIn("FBXLoader.js", html) self.assertIn('data-seat-layout="five-seat"', html) self.assertIn('data-active-seat="front-right"', html) self.assertNotIn('placeholder="请输入服务端配置令牌"', html) self.assertIn('data-callout="backrest_recline"', html) self.assertIn('data-callout="legrest_extension"', html) self.assertIn('data-callout="backrest_included_angle"', html) self.assertIn('data-callout="legrest_included_angle"', html) self.assertIn('data-twin-function="ventilation"', html) self.assertIn('data-twin-function="heating"', html) self.assertIn('class="twin-level-meter"', html) self.assertIn('id="presetIndicator"', html) self.assertIn('id="vehicleSelect"', html) self.assertIn('id="saveModeToggle"', html) self.assertIn("根据指令创建", html) self.assertIn('aria-label="快捷功能模式"', html) self.assertIn('id="modeRailPrev"', html) self.assertIn('id="modeRailNext"', html) self.assertIn('id="modeGalleryToggle"', html) self.assertIn('id="modeGallery"', html) self.assertIn('id="modeGalleryGroups"', html) self.assertIn("全部功能模式", html) self.assertIn('id="modeEditorBackdrop"', html) self.assertIn('id="modeEditorName"', html) self.assertIn('id="modeEditorInstruction"', html) self.assertIn('id="modeEditorExecute"', html) self.assertIn('id="modeEditorComplete"', html) self.assertIn('id="modeEditorFlowPreview"', html) self.assertIn('class="mode-editor-flow-preview empty"', html) self.assertIn('id="saveModeSource"', html) self.assertNotIn("主驾 24℃", html) populate_modes = app_script[ app_script.index("function populateModeChips()") : app_script.index("function setModeExecutionState") ] self.assertNotIn("ch1_body_can_driver_only", populate_modes) self.assertIn("function populateModeGallery", app_script) self.assertIn("function updateModeRailNavigation", app_script) self.assertIn("function scrollModeRail", app_script) self.assertIn('className = "mode-gallery-card"', app_script) self.assertIn("#modeGallery button[data-mode-profile-id]", app_script) self.assertIn("function setModeExecutionState", app_script) self.assertIn("function requestCommandAndWait", app_script) self.assertIn("已创建并添加到功能模式栏", app_script) self.assertLess( app_script.index('id: "personal"'), app_script.index('id: "frequent"'), ) self.assertIn('button.dataset.modeOrigin === "user"', app_script) self.assertIn('userDivider.dataset.modeDivider = "user"', app_script) self.assertIn('deleteUserModeProfile(deleteButton.dataset.modeDelete)', app_script) self.assertIn('method: "DELETE"', app_script) self.assertIn('deleteButton.dataset.modeDelete = mode.id', app_script) self.assertIn('editButton.dataset.modeEdit = mode.id', app_script) self.assertIn("function openModeEditor", app_script) self.assertIn("function executeModeEditorInstruction", app_script) self.assertIn("function completeModeEditor", app_script) self.assertIn("function modeEditorReadableCommands", app_script) self.assertIn("function modeEditorMessagesForOperation", app_script) self.assertIn("function modeEditorAppendCommandCard", app_script) self.assertIn('title.textContent = flow?.control_transport === "wireless_mqtt" ? "MQTT 消息序列"', app_script) self.assertIn("message.payload", app_script) self.assertIn("source_request_id: editor.sourceRequestId", app_script) self.assertIn('method: "POST"', app_script) self.assertIn('data-domain-panel="climate"', html) self.assertIn('data-domain-panel="roof"', html) self.assertIn('data-domain-panel="ambient_light"', html) self.assertIn('data-domain-panel="window"', html) self.assertIn('data-domain-panel="front_light"', html) self.assertIn('data-domain-panel="vehicle_status"', html) self.assertIn('id="vehicleGear"', html) self.assertIn('id="vehicleSpeed"', html) self.assertNotIn('data-domain-panel="control_link"', html) self.assertIn('id="cabinStateTitle">整车设备状态', html) self.assertIn('id="cabinCarouselPrev"', app_script) self.assertIn('id="cabinCarouselNext"', app_script) self.assertIn('aria-label="左右浏览设备卡片"', app_script) self.assertNotIn('class="cabin-category-tabs"', app_script) self.assertIn('LIVE VEHICLE DIGITAL TWIN', html) self.assertIn('.joint-twin-shell > .twin-band {', console_styles) self.assertIn('grid-template-rows: 62px minmax(0, 1fr)', console_styles) self.assertIn('grid-template-columns: repeat(2, minmax(0, 1fr))', console_styles) self.assertIn('--cabin-cards-per-view: 2', console_styles) self.assertIn('const cabinCardsPerView = () => 2', app_script) self.assertIn('scroll-snap-type: x mandatory', console_styles) self.assertIn('data-device-power-value', app_script) self.assertIn('查看${name}功率详情', app_script) self.assertIn('name: "车内灯光", measurable: false', app_script) self.assertIn('.device-power-entry.unavailable', console_styles) self.assertIn('powerDrawer.id = "devicePowerDrawer"', app_script) self.assertIn('window.updateDevicePowerTelemetry = updatePowerTelemetry', app_script) self.assertIn('["longitudinal_motor", "水平电机"]', app_script) self.assertIn('["cushion_motor", "坐盆电机"]', app_script) self.assertIn('["backrest_motor", "座椅靠背电机"]', app_script) self.assertIn('["legrest_motor", "腿托电机"]', app_script) self.assertIn('["ventilation_control", "通风控制"]', app_script) self.assertIn('["heating_control", "加热控制"]', app_script) self.assertIn('["compressor", "压缩机"]', app_script) self.assertIn('["electronic_fan", "电子风扇"]', app_script) self.assertIn('["blower", "鼓风机"]', app_script) self.assertIn('["thermal_management_module", "空调热管理集成模块"]', app_script) self.assertIn('["compressor_controller", "压缩机电控"]', app_script) self.assertIn('["electronic_cooling_pump", "电控冷却水泵"]', app_script) self.assertIn('["front_left_window", "左前车窗"]', app_script) self.assertIn('["front_right_window", "右前车窗"]', app_script) self.assertIn('["rear_left_window", "左后车窗"]', app_script) self.assertIn('["rear_right_window", "右后车窗"]', app_script) self.assertIn('["front_left_combination_lamp", "左前组合灯"]', app_script) self.assertIn('["front_right_combination_lamp", "右前组合灯"]', app_script) self.assertIn('name: "主驾座椅", measurable: true, components: []', app_script) self.assertIn('name: "遮阳帘", measurable: true, components: []', app_script) self.assertIn("仅支持设备总功率", app_script) self.assertIn('.device-power-drawer {', console_styles) self.assertNotIn('class="roof-visual"', html) self.assertNotIn('class="ambient-visual"', html) self.assertIn('id="driverAirDirection"', html) self.assertIn('id="passengerAirDirection"', html) self.assertNotIn('id="transmittanceValue"', html) self.assertNotIn('id="roofMotorState"', html) self.assertIn('id="ambientBrightness"', html) self.assertNotIn('class="example-group"', html) self.assertNotIn('id="safetyStrip"', html) self.assertNotIn('class="pipeline"', html) self.assertNotIn('class="topbar"', html) self.assertNotIn('data-axis="height"', html) self.assertEqual(runtime["domain"], "cabin") self.assertEqual( runtime["hardware_execution_allowed"], ( runtime["hardware"]["configured"] or runtime["vehicle_can"]["configured"] or runtime["vehicle_mqtt"]["configured"] ), ) self.assertEqual(runtime["hardware"]["supported_targets"], ["front_right"]) self.assertEqual( runtime["hardware"]["concurrency_policy"], "single_flight_fail_fast", ) self.assertEqual( runtime["hardware"]["access_policy"], "local_approval_idle_lease", ) self.assertEqual( runtime["hardware"]["access_inactivity_timeout_seconds"], 180 ) self.assertFalse(runtime["hardware"]["busy"]) self.assertEqual( runtime["hardware"]["supported_seat_instances"], ["passenger"] ) self.assertGreaterEqual(runtime["matrix"]["messages"], 25) self.assertGreaterEqual(runtime["matrix"]["signals"], 128) self.assertEqual(runtime["motion_position_range_pct"], [0, 100]) self.assertEqual(runtime["default_vehicle_id"], "denza_n7") self.assertEqual(runtime["vehicle"]["name"], "腾势 N7") self.assertIn( "denza_n7", {vehicle["id"] for vehicle in runtime["vehicles"]}, ) self.assertIn( "zero_gravity", {mode["id"] for mode in runtime["modes"]}, ) self.assertEqual( runtime["vehicle"]["seat"]["motors"]["longitudinal"]["range"], 215, ) self.assertEqual( runtime["vehicle"]["seat"]["motors"]["longitudinal"]["reference_zero"], "forwardmost", ) self.assertEqual( runtime["vehicle"]["seat"]["motors"]["cushion_tilt"]["reference_zero"], "highest", ) capability_names = {item["name"] for item in runtime["capabilities"]} self.assertIn("seat.save_posture_preset", capability_names) self.assertIn("seat.apply_posture_preset", capability_names) self.assertIn("climate.set_temperature", capability_names) self.assertIn("roof.set_transmittance", capability_names) self.assertIn("ambient_light.set_illumination", capability_names) passenger_seat_capabilities = [ item for item in runtime["capabilities"] if item["name"].startswith("seat.") and not item["name"].startswith("seat.set_driver_") and item["name"] != "seat.recall_driver_memory_native" ] driver_seat_capabilities = [ item for item in runtime["capabilities"] if item["name"].startswith("seat.set_driver_") or item["name"] == "seat.recall_driver_memory_native" ] self.assertTrue(passenger_seat_capabilities) self.assertTrue( all( item["target_scopes"] == ["passenger"] for item in passenger_seat_capabilities ) ) self.assertTrue(driver_seat_capabilities) self.assertTrue( all( item["target_scopes"] == ["driver", "front_left"] for item in driver_seat_capabilities ) ) def test_self_extension_vehicle_success_contract_is_strict(self) -> None: with self.opener.open( f"{self.base_url}/self-extension.js?v=extension-44", timeout=3 ) as response: extension_script = response.read().decode("utf-8") with self.opener.open( f"{self.base_url}/app.js?v=dual-seat-107", timeout=3 ) as response: app_script = response.read().decode("utf-8") self.assertIn('code: "SEAT · CH1 BODY CAN"', extension_script) self.assertIn('code: "SEAT · CH0 T2G CAN"', extension_script) self.assertIn('id: "passenger-seat"', extension_script) self.assertIn('parameters: ["通风 0–3 档", "加热 0–2 档"', extension_script) self.assertIn('command: "seat driver vent low"', extension_script) self.assertIn('command: "seat driver vent high"', extension_script) self.assertIn('command: "seat driver vent off"', extension_script) self.assertIn('command: "seat driver heat off"', extension_script) self.assertIn('command: "ac off"', extension_script) self.assertIn('command: "light indoor off"', extension_script) self.assertIn('command: "light reading all off"', extension_script) self.assertIn('command: "light ambient off"', extension_script) self.assertIn('command: `window ${positionName} position ${position}`', extension_script) self.assertNotIn('command: "shade stop"', extension_script) self.assertNotIn('command: "door vehicle lock"', extension_script) self.assertNotIn('command: "door vehicle unlock"', extension_script) self.assertNotIn('command: "door rear unlock"', extension_script) self.assertNotIn('code: "OUTLET · 4 ZONES"', extension_script) self.assertNotIn('command: "打开全部车窗"', extension_script) self.assertNotIn('command: "打开副驾座椅通风"', extension_script) self.assertNotIn('command: "把副驾座椅向后调一点"', extension_script) self.assertIn('result?.status === "succeeded"', extension_script) self.assertIn("everyOperationSucceeded", extension_script) self.assertIn("everyHardwareOperationConfirmed", extension_script) self.assertIn("result?.physical_feedback_confirmed === true", extension_script) self.assertIn("cognitive-agent:extension-vehicle-result", extension_script) self.assertIn("MASSAGE", extension_script) self.assertIn("INTERIOR", extension_script) self.assertIn("AMBIENT", extension_script) self.assertIn( "body: JSON.stringify({ command, conversation_id: sharedConversationId() })", extension_script, ) self.assertIn('window.addEventListener("cognitive-agent:result"', extension_script) self.assertIn('id="extensionVehicleStateSummary"', extension_script) self.assertIn('id="extensionTwinHost"', extension_script) self.assertIn('data-extension-current-action', extension_script) self.assertIn('function syncSharedTwinMount()', extension_script) self.assertIn('function updateVehicleCardExecution(', extension_script) self.assertNotIn("/api/hardware/seat-state", extension_script) self.assertIn( 'window.addEventListener("cognitive-agent:extension-vehicle-result"', app_script, ) self.assertIn("applyAuthoritativeVehicleState", app_script) self.assertIn('payload.conversation_id !== state.conversationId', app_script) self.assertIn( 'target_policy === "ch1_body_can_driver_only"', app_script, ) self.assertIn("async function synchronizePassengerSeatState()", app_script) self.assertIn("/api/hardware/seat-state", app_script) self.assertIn("async function synchronizeDriverSeatState()", app_script) self.assertIn("/api/hardware/vehicle-state", app_script) self.assertNotIn("passenger: mergedActuatorState.driver", app_script) self.assertNotIn("approvalText", app_script) def test_self_extension_vehicle_endpoint_preserves_unsuccessful_feedback( self, ) -> None: service_result = { "status": "partially_succeeded", "provider_used": "local_rules", "request_id": "req-extension-partial", "conversation_id": "conv-extension-shared-e2e", "conversation_turn": 4, "decision_summary": ["空调完成，遮阳帘反馈超时。"], "validation": [], "execution_report": { "spec": { "status": "failed", "physical_state": { "source": "physical_can_ch1", "complete": False, }, "results": [ { "ref": "climate-1", "domain": "climate", "status": "succeeded", "telemetry": { "simulated": False, "feedback_observed": True, "feedback_match": True, "feedback_frame": {"frame_id": "0x404"}, }, }, { "ref": "roof-1", "domain": "roof", "status": "failed", "message": "反馈超时", "telemetry": {"simulated": False}, }, ], } }, "domain_state": {"climate": {"power": "on"}}, "actuator_state": {"driver": {"heating": {"state": "off"}}}, "deferred_requests": [], } request = urllib.request.Request( f"{self.base_url}/api/extension/vehicle/execute", method="POST", headers={"Content-Type": "application/json"}, data=json.dumps( { "command": "打开空调并关闭遮阳帘", "conversation_id": "conv-extension-shared-e2e", } ).encode("utf-8"), ) with patch.object(self.service, "process", return_value=service_result): with self.opener.open(request, timeout=3) as response: result = json.loads(response.read().decode("utf-8")) self.assertFalse(result["success"]) self.assertEqual(result["status"], "partially_succeeded") self.assertEqual(result["conversation_id"], "conv-extension-shared-e2e") self.assertEqual(result["conversation_turn"], 4) self.assertEqual(result["execution_report"], service_result["execution_report"]) self.assertEqual(result["domain_state"], service_result["domain_state"]) self.assertEqual(result["actuator_state"], service_result["actuator_state"]) self.assertEqual(result["physical_state"]["source"], "physical_can_ch1") self.assertEqual(len(result["physical_feedback"]), 2) self.assertFalse(result["physical_feedback_confirmed"]) def test_self_extension_gateway_uses_unified_authenticated_service(self) -> None: original_extension = SeatDemoHandler.extension SeatDemoHandler.extension = SelfExtensionService(SimulationGateway()) try: with self.opener.open( f"{self.base_url}/api/extension/health", timeout=3 ) as response: health = json.loads(response.read().decode("utf-8")) request = urllib.request.Request( f"{self.base_url}/api/extension/gscu/emit", method="POST", headers={"Content-Type": "application/json"}, data=json.dumps({"slot": 9}).encode("utf-8"), ) with self.opener.open(request, timeout=3) as response: result = json.loads(response.read().decode("utf-8")) finally: SeatDemoHandler.extension = original_extension self.assertTrue(health["success"]) self.assertEqual(health["transport"], "simulation") self.assertTrue(health["connected"]) self.assertTrue(result["success"]) self.assertEqual(result["transport"], "simulation") self.assertFalse(result["physical_state_confirmed"]) def test_post_request_returns_complete_dry_run(self) -> None: request = urllib.request.Request( f"{self.base_url}/api/requests", method="POST", headers={"Content-Type": "application/json"}, data=json.dumps( { "utterance": "副驾有点热，有点闷", "provider": "local", "conversation_id": "conv-e2e-browser", "vehicle_id": "denza_n7", } ).encode("utf-8"), ) with self.opener.open(request, timeout=3) as response: result = json.loads(response.read().decode("utf-8")) self.assertEqual(result["status"], "succeeded") self.assertEqual(result["function_config"]["kind"], "FunctionConfig") self.assertEqual(result["command_plan"]["kind"], "CommandPlan") self.assertEqual(result["execution_report"]["spec"]["mode"], "dry_run") self.assertEqual(result["conversation_id"], "conv-e2e-browser") self.assertEqual(result["vehicle"]["id"], "denza_n7") self.assertEqual(result["actuator_state"]["passenger"]["ventilation"]["level"], 1) def test_relative_backrest_phrase_uses_current_conversation_position(self) -> None: conversation_id = "conv-e2e-relative-backrest" for utterance in ("副驾靠背角度调到130度", "副驾靠背向前移动10度"): request = urllib.request.Request( f"{self.base_url}/api/requests", method="POST", headers={"Content-Type": "application/json"}, data=json.dumps( { "utterance": utterance, "provider": "local", "conversation_id": conversation_id, "vehicle_id": "denza_n7", "execution_mode": "dry_run", }, ensure_ascii=False, ).encode("utf-8"), ) with self.opener.open(request, timeout=3) as response: result = json.loads(response.read().decode("utf-8")) self.assertEqual(result["status"], "succeeded") operation = result["function_config"]["spec"]["operations"][0] self.assertEqual(operation["capability"], "seat.set_backrest_recline") self.assertEqual(operation["parameters"]["position_pct"], 36) self.assertEqual(operation["parameters"]["angle_deg"], 120) self.assertEqual(operation["motion"]["direction"], "decrease") self.assertEqual(operation["motion"]["delta_angle_deg"], 10.0) def test_timed_and_oversized_backrest_requests_use_endpoint_fallbacks(self) -> None: conversation_id = "conv-e2e-timed-endpoint-fallbacks" def submit(utterance: str) -> dict: request = urllib.request.Request( f"{self.base_url}/api/requests", method="POST", headers={"Content-Type": "application/json"}, data=json.dumps( { "utterance": utterance, "provider": "local", "conversation_id": conversation_id, "vehicle_id": "denza_n7", "execution_mode": "dry_run", }, ensure_ascii=False, ).encode("utf-8"), ) with self.opener.open(request, timeout=3) as response: return json.loads(response.read().decode("utf-8")) submit("副驾靠背归零") timed = submit("副驾靠背向后移动70秒") self.assertEqual(timed["status"], "succeeded") timed_operation = timed["function_config"]["spec"]["operations"][0] self.assertEqual(timed_operation["parameters"]["position_pct"], 100) self.assertEqual(timed_operation["motion"]["mode"], "timed") self.assertEqual(timed_operation["motion"]["requested_duration_ms"], 70_000) self.assertEqual(timed_operation["motion"]["applied_duration_ms"], 30_000) self.assertTrue(timed_operation["motion"]["duration_clamped"]) self.assertIn("最大允许时长为30秒", "\n".join(timed["decision_summary"])) submit("副驾靠背角度调到130度") overflow = submit("副驾靠背向后移动70度") self.assertEqual(overflow["status"], "succeeded") overflow_operation = overflow["function_config"]["spec"]["operations"][0] self.assertEqual(overflow_operation["parameters"]["position_pct"], 100) self.assertEqual(overflow_operation["parameters"]["angle_deg"], 163) self.assertTrue(overflow_operation["motion"]["range_clamped"]) self.assertEqual( overflow_operation["motion"]["requested_delta_angle_deg"], 70 ) self.assertIn("备用目标", "\n".join(overflow["decision_summary"])) def test_open_zero_gravity_phrase_matches_complete_mode(self) -> None: request = urllib.request.Request( f"{self.base_url}/api/requests", method="POST", headers={"Content-Type": "application/json"}, data=json.dumps( { "utterance": "打开零重力模式", "provider": "local", "conversation_id": "conv-e2e-open-zero-gravity", "vehicle_id": "denza_n7", "execution_mode": "dry_run", }, ensure_ascii=False, ).encode("utf-8"), ) with self.opener.open(request, timeout=3) as response: result = json.loads(response.read().decode("utf-8")) self.assertEqual(result["status"], "succeeded") self.assertIn("zero_gravity", {item["id"] for item in result["matched_modes"]}) operations = result["function_config"]["spec"]["operations"] self.assertEqual(len(operations), 4) self.assertEqual( { operation["capability"]: operation["parameters"]["position_pct"] for operation in operations }, { "seat.set_longitudinal_position": 100, "seat.set_cushion_tilt": 100, "seat.set_backrest_recline": 81, "seat.set_legrest_height": 94, }, ) self.assertEqual(result["execution_report"]["spec"]["mode"], "dry_run") def test_hardware_state_endpoint_returns_physical_passenger_snapshot(self) -> None: synchronized = { "status": "succeeded", "conversation_id": "conv-e2e-physical-state", "source": "physical_can", "complete": True, "missing": [], "actuator_state": { "passenger": { "position": { "longitudinal": {"position_pct": 0}, "cushion_tilt": {"position_pct": 0}, "backrest_recline": {"position_pct": 0}, "legrest_height": {"position_pct": 0}, } } }, } with patch.object( self.service, "synchronize_hardware_state", return_value=synchronized, ) as synchronize: with self.opener.open( f"{self.base_url}/api/hardware/seat-state?" "conversation_id=conv-e2e-physical-state&vehicle_id=denza_n7", timeout=3, ) as response: payload = json.loads(response.read().decode("utf-8")) self.assertEqual(payload["source"], "physical_can") self.assertEqual( payload["actuator_state"]["passenger"]["position"]["longitudinal"]["position_pct"], 0, ) synchronize.assert_called_once_with( "conv-e2e-physical-state", "denza_n7", control_transport="wired_can", hardware_client_id=None, hardware_access_bypass=True, ) def test_wireless_passenger_snapshot_uses_mqtt_can0(self) -> None: physical_state = { "position": { "longitudinal": { "position_pct": 42, "position_valid": True, "motion_state": "stopped", "fault": False, } }, "captured_at": "2026-08-18T00:00:00Z", "complete": True, "missing": [], } executor = SimpleNamespace(read_physical_state=lambda: dict(physical_state)) with ( patch.object( self.service, "vehicle_mqtt_settings", SimpleNamespace(state_configured=True), ), patch.object( self.service._vehicle_mqtt_transport, "bus", return_value=object(), ) as mqtt_bus, patch.object( self.service, "_new_hardware_executor", return_value=executor, ) as new_executor, patch.object( self.service.hardware_execution, "reserve", side_effect=AssertionError( "wireless cached state must not reserve the physical bench" ), ), ): payload = self.service.synchronize_hardware_state( "conv-e2e-wireless-seat-state", "denza_n7", control_transport="wireless_mqtt", hardware_access_bypass=True, ) bus_factory = new_executor.call_args.args[0] bus_factory() mqtt_bus.assert_called_once_with(channel=0, replay_latest=True) self.assertTrue(new_executor.call_args.kwargs["remote_transport"]) self.assertEqual(payload["source"], "wireless_mqtt_ch0") self.assertEqual(payload["control_transport"], "wireless_mqtt") self.assertEqual( payload["actuator_state"]["passenger"]["position"]["longitudinal"]["position_pct"], 42, ) def test_wireless_vehicle_snapshot_does_not_reserve_physical_bench(self) -> None: snapshot = SimpleNamespace( feedback_states=(), observed_ids=frozenset(), captured_at="2026-08-18T00:00:00Z", capture_sequence=17, ) with ( patch.object( self.service, "vehicle_mqtt_settings", SimpleNamespace(state_configured=True), ), patch.object( self.service._vehicle_mqtt_transport, "snapshot", return_value=snapshot, ), patch.object( self.service.hardware_execution, "reserve", side_effect=AssertionError( "wireless cached state must not reserve the physical bench" ), ), ): payload = self.service.synchronize_vehicle_state( "conv-e2e-wireless-vehicle-state", "denza_n7", control_transport="wireless_mqtt", hardware_access_bypass=True, ) self.assertEqual(payload["source"], "wireless_mqtt_ch1") self.assertEqual(payload["capture_sequence"], 17) def test_gpt_http_failure_falls_back_without_returning_502(self) -> None: original_api_key = self.service.openai.api_key self.service.openai.api_key = "test-key" request = urllib.request.Request( f"{self.base_url}/api/requests", method="POST", headers={"Content-Type": "application/json"}, data=json.dumps( { "utterance": "打开副驾座椅通风，同时播放音乐", "provider": "openai", "conversation_id": "conv-e2e-gpt-fallback", "execution_mode": "dry_run", } ).encode("utf-8"), ) try: with patch.object( self.service.openai, "interpret", side_effect=OpenAIIntentError("invalid_api_key"), ): with self.opener.open(request, timeout=3) as response: result = json.loads(response.read().decode("utf-8")) self.assertEqual(response.status, 201) finally: self.service.openai.api_key = original_api_key self.assertEqual(result["status"], "succeeded") self.assertEqual(result["provider_used"], "local_rules_fallback") self.assertEqual(result["execution_report"]["spec"]["mode"], "dry_run") def test_two_stage_http_api_returns_local_commands_before_cloud_supplement( self, ) -> None: original_api_key = self.service.openai.api_key self.service.openai.api_key = "two-stage-http-test-key" preview_request = urllib.request.Request( f"{self.base_url}/api/requests/preview", method="POST", headers={"Content-Type": "application/json"}, data=json.dumps( { "utterance": "空调温度设置为24度，播放音乐", "provider": "openai", "conversation_id": "conv-e2e-two-stage-http", "execution_mode": "dry_run", "reasoning_effort": "medium", "client_type": "web_console", } ).encode("utf-8"), ) model_output = { "detected_domains": ["ambient_light"], "requests": [ { "domain": "ambient_light", "capability": "ambient_light.set_power", "target_scope": "vehicle", "operation": "set", "duration_ms": None, "parameters": [{"name": "state", "value": "on"}], "reason": "补充播放场景氛围灯。", } ], "ambiguities": [], "decision_summary": ["仅补充本地未解析片段。"], "confidence": 0.9, } try: with patch.object( self.service.openai, "interpret", side_effect=AssertionError("preview must not invoke GPT"), ): with self.opener.open(preview_request, timeout=3) as response: preview = json.loads(response.read().decode("utf-8")) self.assertEqual(response.status, 201) local_operation = next( operation for config in preview["function_configs"].values() for operation in config["spec"]["operations"] ) self.assertEqual(local_operation["parse_source"], "local_rule") self.assertTrue(preview["supplement_required"]) supplement_request = urllib.request.Request( f"{self.base_url}/api/requests/{preview['request_id']}/supplement", method="POST", headers={"Content-Type": "application/json"}, data=json.dumps( { "provider": "openai", "reasoning_effort": "medium", "client_type": "web_console", } ).encode("utf-8"), ) with patch.object( self.service.openai, "interpret", return_value=model_output, ) as interpret: with self.opener.open(supplement_request, timeout=3) as response: result = json.loads(response.read().decode("utf-8")) self.assertEqual(response.status, 201) self.assertEqual(interpret.call_args.args[0], "播放音乐") finally: self.service.openai.api_key = original_api_key self.assertEqual(result["phase"], "two_stage_complete") sources = { operation["capability"]: operation["parse_source"] for config in result["function_configs"].values() for operation in config["spec"]["operations"] } self.assertEqual(sources["climate.set_temperature"], "local_rule") self.assertEqual(sources["ambient_light.set_power"], "cloud_supplement") def test_preview_streams_validated_commands_before_terminal_result(self) -> None: request = urllib.request.Request( f"{self.base_url}/api/requests/preview", method="POST", headers={ "Content-Type": "application/json", "Accept": "application/x-ndjson", }, data=json.dumps( { "utterance": "空调24度、风量2档", "provider": "local", "conversation_id": "conv-e2e-command-stream", "execution_mode": "dry_run", "client_type": "web_console", }, ensure_ascii=False, ).encode("utf-8"), ) with self.opener.open(request, timeout=3) as response: events = [ json.loads(line) for line in response.read().decode("utf-8").splitlines() if line.strip() ] self.assertEqual( response.headers.get_content_type(), "application/x-ndjson", ) self.assertEqual( [event["type"] for event in events], ["validated_commands", "result"], ) validated = events[0]["payload"] self.assertEqual(validated["status"], "processing") self.assertIsNone(validated["execution_report"]) self.assertTrue(validated["command_plans"]) self.assertEqual(events[1]["payload"]["status"], "succeeded") def test_stream_failure_uses_cli_outcome_and_updates_display_sync_terminal_state( self, ) -> None: request = urllib.request.Request( f"{self.base_url}/api/requests", method="POST", headers={ "Content-Type": "application/json", "Accept": "application/x-ndjson", }, data=json.dumps( { "utterance": "打开副驾座椅通风", "provider": "local", "conversation_id": "conv-e2e-stream-can-failure", "execution_mode": "hardware", "client_type": "web_console", }, ensure_ascii=False, ).encode("utf-8"), ) with patch.object( self.service, "process", side_effect=CanGatewayError("VCI_OpenDevice failed"), ): with self.opener.open(request, timeout=3) as response: events = [ json.loads(line) for line in response.read().decode("utf-8").splitlines() if line.strip() ] with self.opener.open( f"{self.base_url}/api/vehicle-display-sync", timeout=3 ) as response: snapshot = json.loads(response.read().decode("utf-8")) self.assertEqual(events, [{ "type": "error", "error": "can_transport_failed", "message": "VCI_OpenDevice failed", "outcome": "failed_or_blocked", }]) self.assertEqual(snapshot["task"]["status"], "failed") self.assertEqual(snapshot["task"]["utterance"], "打开副驾座椅通风") self.assertEqual( snapshot["task"]["error"], { "code": "can_transport_failed", "message": "VCI_OpenDevice failed", "outcome": "failed_or_blocked", }, ) def test_frontend_suite_covers_every_physical_seat_capability(self) -> None: suite_path = ( REPOSITORY_ROOT / "tests" / "fixtures" / "seat_frontend_capabilities.json" ) suite = json.loads(suite_path.read_text(encoding="utf-8")) required = set(suite["required_hardware_capabilities"]) covered: set[str] = set() original_api_key = self.service.openai.api_key self.service.openai.api_key = "frontend-suite-test-key" try: for case in suite["cases"]: with self.subTest(case=case["id"]): requests = [] for expected in case["operations"]: covered.add(expected["capability"]) requests.append( { "domain": "seat", "capability": expected["capability"], "target_scope": expected["target"], "operation": "set", "parameters": [ {"name": name, "value": value} for name, value in expected["parameters"].items() ], "reason": f"frontend test case {case['id']}", } ) model_plan = { "detected_domains": ["seat"], "requests": requests, "ambiguities": [], "decision_summary": [f"执行前端测试用例 {case['id']}。"], "confidence": 0.99, } request = urllib.request.Request( f"{self.base_url}/api/requests", method="POST", headers={"Content-Type": "application/json"}, data=json.dumps( { "utterance": case["utterance"], "provider": "openai", "conversation_id": f"conv-frontend-suite-{case['id'].lower()}", "execution_mode": "dry_run", "vehicle_id": "denza_n7", }, ensure_ascii=False, ).encode("utf-8"), ) with patch.object( self.service.openai, "interpret", return_value=model_plan, ): try: with self.opener.open(request, timeout=3) as response: result = json.loads(response.read().decode("utf-8")) except urllib.error.HTTPError as exc: detail = exc.read().decode("utf-8", errors="replace") self.fail(f"frontend case {case['id']} returned {exc.code}: {detail}") self.assertEqual(result["status"], "succeeded") operations = result["function_config"]["spec"]["operations"] self.assertEqual(len(operations), len(case["operations"])) self.assertEqual( result["execution_report"]["spec"]["mode"], "dry_run" ) self.assertEqual( len(result["execution_report"]["spec"]["results"]), len(case["operations"]), ) for expected, actual in zip(case["operations"], operations): self.assertEqual(actual["capability"], expected["capability"]) self.assertEqual( actual["target"]["instance"], expected["target"] ) for name, value in expected["parameters"].items(): self.assertEqual(actual["parameters"][name], value) finally: self.service.openai.api_key = original_api_key self.assertEqual(covered, required) safety_ids = {item["id"] for item in suite["safety_cases"]} self.assertEqual( safety_ids, {"SAFE-01", "SAFE-02", "SAFE-03", "RECOVERY-01"}, ) def test_hardware_request_is_rejected_when_server_gate_is_locked(self) -> None: request = urllib.request.Request( f"{self.base_url}/api/requests", method="POST", headers={"Content-Type": "application/json"}, data=json.dumps( { "utterance": "副驾座椅加热开到一档", "provider": "local", "conversation_id": "conv-e2e-hardware-locked", "execution_mode": "hardware", "approval_token": "must-not-be-persisted", } ).encode("utf-8"), ) with self.assertRaises(urllib.error.HTTPError) as caught: self.opener.open(request, timeout=3) self.assertEqual(caught.exception.code, 403) payload = json.loads(caught.exception.read().decode("utf-8")) self.assertEqual(payload["error"], "hardware_safety_gate") artifact_text = "\n".join( path.read_text(encoding="utf-8") for path in Path(self.temporary_directory.name).rglob("*") if path.is_file() ) self.assertNotIn("must-not-be-persisted", artifact_text) def test_local_decision_point_selects_one_public_terminal_and_can_stop_it(self) -> None: client_a = "hwc-e2e-client-a1" client_b = "hwc-e2e-client-b2" def post(path: str, payload: dict, *, public: bool = False): headers = {"Content-Type": "application/json"} if public: headers["X-Forwarded-For"] = "203.0.113.25" request = urllib.request.Request( f"{self.base_url}{path}", method="POST", headers=headers, data=json.dumps(payload, ensure_ascii=False).encode("utf-8"), ) return self.opener.open(request, timeout=3) with post( "/api/hardware/access/request", {"client_id": client_a, "label": "用户电脑 A"}, public=True, ) as response: first = json.loads(response.read().decode("utf-8")) self.assertEqual(response.status, 202) with post( "/api/hardware/access/request", {"client_id": client_b, "label": "用户电脑 B"}, public=True, ) as response: second = json.loads(response.read().decode("utf-8")) self.assertEqual(first["queue_position"], 1) self.assertEqual(second["queue_position"], 2) with self.opener.open( f"{self.base_url}/api/local/hardware-access", timeout=3 ) as response: local_snapshot = json.loads(response.read().decode("utf-8")) self.assertEqual(len(local_snapshot["pending"]), 2) with post( "/api/local/hardware-access/approve", {"client_id": client_b} ) as response: approved = json.loads(response.read().decode("utf-8")) self.assertEqual(approved["active"]["client_id"], client_b) with self.assertRaises(urllib.error.HTTPError) as remote_admin: request = urllib.request.Request( f"{self.base_url}/api/local/hardware-access", headers={"X-Forwarded-For": "203.0.113.25"}, ) self.opener.open(request, timeout=3) self.assertEqual(remote_admin.exception.code, 403) def hardware_request(client_id: str, conversation: str): return post( "/api/requests", { "utterance": "关闭副驾座椅加热", "provider": "local", "conversation_id": conversation, "execution_mode": "hardware", "hardware_client_id": client_id, }, public=True, ) fake_result = {"status": "succeeded", "request_id": "req-approved-client"} with patch.object( self.service, "_process_once", return_value=fake_result ) as process_once: with hardware_request(client_b, "conv-approved-client") as response: self.assertEqual(response.status, 201) with self.assertRaises(urllib.error.HTTPError) as unapproved: hardware_request(client_a, "conv-unapproved-client") denied_payload = json.loads(unapproved.exception.read().decode("utf-8")) self.assertEqual(unapproved.exception.code, 403) self.assertEqual(denied_payload["error"], "hardware_access_required") self.assertEqual(denied_payload["access_state"], "pending") self.assertEqual(process_once.call_count, 1) with post( "/api/hardware/access/release", {"client_id": client_b}, public=True, ) as response: released = json.loads(response.read().decode("utf-8")) self.assertEqual(released["state"], "none") self.assertIsNone(self.service.hardware_access.admin_snapshot()["active"]) def test_local_hardware_decision_page_is_not_exposed_through_proxy(self) -> None: with self.opener.open( f"{self.base_url}/hardware-control", timeout=3 ) as response: html = response.read().decode("utf-8") self.assertIn("真实台架本地决策台", html) proxied = urllib.request.Request( f"{self.base_url}/hardware-control", headers={"X-Forwarded-For": "203.0.113.25"}, ) with self.assertRaises(urllib.error.HTTPError) as caught: self.opener.open(proxied, timeout=3) self.assertEqual(caught.exception.code, 403) def test_overlapping_hardware_request_returns_conflict_without_entering_can(self) -> None: entered = threading.Event() release = threading.Event() first_result: dict[str, object] = {} cookie_header = "; ".join( f"{cookie.name}={cookie.value}" for cookie in self.cookie_jar ) def request(conversation_id: str) -> urllib.request.Request: return urllib.request.Request( f"{self.base_url}/api/requests", method="POST", headers={ "Content-Type": "application/json", "Cookie": cookie_header, }, data=json.dumps( { "utterance": "关闭副驾座椅加热", "provider": "local", "conversation_id": conversation_id, "execution_mode": "hardware", }, ensure_ascii=False, ).encode("utf-8"), ) def occupy_bench(*_args: object, **_kwargs: object) -> dict[str, object]: entered.set() self.assertTrue(release.wait(timeout=3)) return {"status": "succeeded", "request_id": "req-exclusive-owner"} def submit_owner() -> None: with urllib.request.urlopen( request("conv-exclusive-owner"), timeout=5 ) as response: first_result["http_status"] = response.status first_result["payload"] = json.loads( response.read().decode("utf-8") ) with patch.object( self.service, "_process_once", side_effect=occupy_bench, ) as process_once: owner = threading.Thread(target=submit_owner) owner.start() self.assertTrue(entered.wait(timeout=2)) with self.assertRaises(urllib.error.HTTPError) as caught: urllib.request.urlopen( request("conv-exclusive-contender"), timeout=3 ) payload = json.loads(caught.exception.read().decode("utf-8")) self.assertEqual(caught.exception.code, 409) self.assertEqual(caught.exception.headers["Retry-After"], "5") self.assertEqual(payload["error"], "hardware_busy") self.assertIn("未进入CAN链路", payload["message"]) self.assertEqual(payload["retry_after_seconds"], 5) self.assertEqual(process_once.call_count, 1) state_request = urllib.request.Request( f"{self.base_url}/api/hardware/seat-state?" "conversation_id=conv-exclusive-state-reader&vehicle_id=denza_n7", headers={"Cookie": cookie_header}, ) with self.assertRaises(urllib.error.HTTPError) as state_caught: urllib.request.urlopen(state_request, timeout=3) state_payload = json.loads( state_caught.exception.read().decode("utf-8") ) self.assertEqual(state_caught.exception.code, 409) self.assertEqual(state_payload["error"], "hardware_busy") release.set() owner.join(timeout=5) self.assertFalse(owner.is_alive()) self.assertEqual(first_result["http_status"], 201) self.assertFalse(self.service.hardware_execution.public_info()["busy"]) if __name__ == "__main__": unittest.main()