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.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=20, 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.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 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-7", 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-7"', html) self.assertIn("HARDWARE_REQUEST_TIMEOUT_MS = 15 * 60 * 1000", app_script) self.assertIn( "requestTimeoutForExecutionMode(state.executionMode)", 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"]["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", ) 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) if __name__ == "__main__": unittest.main()