Miayui
435 lines
18 KiB
Python
435 lines
18 KiB
Python
from __future__ import annotations
|
||
|
||
import json
|
||
from pathlib import Path
|
||
import shutil
|
||
import subprocess
|
||
import unittest
|
||
|
||
from cognitive_agent.application.conversation_state import ConversationStateStore
|
||
|
||
|
||
REPOSITORY_ROOT = Path(__file__).resolve().parents[2]
|
||
APP_SOURCE_PATH = REPOSITORY_ROOT / "apps" / "web-console" / "app.js"
|
||
EXTENSION_SOURCE_PATH = (
|
||
REPOSITORY_ROOT / "apps" / "web-console" / "self-extension.js"
|
||
)
|
||
|
||
|
||
def function_source(source: str, signature: str, next_signature: str) -> str:
|
||
start = source.index(signature)
|
||
end = source.index(next_signature, start)
|
||
return source[start:end]
|
||
|
||
|
||
class FrontendVehicleStateSyncContractTests(unittest.TestCase):
|
||
@classmethod
|
||
def setUpClass(cls) -> None:
|
||
cls.app_source = APP_SOURCE_PATH.read_text(encoding="utf-8")
|
||
cls.extension_source = EXTENSION_SOURCE_PATH.read_text(encoding="utf-8")
|
||
|
||
def test_both_modules_use_the_same_browser_conversation(self) -> None:
|
||
key = "cognitive-agent.conversation-id"
|
||
self.assertIn(
|
||
f'sessionStorage.getItem("{key}")',
|
||
self.app_source,
|
||
)
|
||
self.assertIn(
|
||
f'sessionStorage.setItem("{key}", conversationId)',
|
||
self.app_source,
|
||
)
|
||
self.assertIn(
|
||
f'const CONVERSATION_KEY = "{key}"',
|
||
self.extension_source,
|
||
)
|
||
self.assertIn(
|
||
"body: JSON.stringify({ command, conversation_id: sharedConversationId() })",
|
||
self.extension_source,
|
||
)
|
||
|
||
def test_shared_twin_theme_is_fixed_to_user(self) -> None:
|
||
self.assertIn("function applyTwinSceneTheme(", self.app_source)
|
||
self.assertIn('canvas.dataset.themeRole = "user"', self.app_source)
|
||
self.assertIn(
|
||
'canvas.dataset.themeAccent = `#${palette.accent.toString(16).padStart(6, "0")}`',
|
||
self.app_source,
|
||
)
|
||
for blue_theme_token in (
|
||
"accent: 0x2f80ed",
|
||
"vehicleBody: 0x91c6f5",
|
||
"vehicleOutline: 0x2879d8",
|
||
"seatShell: 0x2b76c9",
|
||
"glass: 0x4b9bea",
|
||
"lamp: 0x2e7fd4",
|
||
):
|
||
self.assertIn(blue_theme_token, self.app_source)
|
||
self.assertNotIn("accent: 0x0e7773", self.app_source)
|
||
self.assertNotIn("cognitive-agent:demo-role-change", self.app_source)
|
||
|
||
def test_shared_twin_ground_uses_theme_radial_alpha_texture(self) -> None:
|
||
self.assertIn("function createTwinGroundTexture(", self.app_source)
|
||
self.assertIn("new THREE.PlaneGeometry(30, 30)", self.app_source)
|
||
self.assertIn(
|
||
'surfaceGradient.addColorStop(0, "rgba(255, 255, 255, 1)")',
|
||
self.app_source,
|
||
)
|
||
self.assertIn(
|
||
'surfaceGradient.addColorStop(1, "rgba(255, 255, 255, 0)")',
|
||
self.app_source,
|
||
)
|
||
self.assertIn("const outerRadius = size * 0.48", self.app_source)
|
||
self.assertIn(
|
||
"const gridColor = accent.clone().lerp(new THREE.Color(0xffffff), 0.52)",
|
||
self.app_source,
|
||
)
|
||
self.assertIn("gridContext.lineWidth = major ? 0.9 : 0.45", self.app_source)
|
||
self.assertIn('gridContext.globalCompositeOperation = "destination-in"', self.app_source)
|
||
self.assertIn("applyTwinGroundTheme(twin, palette)", self.app_source)
|
||
self.assertNotIn("new THREE.GridHelper(10, 20", self.app_source)
|
||
|
||
def test_lan_http_origin_does_not_require_crypto_random_uuid(self) -> None:
|
||
id_source = function_source(
|
||
self.app_source,
|
||
"function createOpaqueId(prefix) {",
|
||
"\nfunction createConversationId",
|
||
)
|
||
self.assertIn("globalThis.crypto?.randomUUID", id_source)
|
||
self.assertIn("globalThis.crypto?.getRandomValues", id_source)
|
||
self.assertIn("Math.random()", id_source)
|
||
self.assertIn('return createOpaqueId("conv")', self.app_source)
|
||
self.assertIn('createOpaqueId("hwc")', self.app_source)
|
||
|
||
def test_extension_accepts_only_explicit_physical_ch1_state(self) -> None:
|
||
authority_source = function_source(
|
||
self.extension_source,
|
||
" function isAuthoritativeVehicleStatePayload(payload) {",
|
||
"\n function renderAuthoritativeVehicleState",
|
||
)
|
||
self.assertIn(
|
||
'payload.conversation_id !== sharedConversationId()', authority_source
|
||
)
|
||
self.assertIn(
|
||
"containsAuthoritativePhysicalCh1State(payload.actuator_state)",
|
||
authority_source,
|
||
)
|
||
self.assertIn(
|
||
"containsAuthoritativePhysicalCh1State(payload.domain_state)",
|
||
authority_source,
|
||
)
|
||
self.assertNotIn("physical_feedback_confirmed", authority_source)
|
||
|
||
summary_source = function_source(
|
||
self.extension_source,
|
||
" function renderAuthoritativeVehicleState(payload) {",
|
||
"\n function setMode",
|
||
)
|
||
self.assertNotIn(
|
||
"items.length === 1 && payload.physical_feedback_confirmed",
|
||
summary_source,
|
||
)
|
||
for state_path in (
|
||
"ventilation?.feedback_source",
|
||
"heating?.feedback_source",
|
||
"climate?.feedback_source",
|
||
"roof?.feedback_source",
|
||
"frontLeftWindow?.feedback_source",
|
||
):
|
||
self.assertIn(f'{state_path} === "physical_can_ch1"', summary_source)
|
||
|
||
def test_orchestration_renders_only_the_authoritative_ch1_patch(self) -> None:
|
||
apply_source = function_source(
|
||
self.app_source,
|
||
"function applyAuthoritativeVehicleState(payload) {",
|
||
'\nwindow.addEventListener("cognitive-agent:extension-vehicle-result"',
|
||
)
|
||
self.assertIn(
|
||
"const actuatorPatch = authoritativeCh1StatePatch(payload.actuator_state)",
|
||
apply_source,
|
||
)
|
||
self.assertIn(
|
||
"const domainPatch = authoritativeCh1StatePatch(payload.domain_state)",
|
||
apply_source,
|
||
)
|
||
self.assertIn("renderSeatState(actuatorPatch)", apply_source)
|
||
self.assertIn("renderDomainState(domainPatch)", apply_source)
|
||
self.assertIn("if (actuatorPatch?.driver)", apply_source)
|
||
self.assertNotIn("renderSeatState(payload.actuator_state)", apply_source)
|
||
self.assertNotIn("renderDomainState(payload.domain_state)", apply_source)
|
||
|
||
render_source = function_source(
|
||
self.app_source,
|
||
"function renderResult(result) {",
|
||
"\nasync function submitRequest",
|
||
)
|
||
ch1_branch = function_source(
|
||
render_source,
|
||
" if (ch1HardwareResult) {",
|
||
" updateTwinExplicitAngles",
|
||
)
|
||
self.assertIn("applyAuthoritativeVehicleState(result)", ch1_branch)
|
||
self.assertIn("} else {", ch1_branch)
|
||
self.assertIn("renderSeatState(result.actuator_state)", ch1_branch)
|
||
self.assertIn("renderDomainState(result.domain_state)", ch1_branch)
|
||
|
||
def test_authoritative_patch_accepts_wired_and_wireless_without_promoting_siblings(self) -> None:
|
||
node = shutil.which("node")
|
||
if node is None:
|
||
self.skipTest("Node.js is required to execute the frontend patch helper")
|
||
source_helper = function_source(
|
||
self.app_source,
|
||
"function isAuthoritativeCh1Source(value) {",
|
||
"\nfunction renderCabinLiveStatus",
|
||
)
|
||
patch_source = function_source(
|
||
self.app_source,
|
||
"function authoritativeCh1StatePatch(value) {",
|
||
"\nfunction isAuthoritativeVehicleStatePayload",
|
||
)
|
||
state = {
|
||
"climate": {
|
||
"feedback_source": "physical_can_ch1",
|
||
"power": "on",
|
||
},
|
||
"roof": {"shade_position_pct": 100},
|
||
"ambient_light": {
|
||
"feedback_source": "wireless_mqtt_ch1",
|
||
"power": "on",
|
||
},
|
||
"window": {
|
||
"positions": {"front_left": {"position_pct": 50}}
|
||
},
|
||
}
|
||
script = (
|
||
'const AUTHORITATIVE_CH1_SOURCES = new Set(["physical_can_ch1", "wireless_mqtt_ch1"]);\n'
|
||
f"{source_helper}\n"
|
||
f"{patch_source}\n"
|
||
f"const input = {json.dumps(state)};\n"
|
||
"process.stdout.write(JSON.stringify(authoritativeCh1StatePatch(input)));"
|
||
)
|
||
completed = subprocess.run(
|
||
[node, "-e", script],
|
||
check=True,
|
||
capture_output=True,
|
||
text=True,
|
||
)
|
||
self.assertEqual(
|
||
json.loads(completed.stdout),
|
||
{
|
||
"climate": state["climate"],
|
||
"ambient_light": state["ambient_light"],
|
||
},
|
||
)
|
||
|
||
def test_dual_channel_runtime_polls_each_seat_without_aliasing(self) -> None:
|
||
passenger_source = function_source(
|
||
self.app_source,
|
||
"async function synchronizePassengerSeatState() {",
|
||
"\nasync function synchronizeDriverSeatState()",
|
||
)
|
||
driver_source = function_source(
|
||
self.app_source,
|
||
"async function synchronizeDriverSeatState() {",
|
||
"\nasync function synchronizePhysicalSeatState()",
|
||
)
|
||
combined_source = function_source(
|
||
self.app_source,
|
||
"async function synchronizePhysicalSeatState() {",
|
||
"\nasync function loadAuthStatus()",
|
||
)
|
||
self.assertIn('requestedTransport === "wireless_mqtt"', passenger_source)
|
||
self.assertIn("selectedVehicleTransport()?.state_configured", passenger_source)
|
||
self.assertIn("control_transport: requestedTransport", passenger_source)
|
||
self.assertIn("requestedTransport !== state.controlTransport", passenger_source)
|
||
self.assertIn('payload.source === "wireless_mqtt_ch0"', passenger_source)
|
||
self.assertIn("/api/hardware/seat-state", passenger_source)
|
||
self.assertIn("selectedVehicleTransport()?.state_configured", driver_source)
|
||
self.assertIn("control_transport: requestedTransport", driver_source)
|
||
self.assertIn("requestedTransport !== state.controlTransport", driver_source)
|
||
self.assertIn("/api/hardware/vehicle-state", driver_source)
|
||
self.assertIn('state.controlTransport === "wireless_mqtt"', combined_source)
|
||
self.assertIn("state.passengerStateSyncPromise", combined_source)
|
||
self.assertLess(
|
||
combined_source.index("synchronizePassengerSeatState()"),
|
||
combined_source.index("synchronizeDriverSeatState()"),
|
||
)
|
||
render_source = function_source(
|
||
self.app_source,
|
||
"function renderSeatState(actuatorState) {",
|
||
"\nfunction renderDomainState",
|
||
)
|
||
self.assertNotIn("passenger: mergedActuatorState.driver", render_source)
|
||
|
||
def test_state_polling_surfaces_structured_403_reason(self) -> None:
|
||
passenger_source = function_source(
|
||
self.app_source,
|
||
"async function synchronizePassengerSeatState() {",
|
||
"\nasync function synchronizeDriverSeatState()",
|
||
)
|
||
driver_source = function_source(
|
||
self.app_source,
|
||
"async function synchronizeDriverSeatState() {",
|
||
"\nasync function synchronizePhysicalSeatState()",
|
||
)
|
||
live_status_source = function_source(
|
||
self.app_source,
|
||
"function renderCabinLiveStatus(",
|
||
"\nfunction refreshIcons",
|
||
)
|
||
telemetry_source = function_source(
|
||
self.app_source,
|
||
"function recordVehicleStateTelemetry(payload) {",
|
||
"\nfunction applyAuthoritativeVehicleState",
|
||
)
|
||
|
||
self.assertIn("throw await apiResponseError(response)", passenger_source)
|
||
self.assertIn("throw await apiResponseError(response)", driver_source)
|
||
self.assertIn("errorStatus: Number(error.status) || null", driver_source)
|
||
self.assertIn("errorReason: error.serverMessage", driver_source)
|
||
self.assertIn("snapshot?.errorStatus === 403", live_status_source)
|
||
self.assertIn("`${transportLabel} · 403：${reason}`", live_status_source)
|
||
self.assertIn('telemetryState.textContent = "403 拒绝"', live_status_source)
|
||
self.assertIn('setSummarySystem("403 · 状态读取被拒绝", "warning")', live_status_source)
|
||
self.assertIn("errorStatus: null", telemetry_source)
|
||
self.assertIn('errorReason: ""', telemetry_source)
|
||
|
||
def test_persistent_linux_monitor_uses_backend_selected_fast_polling(self) -> None:
|
||
interval_source = function_source(
|
||
self.app_source,
|
||
"function physicalStatePollIntervalMs() {",
|
||
"\nfunction refreshIcons",
|
||
)
|
||
self.assertIn("selectedVehicleTransport()?.state_sync?.poll_interval_ms", interval_source)
|
||
self.assertIn("Math.max(250", interval_source)
|
||
self.assertIn("DEFAULT_PHYSICAL_STATE_POLL_INTERVAL_MS = 5000", self.app_source)
|
||
self.assertIn("}, physicalStatePollIntervalMs());", self.app_source)
|
||
self.assertIn("restartPhysicalStatePolling();", self.app_source)
|
||
|
||
def test_transport_switch_clears_stale_vehicle_values_before_new_feedback(self) -> None:
|
||
select_source = function_source(
|
||
self.app_source,
|
||
"function selectControlTransport(transport) {",
|
||
"\nfunction renderHardwareAccess",
|
||
)
|
||
reset_source = function_source(
|
||
self.app_source,
|
||
"function resetVehicleStatusDisplay() {",
|
||
"\nfunction mergeSeatActuatorState",
|
||
)
|
||
|
||
self.assertIn("resetVehicleStatusDisplay();", select_source)
|
||
self.assertIn('setText("#windowState", "UNKNOWN")', reset_source)
|
||
self.assertIn('setText("#frontLightState", "UNKNOWN")', reset_source)
|
||
self.assertIn('setText("#cabinSummaryDoors", "状态未知")', reset_source)
|
||
self.assertIn('meter.dataset.state = "unknown"', reset_source)
|
||
|
||
def test_vehicle_selection_cannot_replay_unfiltered_state_in_ch1_mode(self) -> None:
|
||
select_source = function_source(
|
||
self.app_source,
|
||
"function selectVehicle(vehicleId, { silent = false } = {}) {",
|
||
"\nfunction populateVehicleSelector",
|
||
)
|
||
driver_branch = select_source.index("if (isDriverOnlyCh1Runtime())")
|
||
filtered_snapshot = select_source.index(
|
||
"state.vehicleStateSnapshot?.actuatorState",
|
||
driver_branch,
|
||
)
|
||
unfiltered_result = select_source.index("state.result?.actuator_state")
|
||
self.assertLess(driver_branch, filtered_snapshot)
|
||
self.assertLess(filtered_snapshot, unfiltered_result)
|
||
self.assertIn("} else if (state.result?.actuator_state)", select_source)
|
||
|
||
def test_driver_only_ch1_has_no_hidden_browser_lease_gate(self) -> None:
|
||
submit_source = function_source(
|
||
self.app_source,
|
||
"async function submitRequest(event) {",
|
||
"\nasync function copyArtifact",
|
||
)
|
||
self.assertIn("const driverOnlyCh1 = isDriverOnlyCh1Runtime()", submit_source)
|
||
self.assertIn("&& !driverOnlyCh1", submit_source)
|
||
self.assertIn(
|
||
'&& state.controlTransport !== "wireless_mqtt"',
|
||
submit_source,
|
||
)
|
||
startup_source = self.app_source[
|
||
self.app_source.index('document.addEventListener("DOMContentLoaded"') :
|
||
]
|
||
self.assertIn("if (!isDriverOnlyCh1Runtime())", startup_source)
|
||
self.assertIn("loadHardwareAccessStatus", startup_source)
|
||
|
||
def test_state_results_flow_in_both_directions(self) -> None:
|
||
self.assertIn(
|
||
'window.addEventListener("cognitive-agent:extension-vehicle-result"',
|
||
self.app_source,
|
||
)
|
||
self.assertIn(
|
||
'window.addEventListener("cognitive-agent:result"',
|
||
self.extension_source,
|
||
)
|
||
|
||
|
||
class PhysicalMarkerInvalidationContractTests(unittest.TestCase):
|
||
def test_logical_door_and_window_actions_replace_stale_physical_roots(self) -> None:
|
||
store = ConversationStateStore()
|
||
conversation_id = "conv-frontend-marker-invalidation"
|
||
store.synchronize_physical_ch1_feedback(
|
||
conversation_id,
|
||
[
|
||
{
|
||
"source": "physical_can_ch1",
|
||
"channel": 1,
|
||
"frame_id": "0x3B6",
|
||
"feedback_frame_id": "0x3B6",
|
||
"valid": True,
|
||
"verification": {
|
||
"authoritative": True,
|
||
"channel": 1,
|
||
"simulated": False,
|
||
"feedback_required": True,
|
||
"feedback_observed": True,
|
||
"feedback_match": True,
|
||
"frame_id": "0x3B6",
|
||
},
|
||
"domain_state": {
|
||
"door": {"locks": {"rear_left": {"state": "lock"}}},
|
||
"window": {
|
||
"positions": {
|
||
"front_left": {
|
||
"position_pct": 20,
|
||
"closed": False,
|
||
}
|
||
}
|
||
},
|
||
},
|
||
}
|
||
],
|
||
)
|
||
physical = store.snapshot(conversation_id)["domain_devices"]
|
||
self.assertEqual(physical["door"]["feedback_source"], "physical_can_ch1")
|
||
self.assertEqual(physical["window"]["feedback_source"], "physical_can_ch1")
|
||
|
||
updated = store.apply_successful_actions(
|
||
conversation_id,
|
||
"logical projection must not look physical",
|
||
[
|
||
{
|
||
"domain": "door",
|
||
"capability": "door.set_rear_lock",
|
||
"target": "rear",
|
||
"parameters": {"state": "unlock"},
|
||
},
|
||
{
|
||
"domain": "window",
|
||
"capability": "window.set_position",
|
||
"target": "front_left",
|
||
"parameters": {"position_pct": 60},
|
||
},
|
||
],
|
||
[],
|
||
)["domain_devices"]
|
||
self.assertEqual(updated["door"]["feedback_source"], "virtual")
|
||
self.assertEqual(updated["window"]["feedback_source"], "virtual")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|