vee
523 lines
23 KiB
Python
Executable File
523 lines
23 KiB
Python
Executable File
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()
|