vee/Cognitive_Agent
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: validate real-bench seat orchestration and self-repair

Browse Source
This commit is contained in:
vee
2026-08-09 19:00:16 +08:00
parent 3bceb0d5a1
commit 3701192a11
41 changed files with 10403 additions and 3392 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/e2e/test_seat_web_api.py
+8 -2
View File
@@ -115,7 +115,7 @@ class SeatWebApiTests(unittest.TestCase):
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-43", timeout=3) as response:
with self.opener.open(f"{self.base_url}/app.js?v=seat-hil-44", 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-6", timeout=3
@@ -150,7 +150,7 @@ class SeatWebApiTests(unittest.TestCase):
self.assertIn("整车孪生界面", html)
self.assertIn("<strong>阅读灯</strong>", html)
self.assertNotIn("透明车体 · 实体设备", html)
self.assertIn('src="/app.js?v=seat-hil-43"', html)
self.assertIn('src="/app.js?v=seat-hil-44"', html)
self.assertIn('src="/gesture-control.js?v=android-pad-6"', html)
self.assertNotIn('class="vehicle-system-hud"', html)
self.assertIn('src="/console-v2.js?v=desktop-refactor-28"', html)
@@ -229,6 +229,12 @@ class SeatWebApiTests(unittest.TestCase):
self.assertNotIn('id="releaseHardwareAccess"', html)
self.assertIn('const dynamicModeOrder = ["rocking_horse", "dynamic_cradle"]', app_script)
self.assertIn('divider.className = "mode-group-divider"', app_script)
self.assertIn(
'mode === "hardware" && !state.runtime?.hardware_execution_allowed',
app_script,
)
self.assertIn("hardwareButton.disabled = !hardwareExecutionReady", 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)
tests/e2e/test_self_repair_api.py
+27 -6
View File
@@ -5,6 +5,7 @@ import threading
import urllib.error
import urllib.request
from http.server import ThreadingHTTPServer
from pathlib import Path
import pytest
@@ -134,15 +135,15 @@ class TestSelfRepairApi:
with urllib.request.urlopen(f"{self.base_url}/", timeout=3) as response:
html = response.read().decode("utf-8")
with urllib.request.urlopen(
f"{self.base_url}/self-repair.js?v=repair-15", timeout=3
f"{self.base_url}/self-repair.js?v=repair-17", timeout=3
) as response:
script = response.read().decode("utf-8")
with urllib.request.urlopen(
f"{self.base_url}/self-repair.css?v=repair-15", timeout=3
f"{self.base_url}/self-repair.css?v=repair-17", timeout=3
) as response:
styles = response.read().decode("utf-8")
with urllib.request.urlopen(
f"{self.base_url}/self-repair-seat3d.js?v=repair-15", timeout=3
f"{self.base_url}/self-repair-seat3d.js?v=repair-17", timeout=3
) as response:
seat3d = response.read().decode("utf-8")
with urllib.request.urlopen(
@@ -150,10 +151,15 @@ class TestSelfRepairApi:
) as response:
console_script = response.read().decode("utf-8")
assert 'href="/self-repair.css?v=repair-15"' in html
assert 'src="/self-repair.js?v=repair-15"' in html
assert 'src="/self-repair-seat3d.js?v=repair-15"' in html
assert 'href="/self-repair.css?v=repair-17"' in html
assert 'src="/self-repair.js?v=repair-17"' in html
assert 'src="/self-repair-seat3d.js?v=repair-17"' in html
assert "量产故障用例" in script
assert 'id="repairConfirmDialog"' in script
assert "confirmAction(" in script
assert "window.confirm" not in script
assert 'hardware.link?.adapter === "OK"' in script
assert 'PASSED: ["21针通过", "online"]' in script
assert "connectorRows" in script
assert "Polling refreshes data only" in script
assert "renderRecords();\n showView(activeView);" not in script
@@ -172,3 +178,18 @@ class TestSelfRepairApi:
assert 'window.addEventListener("self-repair-seat-reset"' in seat3d
assert '["self-calibration", "self-repair"].includes(module)' in console_script
assert 'roleSelect.disabled = ui.activeModule === "self-repair"' in console_script
def test_self_repair_bridge_uses_the_backend_runtime_directory() -> None:
source = (
Path(__file__).resolve().parents[2]
/ "tools"
/ "zcan"
/ "seat_self_repair_bridge.py"
).read_text(encoding="utf-8")
assert '"artifacts", "tmp", "self-repair", "runtime"' in source
assert 'os.path.join(BASE_DIR, "seat_function_runtime")' not in source
assert "A silent or disconnected CAN bus must not kill" in source
assert 'self.message("GUI STOP_ALL发送失败' in source
assert 'self.message("GUI thermal off发送失败' in source
tests/fixtures/seat_frontend_hil_suite.json
Vendored
+5 -1
View File
@@ -353,7 +353,11 @@
]},
{"mode_id": "zero_gravity", "expected": {"longitudinal": 100, "cushion_tilt": 100, "backrest_recline": 81, "legrest_height": 94}, "variants": [
{"id": "MODE-ZERO-GRAVITY-OPEN", "utterance": "打开零重力模式"}, {"id": "MODE-ZERO-GRAVITY-ENTER", "utterance": "副驾进入零重力"}, {"id": "MODE-ZERO-GRAVITY-ENGLISH", "utterance": "开启 zero-gravity 模式"}
]}
]},
{"id": "MODE-ROCKING-HORSE-START", "mode_id": "rocking_horse", "utterance": "开启木马模式", "expected": {}, "expected_dynamic_mode": {"mode_id": 1, "state": "running"}},
{"id": "MODE-ROCKING-HORSE-STOP", "utterance": "关闭木马模式", "expected": {}, "expected_dynamic_mode": {"mode_id": 0, "state": "idle"}},
{"id": "MODE-DYNAMIC-CRADLE-START", "mode_id": "dynamic_cradle", "utterance": "开启动态摇篮模式", "expected": {}, "expected_dynamic_mode": {"mode_id": 2, "state": "running"}},
{"id": "MODE-DYNAMIC-CRADLE-STOP", "utterance": "关闭动态摇篮模式", "expected": {}, "expected_dynamic_mode": {"mode_id": 0, "state": "idle"}}
],
"model_cases": [
{
tests/integration/test_seat_frontend_input_coverage.py
+20 -1
View File
@@ -66,7 +66,11 @@ class SeatFrontendInputCoverageTests(unittest.TestCase):
{"thermal", "motion", "combined", "mode"},
)
self.assertEqual(
{case["mode_id"] for case in self.positive_cases if case["group"] == "mode"},
{
case["mode_id"]
for case in self.positive_cases
if case["group"] == "mode" and case.get("mode_id")
},
{
"easy_entry_exit",
"office",
@@ -79,6 +83,8 @@ class SeatFrontendInputCoverageTests(unittest.TestCase):
"electric_beach_chair",
"nap",
"zero_gravity",
"rocking_horse",
"dynamic_cradle",
},
)
@@ -150,6 +156,19 @@ class SeatFrontendInputCoverageTests(unittest.TestCase):
for field, expected in expected_motion.items():
if field != "axis":
self.assertEqual(operation["motion"][field], expected)
expected_dynamic_mode = case.get("expected_dynamic_mode")
if expected_dynamic_mode:
capability = (
"seat.stop_dynamic_mode"
if expected_dynamic_mode["mode_id"] == 0
else "seat.start_dynamic_mode"
)
self.assertIn(capability, by_capability)
if expected_dynamic_mode["mode_id"]:
self.assertEqual(
by_capability[capability]["parameters"]["mode_id"],
expected_dynamic_mode["mode_id"],
)
def test_every_rejection_utterance_stops_before_execution(self) -> None:
for case in self.rejection_cases:
tests/unit/test_native_can_integration.py
+12
View File
@@ -1,5 +1,6 @@
from __future__ import annotations
import os
import unittest
from pathlib import Path
from unittest.mock import patch
@@ -162,6 +163,17 @@ class NativeCanIntegrationTests(unittest.TestCase):
self.assertEqual(info["data"]["bitrate"], 2_000_000)
self.assertEqual(info["data"]["sample_point_percent"], 80)
def test_enabling_seat_hardware_does_not_implicitly_enable_vehicle_can(self) -> None:
with patch.dict(
os.environ,
{"COGNITIVE_AGENT_HARDWARE_EXECUTION_ENABLED": "true"},
clear=False,
):
os.environ.pop("COGNITIVE_AGENT_VEHICLE_CAN_ENABLED", None)
settings = VehicleCanSettings.from_env(Path(REPOSITORY_ROOT))
self.assertFalse(settings.enabled)
if __name__ == "__main__":
unittest.main()
tests/unit/test_self_repair.py
+23
View File
@@ -1,5 +1,7 @@
from __future__ import annotations
import json
import time
from pathlib import Path
from cognitive_agent.application.self_repair.healing_model import (
@@ -81,3 +83,24 @@ def test_bridge_manager_does_not_arm_visible_tooling_by_default(tmp_path: Path)
assert snapshot["state"] == "IDLE"
assert snapshot["retry_armed"] is False
assert snapshot["attempts"] == 0
def test_bridge_manager_does_not_restart_a_live_busy_request(tmp_path: Path) -> None:
status_path = tmp_path / "bridge.json"
status_path.write_text(
json.dumps({
"heartbeat_epoch": time.time() - 5.0,
"bridge_state": "RUNNING",
"current_request_id": "SELF_HEAL_BENCH_1",
}),
encoding="utf-8",
)
manager = ZcanBridgeManager(
status_path,
tmp_path / "bridge.py",
heartbeat_stale_s=3.0,
busy_heartbeat_grace_s=45.0,
)
assert manager._heartbeat_online() is False
assert manager._bridge_busy_within_grace() is True