Miayui
1864 lines
71 KiB
Python
1864 lines
71 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import tempfile
|
|
import time
|
|
import unittest
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
from unittest.mock import patch
|
|
|
|
from cognitive_agent.application.native_can_runtime import NativeCanRuntime
|
|
from cognitive_agent.application.seat_service import SeatDemoService
|
|
from cognitive_agent.infrastructure.can.gateway import CanFrame
|
|
from cognitive_agent.infrastructure.can.hardware import (
|
|
HardwareSafetyError,
|
|
VehicleMqttSettings,
|
|
)
|
|
from cognitive_agent.infrastructure.can.mqtt_transport import (
|
|
MQTT_CAN_SCHEMA,
|
|
MQTT_EXTENSION_CONTROL_TOPIC,
|
|
MQTT_EXTENSION_TOY_TOPIC,
|
|
MQTT_MODE_SCENE_INGRESS_TOPIC,
|
|
MQTT_MODE_SCENE_TOPIC,
|
|
MQTT_N7_CONTROL_TRANSFER_TOPIC,
|
|
MQTT_PET_TOPIC,
|
|
MqttCanTransport,
|
|
mqtt_extension_toy_payload,
|
|
mqtt_mode_scene_payload,
|
|
mqtt_n7_control_transfer_payload,
|
|
mqtt_pet_payload,
|
|
parse_mqtt_can_payload,
|
|
)
|
|
|
|
REPOSITORY_ROOT = Path(__file__).resolve().parents[2]
|
|
|
|
|
|
class _PublishResult:
|
|
rc = 0
|
|
mid = 17
|
|
|
|
def is_published(self) -> bool:
|
|
return True
|
|
|
|
def wait_for_publish(self, timeout: float) -> bool:
|
|
raise AssertionError("control publish must never block for acknowledgement")
|
|
|
|
|
|
class _FakeMqttClient:
|
|
def __init__(self) -> None:
|
|
self.on_connect = None
|
|
self.on_disconnect = None
|
|
self.on_message = None
|
|
self.subscriptions: list[tuple[str, int]] = []
|
|
self.published: list[tuple[str, str, int, bool]] = []
|
|
self.websocket_options: list[dict[str, object]] = []
|
|
self.credentials: tuple[str, str | None] | None = None
|
|
self.tls_configured = False
|
|
|
|
def reconnect_delay_set(self, **_kwargs) -> None:
|
|
return None
|
|
|
|
def ws_set_options(self, **kwargs: object) -> None:
|
|
self.websocket_options.append(kwargs)
|
|
|
|
def username_pw_set(self, username: str, password: str | None) -> None:
|
|
self.credentials = (username, password)
|
|
|
|
def tls_set(self) -> None:
|
|
self.tls_configured = True
|
|
|
|
def connect_async(self, host: str, port: int, keepalive: int) -> None:
|
|
self.connection = (host, port, keepalive)
|
|
|
|
def loop_start(self) -> None:
|
|
self.on_connect(self, None, {}, 0, None)
|
|
|
|
def loop_stop(self) -> None:
|
|
return None
|
|
|
|
def disconnect(self) -> None:
|
|
return None
|
|
|
|
def subscribe(self, topic: str, qos: int):
|
|
self.subscriptions.append((topic, qos))
|
|
return (0, 1)
|
|
|
|
def publish(self, topic: str, payload: str, qos: int, retain: bool):
|
|
self.published.append((topic, payload, qos, retain))
|
|
return _PublishResult()
|
|
|
|
def emit(
|
|
self,
|
|
topic: str,
|
|
payload: object,
|
|
*,
|
|
retain: bool = False,
|
|
dup: bool = False,
|
|
) -> None:
|
|
message = SimpleNamespace(
|
|
topic=topic,
|
|
payload=json.dumps(payload).encode("utf-8"),
|
|
retain=retain,
|
|
dup=dup,
|
|
)
|
|
self.on_message(self, None, message)
|
|
|
|
|
|
def _settings():
|
|
values = {
|
|
"enabled": True,
|
|
"state_configured": True,
|
|
"configured": True,
|
|
"host": "127.0.0.1",
|
|
"port": 1883,
|
|
"condition_topic": "n7/n7-headunit/can/state",
|
|
"control_topic": "n7/n7-headunit/can/command",
|
|
"qos": 1,
|
|
"keepalive_s": 30,
|
|
"username": None,
|
|
"password": None,
|
|
"tls_enabled": False,
|
|
"feedback_timeout_ms": 100,
|
|
"transport": "tcp",
|
|
"websocket_path": "/mqtt",
|
|
}
|
|
values["public_info"] = lambda: {
|
|
"enabled": True,
|
|
"configured": True,
|
|
"state_configured": True,
|
|
"backend": "mqtt",
|
|
"condition_topic": "n7/n7-headunit/can/state",
|
|
"control_topic": "n7/n7-headunit/can/command",
|
|
}
|
|
return SimpleNamespace(**values)
|
|
|
|
|
|
class MqttCanPayloadTests(unittest.TestCase):
|
|
def test_default_topics_and_public_contracts_match_headunit(self) -> None:
|
|
with patch.dict("os.environ", {}, clear=True):
|
|
settings = VehicleMqttSettings.from_env()
|
|
info = settings.public_info()
|
|
self.assertEqual(settings.condition_topic, "n7/n7-headunit/can/state")
|
|
self.assertEqual(settings.control_topic, "n7/n7-headunit/can/command")
|
|
self.assertFalse(settings.approval_required)
|
|
self.assertEqual(settings.transport, "tcp")
|
|
self.assertEqual(settings.websocket_path, "/mqtt")
|
|
self.assertEqual(info["state_contract"], "n7-can-signals/v1")
|
|
self.assertEqual(info["control_contract"], "n7-headunit-can-command/v1")
|
|
|
|
def test_websocket_transport_settings_are_loaded_from_environment(self) -> None:
|
|
with patch.dict(
|
|
"os.environ",
|
|
{
|
|
"COGNITIVE_AGENT_VEHICLE_MQTT_HOST": "mqtt.example.test",
|
|
"COGNITIVE_AGENT_VEHICLE_MQTT_PORT": "443",
|
|
"COGNITIVE_AGENT_VEHICLE_MQTT_TRANSPORT": "websocket",
|
|
"COGNITIVE_AGENT_VEHICLE_MQTT_WEBSOCKET_PATH": "/mqtt",
|
|
"COGNITIVE_AGENT_VEHICLE_MQTT_TLS": "true",
|
|
},
|
|
clear=True,
|
|
):
|
|
settings = VehicleMqttSettings.from_env()
|
|
|
|
self.assertTrue(settings.adapter_configured)
|
|
self.assertEqual(settings.transport, "websockets")
|
|
self.assertEqual(settings.websocket_path, "/mqtt")
|
|
self.assertTrue(settings.tls_enabled)
|
|
self.assertEqual(settings.public_info()["transport"], "websockets")
|
|
|
|
def test_explicit_shared_approval_policy_still_locks_mqtt_control(self) -> None:
|
|
with patch.dict(
|
|
"os.environ",
|
|
{"COGNITIVE_AGENT_HARDWARE_APPROVAL_TOKEN_REQUIRED": "true"},
|
|
clear=True,
|
|
):
|
|
settings = VehicleMqttSettings.from_env()
|
|
self.assertTrue(settings.approval_required)
|
|
self.assertFalse(settings.configured)
|
|
|
|
def test_accepts_one_canonical_frame_and_common_receive_aliases(self) -> None:
|
|
canonical = parse_mqtt_can_payload(
|
|
json.dumps(
|
|
{
|
|
"schema": MQTT_CAN_SCHEMA,
|
|
"arbitration_id": 0x342,
|
|
"channel": 1,
|
|
"data_hex": "00 00 00 01 00 00 00 00",
|
|
"is_fd": True,
|
|
"bitrate_switch": True,
|
|
}
|
|
)
|
|
)[0]
|
|
alias = parse_mqtt_can_payload(
|
|
json.dumps({"canId": "0x4FA", "data": [0] * 8})
|
|
)[0]
|
|
self.assertEqual(canonical.arbitration_id, 0x342)
|
|
self.assertEqual(canonical.channel, 1)
|
|
self.assertTrue(canonical.is_fd)
|
|
self.assertEqual(alias.arbitration_id, 0x4FA)
|
|
|
|
def test_accepts_vehicle_n7_can_signals_envelope(self) -> None:
|
|
frames = parse_mqtt_can_payload(
|
|
json.dumps(
|
|
{
|
|
"schema": "n7-can-signals/v1",
|
|
"sequence": 335,
|
|
"timestamp": 1787037724090,
|
|
"online": True,
|
|
"signals": [
|
|
{
|
|
"channel": "can1",
|
|
"channelIndex": 1,
|
|
"canId": "0x388",
|
|
"canIdDecimal": 904,
|
|
"frameType": "fd",
|
|
"brs": True,
|
|
"dlc": 8,
|
|
"data": "00 00 00 02 01 00 59 00",
|
|
"bytes": [0, 0, 0, 2, 1, 0, 89, 0],
|
|
"timestamp": 1787037667665,
|
|
"direction": "rx",
|
|
}
|
|
],
|
|
"can1Vehicle": {
|
|
"parsed": {"speedKph": 0, "gear": 1}
|
|
},
|
|
}
|
|
)
|
|
)
|
|
self.assertEqual(len(frames), 1)
|
|
self.assertEqual(frames[0].arbitration_id, 0x388)
|
|
self.assertEqual(frames[0].channel, 1)
|
|
self.assertEqual(frames[0].data, bytes.fromhex("00 00 00 02 01 00 59 00"))
|
|
self.assertTrue(frames[0].is_fd)
|
|
self.assertTrue(frames[0].bitrate_switch)
|
|
self.assertEqual(frames[0].timestamp, 1787037667.665)
|
|
|
|
def test_accepts_nested_can1_vehicle_signals_when_top_level_is_absent(self) -> None:
|
|
frame = parse_mqtt_can_payload(
|
|
json.dumps(
|
|
{
|
|
"schema": "n7-can-signals/v1",
|
|
"can1Vehicle": {
|
|
"signals": [
|
|
{
|
|
"channel": "can1",
|
|
"canId": "0x42E",
|
|
"frameType": "fd",
|
|
"brs": True,
|
|
"bytes": [17, 100, 50, 3, 144, 2, 0, 198],
|
|
}
|
|
]
|
|
},
|
|
}
|
|
)
|
|
)[0]
|
|
self.assertEqual(frame.arbitration_id, 0x42E)
|
|
self.assertTrue(frame.is_fd)
|
|
|
|
def test_accepts_native_passenger_seat_firmware_status(self) -> None:
|
|
frames = parse_mqtt_can_payload(
|
|
json.dumps(
|
|
{
|
|
"schema": "n7-can-signals/v1",
|
|
"can0Seat": {
|
|
"signals": [
|
|
{
|
|
"canId": "0x302",
|
|
"frameType": "classic",
|
|
"bytes": [0] * 8,
|
|
}
|
|
]
|
|
},
|
|
}
|
|
)
|
|
)
|
|
self.assertEqual(len(frames), 1)
|
|
self.assertEqual(frames[0].channel, 0)
|
|
self.assertEqual(frames[0].arbitration_id, 0x302)
|
|
|
|
def test_translates_seat_ota_info_responses_into_four_position_states(self) -> None:
|
|
values = {
|
|
1: (480 << 16) | 407,
|
|
2: (373 << 16) | 266,
|
|
3: (917 << 16) | 451,
|
|
4: (647 << 16) | 575,
|
|
}
|
|
signals = []
|
|
for selector, value in values.items():
|
|
sequence = 0x20 + selector
|
|
signals.extend(
|
|
(
|
|
{
|
|
"channel": "can0",
|
|
"canId": "0x6A0",
|
|
"frameType": "classic",
|
|
"direction": "tx",
|
|
"bytes": [0xC7, sequence, 0x60, selector, 0, 0, 0, 0],
|
|
},
|
|
{
|
|
"channel": "can0",
|
|
"canId": "0x6A2",
|
|
"frameType": "classic",
|
|
"direction": "rx",
|
|
"bytes": [0x7C, sequence, 0x60, 0, *value.to_bytes(4, "little")],
|
|
},
|
|
)
|
|
)
|
|
|
|
frames = parse_mqtt_can_payload(json.dumps({"signals": signals}))
|
|
positions = [frame for frame in frames if frame.arbitration_id == 0x302]
|
|
|
|
self.assertEqual([frame.data[0] for frame in positions], [0, 1, 2, 3])
|
|
self.assertEqual(
|
|
[int.from_bytes(frame.data[5:7], "little") for frame in positions],
|
|
[round(407 * 10_000 / 480), round(266 * 10_000 / 373),
|
|
round(451 * 10_000 / 917), round(575 * 10_000 / 647)],
|
|
)
|
|
self.assertTrue(all(frame.data[2] & (1 << 3) for frame in positions))
|
|
|
|
def test_translates_headunit_6a2_selector_metadata_without_6a0_requests(
|
|
self,
|
|
) -> None:
|
|
values = {
|
|
1: bytes.fromhex("E0 01 E0 01"), # 480 / 480
|
|
2: bytes.fromhex("00 00 75 01"), # 0 / 373
|
|
3: bytes.fromhex("00 00 95 03"), # 0 / 917
|
|
4: bytes.fromhex("BF 01 87 02"), # 447 / 647
|
|
5: bytes.fromhex("11 00 00 00"),
|
|
6: bytes.fromhex("11 00 00 00"),
|
|
7: bytes.fromhex("11 00 02 00"),
|
|
8: bytes.fromhex("11 00 00 00"),
|
|
}
|
|
frames = parse_mqtt_can_payload(
|
|
json.dumps(
|
|
{
|
|
"schema": "n7-can-signals/v1",
|
|
"signals": [
|
|
{
|
|
"channel": "can0",
|
|
"channelIndex": 0,
|
|
"canId": "0x6A2",
|
|
"frameType": "classic",
|
|
"brs": False,
|
|
"dlc": 8,
|
|
"selector": selector,
|
|
"bytes": [0x7C, 0x30 + selector, 0x60, 0, *value],
|
|
"direction": "rx",
|
|
}
|
|
for selector, value in values.items()
|
|
],
|
|
}
|
|
)
|
|
)
|
|
|
|
positions = [frame for frame in frames if frame.arbitration_id == 0x302]
|
|
self.assertEqual([frame.data[0] for frame in positions], [0, 1, 2, 3])
|
|
self.assertEqual(
|
|
[int.from_bytes(frame.data[5:7], "little") for frame in positions],
|
|
[10_000, 0, 0, round(447 * 10_000 / 647)],
|
|
)
|
|
self.assertTrue(all(frame.data[2] & (1 << 3) for frame in positions))
|
|
|
|
def test_rejects_invalid_or_conflicting_headunit_seat_selector_metadata(
|
|
self,
|
|
) -> None:
|
|
with self.assertRaisesRegex(ValueError, "integer from 0 to 14"):
|
|
parse_mqtt_can_payload(
|
|
json.dumps(
|
|
{
|
|
"channel": "can0",
|
|
"canId": "0x6A2",
|
|
"selector": "1",
|
|
"bytes": [0x7C, 1, 0x60, 0, 0xE0, 1, 0xE0, 1],
|
|
}
|
|
)
|
|
)
|
|
with self.assertRaisesRegex(ValueError, "valid 0x6A2"):
|
|
parse_mqtt_can_payload(
|
|
json.dumps(
|
|
{
|
|
"channel": "can0",
|
|
"canId": "0x302",
|
|
"selector": 1,
|
|
"bytes": [0] * 8,
|
|
}
|
|
)
|
|
)
|
|
|
|
frames = parse_mqtt_can_payload(
|
|
json.dumps(
|
|
{
|
|
"signals": [
|
|
{"channel": "can0", "canId": "0x6A0", "bytes":
|
|
[0xC7, 0x21, 0x60, 1, 0, 0, 0, 0]},
|
|
{"channel": "can0", "canId": "0x6A2", "selector": 2,
|
|
"bytes": [0x7C, 0x21, 0x60, 0, 0xE0, 1, 0xE0, 1]},
|
|
]
|
|
}
|
|
)
|
|
)
|
|
self.assertFalse(any(frame.arbitration_id == 0x302 for frame in frames))
|
|
|
|
def test_rejects_can0_frames_outside_seat_firmware_format(self) -> None:
|
|
for label, frame in (
|
|
(
|
|
"fd",
|
|
{
|
|
"channel": "can0",
|
|
"canId": "0x6A0",
|
|
"frameType": "fd",
|
|
"bytes": [0] * 8,
|
|
},
|
|
),
|
|
(
|
|
"brs",
|
|
{
|
|
"channel": "can0",
|
|
"canId": "0x6A2",
|
|
"frameType": "classic",
|
|
"brs": True,
|
|
"bytes": [0] * 8,
|
|
},
|
|
),
|
|
(
|
|
"dlc",
|
|
{
|
|
"channel": "can0",
|
|
"canId": "0x6A2",
|
|
"frameType": "classic",
|
|
"bytes": [0] * 7,
|
|
},
|
|
),
|
|
):
|
|
with self.subTest(label=label), self.assertRaisesRegex(
|
|
ValueError, "classic CAN, no BRS, and DLC 8"
|
|
):
|
|
parse_mqtt_can_payload(json.dumps(frame))
|
|
|
|
def test_top_level_signals_can_mix_can0_and_can1_without_nested_duplicates(self) -> None:
|
|
frames = parse_mqtt_can_payload(
|
|
json.dumps(
|
|
{
|
|
"schema": "n7-can-signals/v1",
|
|
"signals": [
|
|
{"channel": "can0", "canId": "0x302", "bytes": [0] * 8},
|
|
{"channel": "can1", "canId": "0x342", "bytes": [0] * 8},
|
|
],
|
|
"can0Seat": {
|
|
"signals": [
|
|
{"channel": "can0", "canId": "0x302", "bytes": [0] * 8}
|
|
]
|
|
},
|
|
}
|
|
)
|
|
)
|
|
self.assertEqual(
|
|
[(frame.channel, frame.arbitration_id) for frame in frames],
|
|
[(0, 0x302), (1, 0x342)],
|
|
)
|
|
|
|
def test_rejects_decoded_state_invalid_channels_and_channel_conflicts(self) -> None:
|
|
with self.assertRaisesRegex(ValueError, "requires an ID and data"):
|
|
parse_mqtt_can_payload(json.dumps({"gear": "P", "speed_kmh": 0}))
|
|
with self.assertRaisesRegex(ValueError, "only CAN0 or CAN1"):
|
|
parse_mqtt_can_payload(
|
|
json.dumps({"id": 0x342, "channel": 2, "data": [0] * 8})
|
|
)
|
|
with self.assertRaisesRegex(ValueError, "declarations conflict"):
|
|
parse_mqtt_can_payload(
|
|
json.dumps(
|
|
{
|
|
"id": 0x342,
|
|
"channel": "can1",
|
|
"channelIndex": 0,
|
|
"data": [0] * 8,
|
|
}
|
|
)
|
|
)
|
|
|
|
def test_control_remains_locked_without_existing_hardware_approval(self) -> None:
|
|
settings = VehicleMqttSettings(
|
|
enabled=True,
|
|
approval_required=True,
|
|
approval_token="",
|
|
host="127.0.0.1",
|
|
port=1883,
|
|
condition_topic="n7/n7-headunit/can/state",
|
|
control_topic="n7/n7-headunit/can/command",
|
|
qos=1,
|
|
keepalive_s=30,
|
|
username=None,
|
|
password=None,
|
|
tls_enabled=False,
|
|
feedback_timeout_ms=1_500,
|
|
)
|
|
with patch(
|
|
"cognitive_agent.infrastructure.can.hardware._module_available",
|
|
return_value=True,
|
|
):
|
|
self.assertTrue(settings.state_configured)
|
|
self.assertFalse(settings.configured)
|
|
with self.assertRaisesRegex(HardwareSafetyError, "not configured"):
|
|
settings.authorize(None)
|
|
|
|
|
|
class MqttCanTransportTests(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.client = _FakeMqttClient()
|
|
self.transport = MqttCanTransport(
|
|
_settings(),
|
|
client_factory=lambda: self.client,
|
|
)
|
|
self.transport.start()
|
|
|
|
def tearDown(self) -> None:
|
|
self.transport.close()
|
|
|
|
def test_websocket_client_applies_transport_path_authentication_and_tls(self) -> None:
|
|
client = _FakeMqttClient()
|
|
settings = _settings()
|
|
settings.transport = "websockets"
|
|
settings.websocket_path = "/mqtt"
|
|
settings.username = "n7web"
|
|
settings.password = "secret"
|
|
settings.tls_enabled = True
|
|
transport = MqttCanTransport(settings)
|
|
|
|
with patch("paho.mqtt.client.Client", return_value=client) as constructor:
|
|
transport.start()
|
|
try:
|
|
self.assertEqual(
|
|
constructor.call_args.kwargs["transport"], "websockets"
|
|
)
|
|
self.assertEqual(client.websocket_options, [{"path": "/mqtt"}])
|
|
self.assertEqual(client.credentials, ("n7web", "secret"))
|
|
self.assertTrue(client.tls_configured)
|
|
self.assertEqual(client.connection, ("127.0.0.1", 1883, 30))
|
|
finally:
|
|
transport.close()
|
|
|
|
def test_pet_message_uses_fixed_topic_exact_json_and_never_retains(self) -> None:
|
|
payload = {
|
|
"pet_name": "团团",
|
|
"nfc_id": "E113B203",
|
|
"action": "摇尾巴",
|
|
}
|
|
|
|
publication = self.transport.publish_pet_message(payload)
|
|
|
|
topic, raw_payload, qos, retain = self.client.published[-1]
|
|
self.assertEqual(topic, MQTT_PET_TOPIC)
|
|
self.assertEqual(raw_payload, mqtt_pet_payload(payload))
|
|
self.assertEqual(
|
|
json.loads(raw_payload),
|
|
{"pet_name": "团团", "nfc_id": "E113B203", "action": "摇尾巴"},
|
|
)
|
|
self.assertEqual(qos, 1)
|
|
self.assertFalse(retain)
|
|
self.assertEqual(publication["delivery_status"], "published")
|
|
|
|
def test_pet_message_rejects_extra_fields_and_invalid_nfc_id(self) -> None:
|
|
with self.assertRaisesRegex(Exception, "exactly"):
|
|
self.transport.publish_pet_message(
|
|
{
|
|
"pet_name": "团团",
|
|
"nfc_id": "E113B203",
|
|
"action": "摇尾巴",
|
|
"command": "not-allowed",
|
|
}
|
|
)
|
|
with self.assertRaisesRegex(Exception, "uppercase hexadecimal"):
|
|
self.transport.publish_pet_message(
|
|
{"pet_name": "团团", "nfc_id": "not-a-uid", "action": "摇尾巴"}
|
|
)
|
|
|
|
def test_extension_toy_uses_fixed_topic_qos_one_and_never_retains(self) -> None:
|
|
# This application-level topic is contractually QoS 1 even if the
|
|
# vehicle CAN relay itself is configured with a different QoS.
|
|
self.transport.settings.qos = 0
|
|
payload = {
|
|
"extension_name": "闪烁灯",
|
|
"actions": [
|
|
{
|
|
"type": "infrared_emit",
|
|
"at_ms": 0,
|
|
"domain": "self_extension",
|
|
"capability": "infrared.emit",
|
|
"target_scope": "toy-light-001",
|
|
"operation": "emit",
|
|
"parameters": {"action": "打开", "slot": 2},
|
|
}
|
|
],
|
|
}
|
|
|
|
publication = self.transport.publish_extension_toy_message(payload)
|
|
|
|
topic, raw_payload, qos, retain = self.client.published[-1]
|
|
self.assertEqual(topic, MQTT_EXTENSION_TOY_TOPIC)
|
|
self.assertEqual(raw_payload, mqtt_extension_toy_payload(payload))
|
|
self.assertEqual(json.loads(raw_payload), payload)
|
|
self.assertEqual(qos, 1)
|
|
self.assertFalse(retain)
|
|
self.assertEqual(publication["qos"], 1)
|
|
self.assertEqual(publication["delivery_status"], "published")
|
|
|
|
def test_extension_toy_rejects_non_english_scope_and_extra_fields(self) -> None:
|
|
action = {
|
|
"type": "infrared_emit",
|
|
"at_ms": 0,
|
|
"domain": "self_extension",
|
|
"capability": "infrared.emit",
|
|
"target_scope": "闪烁灯",
|
|
"operation": "emit",
|
|
"parameters": {"action": "打开", "slot": 2},
|
|
}
|
|
with self.assertRaisesRegex(Exception, "English stable device id"):
|
|
self.transport.publish_extension_toy_message(
|
|
{"extension_name": "闪烁灯", "actions": [action]}
|
|
)
|
|
with self.assertRaisesRegex(Exception, "requires exactly"):
|
|
self.transport.publish_extension_toy_message(
|
|
{
|
|
"extension_name": "闪烁灯",
|
|
"actions": [{**action, "target_scope": "toy-light-001"}],
|
|
"execute": True,
|
|
}
|
|
)
|
|
|
|
def test_extension_control_is_dispatched_as_raw_bytes_without_blocking(self) -> None:
|
|
received: list[bytes] = []
|
|
self.transport.set_extension_control_handler(received.append)
|
|
payload = {
|
|
"domain": "self_extension",
|
|
"operation_count": 1,
|
|
"actions": [
|
|
{
|
|
"type": "infrared_emit",
|
|
"at_ms": 0,
|
|
"domain": "self_extension",
|
|
"capability": "infrared.emit",
|
|
"target_scope": "toy-light-001",
|
|
"operation": "emit",
|
|
"parameters": {"action": "打开", "slot": 2},
|
|
}
|
|
],
|
|
}
|
|
|
|
self.client.emit(MQTT_EXTENSION_CONTROL_TOPIC, payload)
|
|
|
|
self.assertEqual(len(received), 1)
|
|
self.assertEqual(json.loads(received[0]), payload)
|
|
status = self.transport.public_info()["extension_control"]
|
|
self.assertTrue(status["handler_configured"])
|
|
self.assertEqual(status["received_count"], 1)
|
|
self.assertEqual(status["dispatched_count"], 1)
|
|
self.assertEqual(status["rejected_count"], 0)
|
|
|
|
def test_extension_control_rejects_retained_and_duplicate_deliveries(self) -> None:
|
|
received: list[bytes] = []
|
|
rejections: list[tuple[str, str]] = []
|
|
self.transport.set_extension_control_handler(received.append)
|
|
self.transport.set_extension_control_rejection_handler(
|
|
lambda error, kind: rejections.append((error, kind))
|
|
)
|
|
payload = {
|
|
"domain": "self_extension",
|
|
"operation_count": 1,
|
|
"actions": [],
|
|
}
|
|
|
|
self.client.emit(MQTT_EXTENSION_CONTROL_TOPIC, payload, retain=True)
|
|
self.client.emit(MQTT_EXTENSION_CONTROL_TOPIC, payload, dup=True)
|
|
|
|
self.assertEqual(received, [])
|
|
status = self.transport.public_info()["extension_control"]
|
|
self.assertEqual(status["received_count"], 2)
|
|
self.assertEqual(status["dispatched_count"], 0)
|
|
self.assertEqual(status["rejected_count"], 2)
|
|
self.assertEqual(status["retained_count"], 1)
|
|
self.assertEqual(status["duplicate_count"], 1)
|
|
self.assertTrue(status["rejection_handler_configured"])
|
|
self.assertEqual(
|
|
rejections,
|
|
[
|
|
("retained extension/ctl messages are not executable", "retained"),
|
|
("duplicate extension/ctl delivery was ignored", "duplicate"),
|
|
],
|
|
)
|
|
|
|
def test_mode_scene_uses_fixed_topic_exact_catalog_card_and_never_retains(
|
|
self,
|
|
) -> None:
|
|
scene = {
|
|
"id": "mode-reading",
|
|
"label": "阅读模式",
|
|
"name": "阅读模式",
|
|
"background_time": "noon",
|
|
"description": "由自然语言生成的自定义模式。",
|
|
"icon": "bookmark-plus",
|
|
"aliases": ["阅读模式", "阅读"],
|
|
"origin": "user",
|
|
"deletable": True,
|
|
"behavior": "static",
|
|
"domains": ["climate"],
|
|
"operation_count": 1,
|
|
"scope": {"common": False, "vehicle_ids": ["denza_n7"]},
|
|
"profile_revision": "1234567890abcdef",
|
|
"actions": [
|
|
{
|
|
"type": "climate_temperature",
|
|
"at_ms": 0,
|
|
"domain": "climate",
|
|
"capability": "climate.set_temperature",
|
|
"target_scope": "passenger",
|
|
"operation": "set",
|
|
"parameters": {"temperature_c": 24},
|
|
"zone": "passenger",
|
|
"celsius": 24,
|
|
}
|
|
],
|
|
}
|
|
|
|
publication = self.transport.publish_mode_scene(scene)
|
|
|
|
topic, raw_payload, qos, retain = self.client.published[-1]
|
|
self.assertEqual(topic, MQTT_MODE_SCENE_TOPIC)
|
|
self.assertEqual(raw_payload, mqtt_mode_scene_payload(scene))
|
|
self.assertEqual(json.loads(raw_payload), scene)
|
|
self.assertNotIn("scenes", json.loads(raw_payload))
|
|
self.assertEqual(qos, 1)
|
|
self.assertFalse(retain)
|
|
self.assertEqual(publication["delivery_status"], "published")
|
|
|
|
def test_mode_scene_rejects_catalog_wrapper(self) -> None:
|
|
with self.assertRaisesRegex(Exception, "one scene only"):
|
|
self.transport.publish_mode_scene(
|
|
{
|
|
"service": "n7-scenes",
|
|
"scene_count": 1,
|
|
"scenes": [{"id": "mode-reading"}],
|
|
}
|
|
)
|
|
|
|
def test_n7_mode_scene_ingress_dispatches_raw_bytes_and_rejects_retained(
|
|
self,
|
|
) -> None:
|
|
received: list[bytes] = []
|
|
self.transport.set_mode_scene_ingress_handler(received.append)
|
|
scene = {
|
|
"id": "n7-reading",
|
|
"label": "N7 阅读模式",
|
|
"name": "N7 阅读模式",
|
|
"background_time": "noon",
|
|
"description": "车端同步模式。",
|
|
"icon": "book-open",
|
|
"aliases": ["N7 阅读模式"],
|
|
"origin": "user",
|
|
"deletable": True,
|
|
"behavior": "static",
|
|
"domains": ["interior_light"],
|
|
"operation_count": 1,
|
|
"scope": {"common": False, "vehicle_ids": ["denza_n7"]},
|
|
"profile_revision": "remote-revision",
|
|
"actions": [
|
|
{
|
|
"type": "interior_light_reading_power",
|
|
"at_ms": 0,
|
|
"domain": "interior_light",
|
|
"capability": "interior_light.set_reading_power",
|
|
"target_scope": "all",
|
|
"operation": "set",
|
|
"parameters": {"state": "on"},
|
|
}
|
|
],
|
|
}
|
|
|
|
self.client.emit(MQTT_MODE_SCENE_INGRESS_TOPIC, scene)
|
|
self.client.emit(MQTT_MODE_SCENE_INGRESS_TOPIC, scene, retain=True)
|
|
|
|
self.assertEqual(len(received), 1)
|
|
self.assertEqual(json.loads(received[0]), scene)
|
|
status = self.transport.public_info()["mode_scene_ingress"]
|
|
self.assertTrue(status["handler_configured"])
|
|
self.assertEqual(status["received_count"], 2)
|
|
self.assertEqual(status["dispatched_count"], 1)
|
|
self.assertEqual(status["rejected_count"], 1)
|
|
self.assertEqual(status["retained_count"], 1)
|
|
|
|
def test_n7_control_transfer_uses_fixed_qos_one_topic_and_never_retains(
|
|
self,
|
|
) -> None:
|
|
payload = {
|
|
"scene_id": "n7-reading",
|
|
"scene_name": "N7 阅读模式",
|
|
"request_id": "req-20260916000000-1234567890abcdef",
|
|
"operation_count": 1,
|
|
"actions": [
|
|
{
|
|
"type": "n7_media",
|
|
"at_ms": 0,
|
|
"domain": "media",
|
|
"capability": "media.play_favorite",
|
|
"target_scope": "vehicle",
|
|
"operation": "play",
|
|
"parameters": {"playlist": "focus"},
|
|
}
|
|
],
|
|
}
|
|
|
|
publication = self.transport.publish_n7_control_transfer(payload)
|
|
|
|
topic, raw_payload, qos, retain = self.client.published[-1]
|
|
self.assertEqual(topic, MQTT_N7_CONTROL_TRANSFER_TOPIC)
|
|
self.assertEqual(raw_payload, mqtt_n7_control_transfer_payload(payload))
|
|
self.assertEqual(json.loads(raw_payload), payload)
|
|
self.assertEqual(qos, 1)
|
|
self.assertFalse(retain)
|
|
self.assertEqual(publication["delivery_status"], "published")
|
|
|
|
def test_condition_frames_use_existing_decoder_and_snapshot_contract(self) -> None:
|
|
self.client.emit(
|
|
"n7/n7-headunit/can/state",
|
|
{
|
|
"frames": [
|
|
{"id": 0x342, "data": [0, 0, 0, 1, 0, 0, 0, 0]},
|
|
{"id": 0x4EB, "data": [7, 0, 0, 0, 0, 0, 0, 0]},
|
|
{"id": 0x407, "data": [0] * 8},
|
|
{"id": 0x494, "data": [0] * 8},
|
|
{"id": 0x4FA, "data": [0] * 8},
|
|
{"id": 0x555, "data": [255] * 8},
|
|
]
|
|
},
|
|
)
|
|
snapshot = self.transport.snapshot()
|
|
self.assertEqual(
|
|
snapshot.observed_ids,
|
|
frozenset({0x342, 0x4EB, 0x407, 0x494, 0x4FA}),
|
|
)
|
|
self.assertTrue(
|
|
all(
|
|
state["verification"]["transport"] == "wireless_mqtt"
|
|
for state in snapshot.feedback_states
|
|
)
|
|
)
|
|
self.assertTrue(
|
|
all(
|
|
state["source"] == "wireless_mqtt_ch1"
|
|
for state in snapshot.feedback_states
|
|
)
|
|
)
|
|
self.assertEqual(
|
|
self.client.subscriptions,
|
|
[
|
|
("n7/n7-headunit/can/state", 1),
|
|
(MQTT_EXTENSION_CONTROL_TOPIC, 1),
|
|
(MQTT_MODE_SCENE_INGRESS_TOPIC, 1),
|
|
],
|
|
)
|
|
|
|
def test_first_partial_live_vehicle_snapshot_does_not_wait_for_missing_ids(self) -> None:
|
|
self.client.emit(
|
|
"n7/n7-headunit/can/state",
|
|
{
|
|
"signals": [
|
|
{
|
|
"channel": "can1",
|
|
"canId": "0x388",
|
|
"frameType": "fd",
|
|
"brs": True,
|
|
"bytes": [0, 0, 0, 2, 1, 0, 89, 0],
|
|
}
|
|
]
|
|
},
|
|
)
|
|
|
|
started = time.perf_counter()
|
|
snapshot = self.transport.snapshot()
|
|
|
|
self.assertLess(time.perf_counter() - started, 0.05)
|
|
self.assertEqual(snapshot.observed_ids, frozenset({0x388}))
|
|
|
|
def test_native_can0_status_maps_to_passenger_frontend_state(self) -> None:
|
|
self.client.emit(
|
|
"n7/n7-headunit/can/state",
|
|
{
|
|
"schema": "n7-can-signals/v1",
|
|
"signals": [
|
|
{
|
|
"channel": "can0",
|
|
"canId": "0x302",
|
|
"frameType": "classic",
|
|
"bytes": [axis, 0x26, 0x08, 0, 0, value & 0xFF,
|
|
(value >> 8) & 0xFF, 0],
|
|
}
|
|
for axis, value in enumerate((1000, 2500, 5000, 9000))
|
|
] + [
|
|
{
|
|
"channel": "can0",
|
|
"canId": "0x4FA",
|
|
"frameType": "classic",
|
|
"bytes": [0x22, 0, 0, 0, 0, 0, 0, 0],
|
|
},
|
|
{
|
|
"channel": "can0",
|
|
"canId": "0x306",
|
|
"frameType": "classic",
|
|
"bytes": [1, 2, 3, 0x0F, 2, 0, 100, 0],
|
|
},
|
|
],
|
|
},
|
|
)
|
|
service = SeatDemoService(REPOSITORY_ROOT)
|
|
service._vehicle_mqtt_transport = self.transport
|
|
service.vehicle_mqtt_settings = SimpleNamespace(state_configured=True)
|
|
|
|
payload = service._synchronize_hardware_state_once(
|
|
"conv-native-can0-state",
|
|
control_transport="wireless_mqtt",
|
|
)
|
|
|
|
passenger = payload["actuator_state"]["passenger"]
|
|
self.assertEqual(
|
|
{
|
|
axis: value["position_pct"]
|
|
for axis, value in passenger["position"].items()
|
|
if axis in {
|
|
"longitudinal",
|
|
"cushion_tilt",
|
|
"backrest_recline",
|
|
"legrest_height",
|
|
}
|
|
},
|
|
{
|
|
"longitudinal": 10,
|
|
"cushion_tilt": 25,
|
|
"backrest_recline": 50,
|
|
"legrest_height": 90,
|
|
},
|
|
)
|
|
self.assertEqual(passenger["ventilation"]["level"], 1)
|
|
self.assertEqual(passenger["heating"]["level"], 3)
|
|
self.assertEqual(passenger["dynamic_mode"]["mode_id"], 1)
|
|
self.assertEqual(passenger["dynamic_mode"]["state"], "running")
|
|
self.assertTrue(
|
|
all(
|
|
item["feedback_source"] == "wireless_mqtt_ch0"
|
|
for item in passenger["position"].values()
|
|
if item.get("position_valid") is not None
|
|
)
|
|
)
|
|
|
|
def test_headunit_6a2_selector_state_maps_to_passenger_frontend_state(self) -> None:
|
|
values = {
|
|
1: bytes.fromhex("E0 01 E0 01"),
|
|
2: bytes.fromhex("00 00 75 01"),
|
|
3: bytes.fromhex("00 00 95 03"),
|
|
4: bytes.fromhex("BF 01 87 02"),
|
|
5: bytes.fromhex("11 00 00 00"),
|
|
6: bytes.fromhex("11 00 00 00"),
|
|
7: bytes.fromhex("11 00 02 00"),
|
|
8: bytes.fromhex("11 00 00 00"),
|
|
}
|
|
self.client.emit(
|
|
"n7/n7-headunit/can/state",
|
|
{
|
|
"schema": "n7-can-signals/v1",
|
|
"signals": [
|
|
{
|
|
"channel": "can0",
|
|
"canId": "0x6A2",
|
|
"selector": selector,
|
|
"frameType": "classic",
|
|
"bytes": [0x7C, 0x50 + selector, 0x60, 0, *value],
|
|
}
|
|
for selector, value in values.items()
|
|
],
|
|
},
|
|
)
|
|
service = SeatDemoService(REPOSITORY_ROOT)
|
|
service._vehicle_mqtt_transport = self.transport
|
|
service.vehicle_mqtt_settings = SimpleNamespace(state_configured=True)
|
|
|
|
payload = service._synchronize_hardware_state_once(
|
|
"conv-headunit-selector-can0-state",
|
|
control_transport="wireless_mqtt",
|
|
)
|
|
|
|
passenger = payload["actuator_state"]["passenger"]
|
|
self.assertEqual(
|
|
{
|
|
axis: passenger["position"][axis]["position_pct"]
|
|
for axis in (
|
|
"longitudinal",
|
|
"cushion_tilt",
|
|
"backrest_recline",
|
|
"legrest_height",
|
|
)
|
|
},
|
|
{
|
|
"longitudinal": 100,
|
|
"cushion_tilt": 0,
|
|
"backrest_recline": 0,
|
|
"legrest_height": 69,
|
|
},
|
|
)
|
|
self.assertEqual(payload["source"], "wireless_mqtt_ch0")
|
|
self.assertTrue(payload["complete"])
|
|
|
|
def test_vehicle_envelope_four_frames_decode_from_raw_can1_bytes(self) -> None:
|
|
self.client.emit(
|
|
"n7/n7-headunit/can/state",
|
|
{
|
|
"schema": "n7-can-signals/v1",
|
|
"sequence": 335,
|
|
"signals": [
|
|
{"channel": "can1", "canId": frame_id, "frameType": "fd",
|
|
"brs": True, "data": data}
|
|
for frame_id, data in (
|
|
("0x388", "00 00 00 02 01 00 59 00"),
|
|
("0x41A", "00 5C 85 08 20 42 C1 FD"),
|
|
("0x42E", "11 64 32 03 90 02 00 C6"),
|
|
("0x3FF", "5B 09 FF 0F 32 61 04 12"),
|
|
)
|
|
],
|
|
# This summary is deliberately inconsistent. Only the raw CAN
|
|
# bytes above are allowed to update authoritative UI state.
|
|
"can1Vehicle": {"parsed": {"speedKph": 123, "gear": 4}},
|
|
},
|
|
)
|
|
|
|
snapshot = self.transport.snapshot()
|
|
self.assertEqual(
|
|
snapshot.observed_ids,
|
|
frozenset({0x388, 0x3FF, 0x41A, 0x42E}),
|
|
)
|
|
by_id = {state["frame_id"]: state for state in snapshot.feedback_states}
|
|
self.assertEqual(
|
|
by_id["0x388"]["domain_state"]["window"]["positions"]["rear_left"]["position_pct"],
|
|
2,
|
|
)
|
|
self.assertEqual(
|
|
by_id["0x41A"]["domain_state"]["roof"]["shade_position_pct"],
|
|
92,
|
|
)
|
|
self.assertEqual(
|
|
by_id["0x42E"]["domain_state"]["ambient_light"]["front_color_index"],
|
|
100,
|
|
)
|
|
self.assertEqual(
|
|
by_id["0x3FF"]["domain_state"]["interior_light"]["ambient_illuminance_lux"],
|
|
510,
|
|
)
|
|
self.assertTrue(
|
|
all(state["source"] == "wireless_mqtt_ch1" for state in by_id.values())
|
|
)
|
|
|
|
def test_stale_safety_signal_cannot_enter_snapshot_or_executor_bus(self) -> None:
|
|
bus = self.transport.bus(channel=1)
|
|
self.client.emit(
|
|
"n7/n7-headunit/can/state",
|
|
{
|
|
"schema": "n7-can-signals/v1",
|
|
"signals": [
|
|
{
|
|
"channel": "can1",
|
|
"channelIndex": 1,
|
|
"canId": "0x342",
|
|
"frameType": "fd",
|
|
"brs": True,
|
|
"data": "00 00 00 01 00 00 00 00",
|
|
"timestamp": round((time.time() - 60) * 1_000),
|
|
}
|
|
],
|
|
},
|
|
)
|
|
snapshot = self.transport.snapshot()
|
|
info = self.transport.public_info()["state_sync"]
|
|
self.assertEqual(info["capture_sequence"], 1)
|
|
self.assertEqual(info["stale_frames"], 1)
|
|
self.assertEqual(info["observed_frame_ids"], [])
|
|
self.assertEqual(snapshot.observed_ids, frozenset())
|
|
self.assertIsNone(bus.receive(0, include_fd=True))
|
|
|
|
def test_stale_non_safety_can1_frames_are_display_only(self) -> None:
|
|
bus = self.transport.bus(channel=1)
|
|
cached_timestamp = round((time.time() - 60) * 1_000)
|
|
self.client.emit(
|
|
"n7/n7-headunit/can/state",
|
|
{
|
|
"schema": "n7-can-signals/v1",
|
|
"signals": [
|
|
{
|
|
"channel": "can1",
|
|
"canId": frame_id,
|
|
"frameType": "fd",
|
|
"brs": True,
|
|
"data": data,
|
|
"timestamp": cached_timestamp,
|
|
}
|
|
for frame_id, data in (
|
|
("0x388", "00 00 00 02 01 00 59 00"),
|
|
("0x41A", "00 5C 85 08 20 42 C1 FD"),
|
|
("0x42E", "11 64 32 03 90 02 00 C6"),
|
|
("0x3FF", "5B 09 FF 0F 32 61 04 12"),
|
|
)
|
|
],
|
|
},
|
|
)
|
|
|
|
snapshot = self.transport.snapshot()
|
|
self.assertEqual(
|
|
snapshot.observed_ids,
|
|
frozenset({0x388, 0x3FF, 0x41A, 0x42E}),
|
|
)
|
|
for state in snapshot.feedback_states:
|
|
verification = state["verification"]
|
|
self.assertFalse(verification["fresh"])
|
|
self.assertTrue(verification["display_only"])
|
|
self.assertFalse(verification["control_eligible"])
|
|
self.assertGreater(verification["cache_age_ms"], 50_000)
|
|
self.assertIsNone(bus.receive(0, include_fd=True))
|
|
|
|
def test_outlet_switch_feedback_is_available_to_the_can1_executor_bus(self) -> None:
|
|
bus = self.transport.bus(channel=1)
|
|
data = bytearray(64)
|
|
data[41:45] = bytes((0x26, 0x80, 0x80, 0x80))
|
|
self.client.emit(
|
|
"n7/n7-headunit/can/state",
|
|
{
|
|
"signals": [
|
|
{
|
|
"channel": "can1",
|
|
"canId": "0x4BB",
|
|
"frameType": "fd",
|
|
"brs": True,
|
|
"bytes": list(data),
|
|
}
|
|
]
|
|
},
|
|
)
|
|
|
|
feedback = bus.receive(0, include_fd=True)
|
|
self.assertIsNotNone(feedback)
|
|
self.assertEqual(feedback.arbitration_id, 0x4BB)
|
|
self.assertEqual(feedback.data[41], 0x26)
|
|
|
|
def test_bus_publishes_only_headunit_command_contract(self) -> None:
|
|
bus = self.transport.bus()
|
|
bus.set_debug_command("shade position 20")
|
|
sent = bus.send(
|
|
0x431,
|
|
bytes.fromhex("10 00 00 00 00 00 00 00"),
|
|
is_fd=True,
|
|
bitrate_switch=True,
|
|
)
|
|
self.assertEqual(sent.channel, 1)
|
|
self.assertEqual(len(self.client.published), 1)
|
|
topic, raw_payload, qos, retain = self.client.published[0]
|
|
payload = json.loads(raw_payload)
|
|
self.assertEqual(topic, "n7/n7-headunit/can/command")
|
|
self.assertEqual(qos, 1)
|
|
self.assertFalse(retain)
|
|
self.assertRegex(payload["requestId"], r"^cognitive-agent-[0-9a-f]{32}$")
|
|
self.assertEqual(payload["command"], "shade position 20")
|
|
self.assertEqual(payload["channel"], "can1")
|
|
self.assertEqual(payload["canId"], "0x431")
|
|
self.assertEqual(payload["frameType"], "fd")
|
|
self.assertTrue(payload["brs"])
|
|
self.assertFalse(payload["extended"])
|
|
self.assertFalse(payload["remote"])
|
|
self.assertEqual(payload["data"], "10 00 00 00 00 00 00 00")
|
|
self.assertEqual(payload["repeat"], 1)
|
|
self.assertEqual(payload["intervalMs"], 0)
|
|
self.assertFalse(payload["expectResponse"])
|
|
self.assertEqual(payload["responseTimeoutMs"], 100)
|
|
publication = bus.published_messages[0]
|
|
self.assertEqual(publication["topic"], topic)
|
|
self.assertEqual(publication["qos"], qos)
|
|
self.assertEqual(publication["retain"], retain)
|
|
self.assertEqual(publication["delivery_status"], "published")
|
|
self.assertEqual(publication["message_id"], 17)
|
|
self.assertEqual(publication["payload"], payload)
|
|
self.assertTrue(
|
|
{
|
|
"schema",
|
|
"arbitration_id",
|
|
"frame_id",
|
|
"data_hex",
|
|
"is_fd",
|
|
"bitrate_switch",
|
|
}.isdisjoint(payload)
|
|
)
|
|
bus.send(
|
|
0x431,
|
|
bytes.fromhex("10 00 00 00 00 00 00 00"),
|
|
is_fd=True,
|
|
bitrate_switch=True,
|
|
)
|
|
second_payload = json.loads(self.client.published[1][1])
|
|
self.assertNotEqual(payload["requestId"], second_payload["requestId"])
|
|
with self.assertRaisesRegex(Exception, "FD command requires BRS"):
|
|
bus.send(0x431, bytes(8), is_fd=True, bitrate_switch=False)
|
|
with self.assertRaisesRegex(Exception, "classic CAN command cannot enable BRS"):
|
|
bus.send(0x431, bytes(8), is_fd=False, bitrate_switch=True)
|
|
with self.assertRaisesRegex(Exception, "approved CAN1 control set"):
|
|
bus.send(0x555, bytes(8), is_fd=True)
|
|
with self.assertRaisesRegex(Exception, "at most 256"):
|
|
bus.set_debug_command("x" * 257)
|
|
self.assertEqual(len(self.client.published), 2)
|
|
self.assertEqual(len(bus.published_messages), 2)
|
|
|
|
def test_publish_returns_queued_without_waiting_for_broker_ack(self) -> None:
|
|
class QueuedPublishResult:
|
|
rc = 0
|
|
mid = 29
|
|
|
|
@staticmethod
|
|
def is_published() -> bool:
|
|
return False
|
|
|
|
@staticmethod
|
|
def wait_for_publish(timeout: float) -> bool:
|
|
raise AssertionError(
|
|
f"wait_for_publish({timeout}) must not be called"
|
|
)
|
|
|
|
with patch.object(
|
|
self.client,
|
|
"publish",
|
|
return_value=QueuedPublishResult(),
|
|
):
|
|
publication = self.transport.publish(
|
|
CanFrame(
|
|
arbitration_id=0x431,
|
|
data=bytes.fromhex("10 00 00 00 00 00 00 00"),
|
|
channel=1,
|
|
is_fd=True,
|
|
bitrate_switch=True,
|
|
),
|
|
command="shade position 20",
|
|
)
|
|
|
|
self.assertEqual(publication["delivery_status"], "queued")
|
|
self.assertEqual(publication["message_id"], 29)
|
|
|
|
def test_can1_whitelist_publishes_rear_lock_candidate_with_debug_command(self) -> None:
|
|
bus = self.transport.bus(channel=1)
|
|
bus.set_debug_command("door rear lock")
|
|
sent = bus.send(0x32B, bytes.fromhex("0A 40 00 00 00 00 00 00"))
|
|
|
|
self.assertEqual(sent.arbitration_id, 0x32B)
|
|
self.assertFalse(sent.is_fd)
|
|
topic, raw_payload, qos, retain = self.client.published[-1]
|
|
payload = json.loads(raw_payload)
|
|
self.assertEqual(topic, "n7/n7-headunit/can/command")
|
|
self.assertEqual(qos, 1)
|
|
self.assertFalse(retain)
|
|
self.assertEqual(payload["command"], "door rear lock")
|
|
self.assertEqual(payload["channel"], "can1")
|
|
self.assertEqual(payload["canId"], "0x32B")
|
|
self.assertEqual(payload["frameType"], "classic")
|
|
self.assertFalse(payload["brs"])
|
|
self.assertEqual(payload["data"], "0A 40 00 00 00 00 00 00")
|
|
|
|
def test_wireless_whitelisted_climate_plan_publishes_without_safety_or_feedback_frames(
|
|
self,
|
|
) -> None:
|
|
service = SeatDemoService(REPOSITORY_ROOT)
|
|
service._vehicle_mqtt_transport = self.transport
|
|
service.vehicle_mqtt_settings = SimpleNamespace(feedback_timeout_ms=1)
|
|
actions = [
|
|
{
|
|
"domain": "climate",
|
|
"capability": "climate.set_temperature",
|
|
"target": "driver",
|
|
"operation": "set",
|
|
"parameters": {"temperature_c": 24},
|
|
},
|
|
{
|
|
"domain": "climate",
|
|
"capability": "climate.set_fan_level",
|
|
"target": "vehicle",
|
|
"operation": "set",
|
|
"parameters": {"level": 2},
|
|
},
|
|
]
|
|
runtime = NativeCanRuntime(REPOSITORY_ROOT)
|
|
config = runtime.build_function_config(
|
|
"climate", actions, "req-wireless-climate", "local_rules"
|
|
)
|
|
plan = runtime.compile(config)
|
|
|
|
with (
|
|
patch(
|
|
"cognitive_agent.application.seat_service.observe_vehicle_safety",
|
|
side_effect=AssertionError(
|
|
"wireless whitelist publish must not require 0x342/0x4EB"
|
|
),
|
|
),
|
|
patch("cognitive_agent.application.native_can_runtime.time.sleep"),
|
|
):
|
|
reports = service._execute_orchestrated_reports(
|
|
{"native_climate": config},
|
|
{"native_climate": plan},
|
|
{"native_climate"},
|
|
{},
|
|
hardware_request=True,
|
|
simulated_hardware_request=False,
|
|
simulation_fault="normal",
|
|
control_transport="wireless_mqtt",
|
|
)
|
|
|
|
report = reports["native_climate"]
|
|
self.assertEqual(report["spec"]["status"], "succeeded")
|
|
self.assertEqual(
|
|
report["spec"]["vehicle_safety_preflight"]["policy"],
|
|
"mqtt_whitelist_publish",
|
|
)
|
|
self.assertEqual(len(self.client.published), 6)
|
|
payloads = [json.loads(item[1]) for item in self.client.published]
|
|
self.assertTrue(
|
|
all(
|
|
topic == "n7/n7-headunit/can/command"
|
|
for topic, _payload, _qos, _retain in self.client.published
|
|
)
|
|
)
|
|
self.assertTrue(all(payload["channel"] == "can1" for payload in payloads))
|
|
self.assertTrue(all(payload["canId"] == "0x1DE" for payload in payloads))
|
|
self.assertEqual(
|
|
[payload["command"] for payload in payloads],
|
|
["ac temp driver 24"] * 3 + ["ac fan 2"] * 3,
|
|
)
|
|
self.assertEqual(
|
|
[payload["data"] for payload in payloads],
|
|
[
|
|
"00 00 00 00 18 00 00 00",
|
|
"00 00 00 00 18 00 00 00",
|
|
"00 00 00 00 18 00 00 00",
|
|
"20 00 00 00 00 00 00 00",
|
|
"20 00 00 00 00 00 00 00",
|
|
"20 00 00 00 00 00 00 00",
|
|
],
|
|
)
|
|
for result in report["spec"]["results"]:
|
|
self.assertEqual(result["status"], "succeeded")
|
|
self.assertEqual(result["telemetry"]["delivery_status"], "published")
|
|
self.assertEqual(
|
|
result["telemetry"]["verification_status"],
|
|
"submitted_unverified",
|
|
)
|
|
self.assertFalse(result["telemetry"]["feedback_required"])
|
|
self.assertFalse(result["telemetry"]["physical_effect_confirmed"])
|
|
self.assertEqual(len(result["telemetry"]["mqtt_messages"]), 3)
|
|
self.assertTrue(
|
|
all(
|
|
message["topic"] == "n7/n7-headunit/can/command"
|
|
for message in result["telemetry"]["mqtt_messages"]
|
|
)
|
|
)
|
|
|
|
def test_wireless_individual_outlet_plan_publishes_observed_frame_triplet(
|
|
self,
|
|
) -> None:
|
|
service = SeatDemoService(REPOSITORY_ROOT)
|
|
service._vehicle_mqtt_transport = self.transport
|
|
service.vehicle_mqtt_settings = SimpleNamespace(feedback_timeout_ms=1)
|
|
runtime = NativeCanRuntime(REPOSITORY_ROOT)
|
|
config = runtime.build_function_config(
|
|
"outlet",
|
|
[
|
|
{
|
|
"domain": "outlet",
|
|
"capability": "outlet.set_switch_state",
|
|
"target": "passenger_right",
|
|
"operation": "set",
|
|
"parameters": {"state": "close"},
|
|
}
|
|
],
|
|
"req-wireless-outlet-switch",
|
|
"local_rules",
|
|
)
|
|
plan = runtime.compile(config)
|
|
|
|
with patch("cognitive_agent.application.native_can_runtime.time.sleep"):
|
|
reports = service._execute_orchestrated_reports(
|
|
{"native_outlet": config},
|
|
{"native_outlet": plan},
|
|
{"native_outlet"},
|
|
{},
|
|
hardware_request=True,
|
|
simulated_hardware_request=False,
|
|
simulation_fault="normal",
|
|
control_transport="wireless_mqtt",
|
|
)
|
|
|
|
self.assertEqual(reports["native_outlet"]["spec"]["status"], "succeeded")
|
|
payloads = [json.loads(item[1]) for item in self.client.published]
|
|
self.assertEqual(len(payloads), 3)
|
|
self.assertTrue(
|
|
all(payload["command"] == "outlet passenger-right close" for payload in payloads)
|
|
)
|
|
self.assertTrue(all(payload["channel"] == "can1" for payload in payloads))
|
|
self.assertTrue(all(payload["canId"] == "0x4EF" for payload in payloads))
|
|
self.assertTrue(all(payload["frameType"] == "fd" for payload in payloads))
|
|
self.assertTrue(all(payload["brs"] is True for payload in payloads))
|
|
self.assertTrue(
|
|
all(payload["data"] == "03 FF FF FF FF 08 00 00" for payload in payloads)
|
|
)
|
|
|
|
def test_wireless_hardware_can1_service_skips_local_hardware_connection(self) -> None:
|
|
service = SeatDemoService(REPOSITORY_ROOT)
|
|
service._vehicle_mqtt_transport = self.transport
|
|
service.vehicle_mqtt_settings = SimpleNamespace(
|
|
enabled=True,
|
|
configured=True,
|
|
feedback_timeout_ms=1,
|
|
authorize=lambda _token: None,
|
|
)
|
|
validated_events: list[dict[str, object]] = []
|
|
|
|
with tempfile.TemporaryDirectory() as temporary_directory:
|
|
service.artifact_root = Path(temporary_directory)
|
|
with (
|
|
patch.object(
|
|
service.hardware_execution,
|
|
"reserve",
|
|
side_effect=AssertionError(
|
|
"wireless CAN1 must not reserve the local physical bench"
|
|
),
|
|
),
|
|
patch("cognitive_agent.application.native_can_runtime.time.sleep"),
|
|
):
|
|
result = service.process(
|
|
"空调温度设置为24度",
|
|
provider="local",
|
|
conversation_id="conv-wireless-can1-no-local-hardware",
|
|
execution_mode="hardware",
|
|
control_transport="wireless_mqtt",
|
|
_on_commands_validated=validated_events.append,
|
|
)
|
|
|
|
self.assertEqual(result["status"], "succeeded")
|
|
self.assertEqual(list(result["command_plans"]), ["native_climate"])
|
|
self.assertEqual(len(validated_events), 1)
|
|
self.assertEqual(
|
|
list(validated_events[0]["command_plans"]),
|
|
["native_climate"],
|
|
)
|
|
payloads = [json.loads(item[1]) for item in self.client.published]
|
|
self.assertEqual(len(payloads), 3)
|
|
self.assertTrue(all(item["channel"] == "can1" for item in payloads))
|
|
self.assertTrue(all(item["command"] == "ac temp driver 24" for item in payloads))
|
|
|
|
def test_wireless_hardware_combines_can0_and_can1_without_local_profile(self) -> None:
|
|
service = SeatDemoService(REPOSITORY_ROOT)
|
|
service._vehicle_mqtt_transport = self.transport
|
|
service.vehicle_mqtt_settings = SimpleNamespace(
|
|
enabled=True,
|
|
configured=True,
|
|
feedback_timeout_ms=1,
|
|
authorize=lambda _token: None,
|
|
)
|
|
validated_events: list[dict[str, object]] = []
|
|
seat_execution: dict[str, object] = {}
|
|
|
|
def execute_seat(
|
|
plan: dict[str, object],
|
|
config: dict[str, object],
|
|
_state: dict[str, object],
|
|
) -> dict[str, object]:
|
|
seat_execution["plan"] = plan
|
|
seat_execution["config"] = config
|
|
operation = config["spec"]["operations"][0] # type: ignore[index]
|
|
request_id = config["metadata"]["request_id"] # type: ignore[index]
|
|
return {
|
|
"apiVersion": "cognitive-agent/v1alpha1",
|
|
"kind": "ExecutionReport",
|
|
"metadata": {
|
|
"id": f"report-{request_id}",
|
|
"name": "wireless CAN0 test report",
|
|
"version": "1.0.0",
|
|
"request_id": request_id,
|
|
"labels": {
|
|
"simulation": "false",
|
|
"seat_feedback": "wireless_mqtt_ch0",
|
|
},
|
|
},
|
|
"spec": {
|
|
"mode": "hil",
|
|
"status": "succeeded",
|
|
"started_at": "2026-08-25T00:00:00Z",
|
|
"finished_at": "2026-08-25T00:00:00Z",
|
|
"results": [
|
|
{
|
|
"ref": operation["id"],
|
|
"domain": "seat",
|
|
"status": "succeeded",
|
|
"message": "wireless CAN0 command published",
|
|
"telemetry": {
|
|
"effective_position_pct": 35,
|
|
"position_valid": True,
|
|
"simulated": False,
|
|
},
|
|
}
|
|
],
|
|
},
|
|
}
|
|
|
|
with tempfile.TemporaryDirectory() as temporary_directory:
|
|
service.artifact_root = Path(temporary_directory)
|
|
with (
|
|
patch.object(
|
|
type(service.hardware_settings),
|
|
"validate_function_config",
|
|
side_effect=AssertionError(
|
|
"wireless mixed CAN0/CAN1 must not check combined profile"
|
|
),
|
|
),
|
|
patch.object(
|
|
service.hardware_execution,
|
|
"reserve",
|
|
side_effect=AssertionError(
|
|
"wireless mixed CAN0/CAN1 must not reserve local hardware"
|
|
),
|
|
),
|
|
patch.object(
|
|
service,
|
|
"_new_hardware_executor",
|
|
return_value=SimpleNamespace(execute=execute_seat),
|
|
) as executor_factory,
|
|
patch("cognitive_agent.application.native_can_runtime.time.sleep"),
|
|
):
|
|
result = service.process(
|
|
"把副驾座椅水平位置调到35%，空调温度设置为24度",
|
|
provider="local",
|
|
conversation_id="conv-wireless-can0-can1-no-local-hardware",
|
|
execution_mode="hardware",
|
|
control_transport="wireless_mqtt",
|
|
_on_commands_validated=validated_events.append,
|
|
)
|
|
|
|
self.assertEqual(result["status"], "succeeded")
|
|
self.assertEqual(
|
|
set(result["command_plans"]),
|
|
{"seat", "native_climate"},
|
|
)
|
|
self.assertEqual(len(validated_events), 1)
|
|
self.assertEqual(
|
|
set(validated_events[0]["command_plans"]),
|
|
{"seat", "native_climate"},
|
|
)
|
|
self.assertEqual(
|
|
seat_execution["plan"]["spec"]["commands"][0]["message"], # type: ignore[index]
|
|
"T2G_Seat_Move_Command_300",
|
|
)
|
|
self.assertTrue(executor_factory.call_args.kwargs["remote_transport"])
|
|
payloads = [json.loads(item[1]) for item in self.client.published]
|
|
self.assertEqual(len(payloads), 3)
|
|
self.assertTrue(all(item["channel"] == "can1" for item in payloads))
|
|
self.assertTrue(all(item["command"] == "ac temp driver 24" for item in payloads))
|
|
|
|
def test_wireless_fixed_controls_and_shade_preset_never_wait_for_state(
|
|
self,
|
|
) -> None:
|
|
service = SeatDemoService(REPOSITORY_ROOT)
|
|
service._vehicle_mqtt_transport = self.transport
|
|
service.vehicle_mqtt_settings = SimpleNamespace(feedback_timeout_ms=1_500)
|
|
runtime = NativeCanRuntime(REPOSITORY_ROOT)
|
|
actions_by_domain = {
|
|
"climate": [
|
|
{
|
|
"domain": "climate",
|
|
"capability": "climate.set_temperature",
|
|
"target": "driver",
|
|
"operation": "set",
|
|
"parameters": {"temperature_c": 24},
|
|
},
|
|
{
|
|
"domain": "climate",
|
|
"capability": "climate.set_fan_level",
|
|
"target": "vehicle",
|
|
"operation": "set",
|
|
"parameters": {"level": 2},
|
|
},
|
|
],
|
|
"roof": [
|
|
{
|
|
"domain": "roof",
|
|
"capability": "roof.set_shade_motion",
|
|
"target": "vehicle",
|
|
"operation": "set",
|
|
"parameters": {"motion": "open"},
|
|
}
|
|
],
|
|
"ambient_light": [
|
|
{
|
|
"domain": "ambient_light",
|
|
"capability": "ambient_light.set_power",
|
|
"target": "vehicle",
|
|
"operation": "set",
|
|
"parameters": {"state": "on"},
|
|
}
|
|
],
|
|
}
|
|
configs = {
|
|
f"native_{domain}": runtime.build_function_config(
|
|
domain,
|
|
actions,
|
|
f"req-wireless-no-wait-{domain}",
|
|
"local_rules",
|
|
)
|
|
for domain, actions in actions_by_domain.items()
|
|
}
|
|
plans = {key: runtime.compile(config) for key, config in configs.items()}
|
|
|
|
with (
|
|
patch.object(
|
|
self.transport,
|
|
"_receive_after",
|
|
side_effect=AssertionError(
|
|
"wireless fixed commands must not wait on can/state"
|
|
),
|
|
) as receive_after,
|
|
patch("cognitive_agent.application.native_can_runtime.time.sleep"),
|
|
):
|
|
reports = service._execute_orchestrated_reports(
|
|
configs,
|
|
plans,
|
|
set(configs),
|
|
{},
|
|
hardware_request=True,
|
|
simulated_hardware_request=False,
|
|
simulation_fault="normal",
|
|
control_transport="wireless_mqtt",
|
|
)
|
|
|
|
receive_after.assert_not_called()
|
|
self.assertEqual(len(self.client.published), 10)
|
|
self.assertTrue(
|
|
all(
|
|
report["spec"]["status"] == "succeeded"
|
|
for report in reports.values()
|
|
)
|
|
)
|
|
self.assertTrue(
|
|
all(
|
|
json.loads(payload)["expectResponse"] is False
|
|
for _topic, payload, _qos, _retain in self.client.published
|
|
)
|
|
)
|
|
|
|
def test_model_added_nonwhitelisted_action_does_not_block_whitelisted_publish(
|
|
self,
|
|
) -> None:
|
|
service = SeatDemoService(REPOSITORY_ROOT)
|
|
service._vehicle_mqtt_transport = self.transport
|
|
service.vehicle_mqtt_settings = SimpleNamespace(
|
|
enabled=True,
|
|
configured=True,
|
|
feedback_timeout_ms=1,
|
|
authorize=lambda _token: None,
|
|
)
|
|
model_output = {
|
|
"detected_domains": ["climate", "ambient_light"],
|
|
"requests": [
|
|
{
|
|
"domain": "ambient_light",
|
|
"capability": "ambient_light.set_mode",
|
|
"target_scope": "vehicle",
|
|
"operation": "set",
|
|
"parameters": [{"name": "mode", "value": "cinema"}],
|
|
"reason": "模型根据观影状态补充了氛围灯模式。",
|
|
},
|
|
{
|
|
"domain": "climate",
|
|
"capability": "climate.set_temperature",
|
|
"target_scope": "driver",
|
|
"operation": "set",
|
|
"parameters": [{"name": "temperature_c", "value": 24}],
|
|
"reason": "用户明确指定温度。",
|
|
},
|
|
{
|
|
"domain": "climate",
|
|
"capability": "climate.set_fan_level",
|
|
"target_scope": "vehicle",
|
|
"operation": "set",
|
|
"parameters": [{"name": "level", "value": 2}],
|
|
"reason": "用户明确指定风量。",
|
|
},
|
|
],
|
|
"ambiguities": [],
|
|
"decision_summary": ["执行用户明确指定的观影空调设置。"],
|
|
"confidence": 0.96,
|
|
}
|
|
|
|
with tempfile.TemporaryDirectory() as temporary_directory:
|
|
service.artifact_root = Path(temporary_directory)
|
|
service.openai.api_key = "test-key"
|
|
with (
|
|
patch.object(service.openai, "interpret", return_value=model_output),
|
|
patch("cognitive_agent.application.native_can_runtime.time.sleep"),
|
|
):
|
|
result = service.process(
|
|
"进入观影状态：空调24度、风量2档，",
|
|
provider="openai",
|
|
conversation_id="conv-wireless-movie-climate",
|
|
execution_mode="hardware",
|
|
control_transport="wireless_mqtt",
|
|
)
|
|
|
|
self.assertEqual(result["status"], "partially_succeeded")
|
|
self.assertEqual(
|
|
[item["capability"] for item in result["deferred_requests"]],
|
|
["ambient_light.set_mode"],
|
|
)
|
|
self.assertEqual(
|
|
result["deferred_requests"][0]["rejection_code"],
|
|
"WIRELESS_MQTT_CONTROL_NOT_WHITELISTED",
|
|
)
|
|
self.assertEqual(len(self.client.published), 6)
|
|
self.assertTrue(
|
|
all(
|
|
topic == "n7/n7-headunit/can/command"
|
|
and json.loads(payload)["canId"] == "0x1DE"
|
|
for topic, payload, _qos, _retain in self.client.published
|
|
)
|
|
)
|
|
|
|
def test_can0_bus_publishes_only_classic_passenger_seat_control(self) -> None:
|
|
bus = self.transport.bus(channel=0)
|
|
sent = bus.send(0x300, bytes.fromhex("01 10 02 00 88 13 30 75"))
|
|
payload = json.loads(self.client.published[-1][1])
|
|
self.assertEqual(sent.channel, 0)
|
|
self.assertEqual(payload["channel"], "can0")
|
|
self.assertEqual(payload["canId"], "0x300")
|
|
self.assertEqual(payload["data"], "01 10 02 00 88 13 30 75")
|
|
self.assertEqual(payload["command"], "")
|
|
self.assertEqual(payload["frameType"], "classic")
|
|
self.assertFalse(payload["brs"])
|
|
bus.set_debug_command("seat passenger heat low")
|
|
bus.send(0x3CE, bytes.fromhex("10 00 00 00 00 00 00 00"))
|
|
bus.send(0x29E, bytes.fromhex("16 00 10 00 00 00 00 00"))
|
|
thermal_payloads = [
|
|
json.loads(item[1]) for item in self.client.published[-2:]
|
|
]
|
|
self.assertEqual(
|
|
[item["canId"] for item in thermal_payloads],
|
|
["0x3CE", "0x29E"],
|
|
)
|
|
self.assertTrue(
|
|
all(item["command"] == "seat passenger heat low" for item in thermal_payloads)
|
|
)
|
|
with self.assertRaisesRegex(Exception, "0x29E thermal control payload"):
|
|
bus.send(0x29E, bytes(8))
|
|
with self.assertRaisesRegex(Exception, "approved CAN0 control set"):
|
|
bus.send(0x431, bytes(8))
|
|
with self.assertRaisesRegex(Exception, "requires classic CAN"):
|
|
bus.send(0x300, bytes.fromhex("01 10 02 00 88 13 30 75"),
|
|
is_fd=True, bitrate_switch=True)
|
|
with self.assertRaisesRegex(Exception, "approved CAN0 control set"):
|
|
bus.send(0x6A0, bytes.fromhex("C7 01 64 00 F4 01 00 00"))
|
|
with self.assertRaisesRegex(Exception, "protocol version 1"):
|
|
bus.send(0x300, bytes(8))
|
|
with self.assertRaisesRegex(Exception, "outside the semantic control whitelist"):
|
|
bus.send(0x300, bytes.fromhex("0A 10 03 00 5A A5 78 00"))
|
|
with self.assertRaisesRegex(Exception, "absolute move payload"):
|
|
bus.send(0x300, bytes.fromhex("01 10 04 00 11 27 30 75"))
|
|
with self.assertRaisesRegex(Exception, "dynamic-mode payload"):
|
|
bus.send(0x300, bytes.fromhex("0E 10 05 00 03 00 00 00"))
|
|
with self.assertRaisesRegex(Exception, "approved channel policy"):
|
|
self.transport.publish(
|
|
CanFrame(arbitration_id=0x431, data=bytes(8), channel=0)
|
|
)
|
|
|
|
def test_can0_native_motion_whitelist_accepts_only_bounded_runtime_opcodes(
|
|
self,
|
|
) -> None:
|
|
bus = self.transport.bus(channel=0)
|
|
valid_payloads = (
|
|
"01 10 01 00 88 13 30 75", # absolute 50%, 30 s
|
|
"02 11 02 00 18 FC 30 75", # native relative -10%, 30 s
|
|
"03 16 03 00 F4 01 84 03", # motor 2 far, 500 ms, 90%
|
|
"08 12 04 00 00 00 00 00", # stop motor 2
|
|
"09 10 05 00 00 00 00 00", # stop all
|
|
"0B 13 06 00 52 4B 78 00", # motor 3 known-range home
|
|
"0E 10 07 00 01 00 00 00", # rocking-horse
|
|
"0F 10 08 00 00 00 00 00", # stop dynamic mode
|
|
)
|
|
|
|
for payload in valid_payloads:
|
|
with self.subTest(payload=payload):
|
|
sent = bus.send(0x300, bytes.fromhex(payload))
|
|
self.assertEqual(sent.arbitration_id, 0x300)
|
|
self.assertFalse(sent.is_fd)
|
|
|
|
with self.assertRaisesRegex(Exception, "requires a runtime sequence"):
|
|
bus.send(0x300, bytes.fromhex("01 10 00 00 88 13 30 75"))
|
|
with self.assertRaisesRegex(Exception, "outside the semantic control whitelist"):
|
|
bus.send(0x300, bytes.fromhex("0D 10 09 00 3C C3 00 00"))
|
|
|
|
def test_bus_receive_views_are_isolated_by_channel(self) -> None:
|
|
can0 = self.transport.bus(channel=0)
|
|
can1 = self.transport.bus(channel=1)
|
|
self.client.emit(
|
|
"n7/n7-headunit/can/state",
|
|
{
|
|
"signals": [
|
|
{"channel": "can0", "canId": "0x6A0", "bytes":
|
|
[0xC7, 0x21, 0x60, 1, 0, 0, 0, 0]},
|
|
{"channel": "can0", "canId": "0x6A2", "bytes":
|
|
[0x7C, 0x21, 0x60, 0, 0xE0, 0x01, 0xE0, 0x01]},
|
|
{
|
|
"channel": "can1",
|
|
"canId": "0x342",
|
|
"frameType": "fd",
|
|
"bytes": [0] * 8,
|
|
},
|
|
]
|
|
},
|
|
)
|
|
self.assertEqual(can0.receive(0).arbitration_id, 0x6A2)
|
|
self.assertEqual(can0.receive(0).arbitration_id, 0x302)
|
|
self.assertEqual(can1.receive(0, include_fd=True).arbitration_id, 0x342)
|
|
self.assertIsNone(can0.receive(0))
|
|
self.assertIsNone(can1.receive(0, include_fd=True))
|
|
|
|
def test_can0_status_bus_can_replay_latest_translated_ota_positions(self) -> None:
|
|
signals = []
|
|
for selector, value in {
|
|
1: (480 << 16) | 407,
|
|
2: (373 << 16) | 266,
|
|
3: (917 << 16) | 451,
|
|
4: (647 << 16) | 575,
|
|
}.items():
|
|
sequence = 0x40 + selector
|
|
signals.extend(
|
|
(
|
|
{"channel": "can0", "canId": "0x6A0", "bytes":
|
|
[0xC7, sequence, 0x60, selector, 0, 0, 0, 0]},
|
|
{"channel": "can0", "canId": "0x6A2", "bytes":
|
|
[0x7C, sequence, 0x60, 0, *value.to_bytes(4, "little")]},
|
|
)
|
|
)
|
|
self.client.emit("n7/n7-headunit/can/state", {"signals": signals})
|
|
|
|
bus = self.transport.bus(channel=0, replay_latest=True)
|
|
replayed = [bus.receive(0) for _ in range(4)]
|
|
|
|
self.assertEqual(
|
|
[frame.data[0] for frame in replayed if frame is not None],
|
|
[0, 1, 2, 3],
|
|
)
|
|
self.assertEqual(bus.receive(0).arbitration_id, 0x6A2)
|
|
self.assertIsNone(bus.receive(0))
|
|
self.assertEqual(
|
|
self.transport.public_info()["state_sync"]["observed_frame_ids_by_channel"]["0"],
|
|
["0x302", "0x6A2"],
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|