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 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.infrastructure.llm.openai_responses import OpenAIIntentError from cognitive_agent.infrastructure.can.access import HardwareAccessController 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"}, 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_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=seat-hil-9", timeout=3) as response: app_script = 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("Cognitive Agent", html) self.assertIn('id="seatTwinCanvas"', html) self.assertIn('data-vehicle-model="byd-tang-2024-reference"', html) self.assertIn("整车孪生界面", html) self.assertIn("透明车体 · 实体设备", html) self.assertIn('src="/app.js?v=seat-hil-9"', html) self.assertIn('id="hardwareAccessField"', html) self.assertIn("hardware_access_required", app_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.assertIn('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('data-domain-panel="climate"', html) self.assertIn('data-domain-panel="roof"', html) self.assertIn('data-domain-panel="ambient_light"', html) self.assertIn("坐垫脚托夹角133度", html) self.assertNotIn('data-axis="height"', html) self.assertEqual(runtime["domain"], "cabin") self.assertEqual( runtime["hardware_execution_allowed"], runtime["hardware"]["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) seat_capabilities = [ item for item in runtime["capabilities"] if item["name"].startswith("seat.") ] self.assertTrue(seat_capabilities) self.assertTrue( all(item["target_scopes"] == ["passenger"] for item in seat_capabilities) ) 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", hardware_client_id=None, hardware_access_bypass=True, ) 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_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()