vee
744 lines
31 KiB
Python
744 lines
31 KiB
Python
from __future__ import annotations
|
|
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
from cognitive_agent.application.native_can_feedback import (
|
|
decode_feedback,
|
|
feedback_matches,
|
|
)
|
|
from cognitive_agent.application.native_can_intent import (
|
|
_parse_exact_command,
|
|
merge_native_can_intent,
|
|
)
|
|
from cognitive_agent.application.native_can_runtime import (
|
|
NativeCanCompileError,
|
|
NativeCanExecutor,
|
|
NativeCanRuntime,
|
|
)
|
|
from cognitive_agent.application.seat_service import REPOSITORY_ROOT
|
|
from cognitive_agent.infrastructure.can.gateway import CanFrame
|
|
from cognitive_agent.runtime.contracts import ContractValidator
|
|
|
|
|
|
class _FakeBus:
|
|
def __init__(
|
|
self,
|
|
received: list[CanFrame] | None = None,
|
|
*,
|
|
fail_send_calls: set[int] | None = None,
|
|
) -> None:
|
|
self.received = list(received or [])
|
|
self.fail_send_calls = set(fail_send_calls or set())
|
|
self.send_attempts: list[tuple[int, bytes, bool, bool]] = []
|
|
self.sent: list[CanFrame] = []
|
|
self.events: list[str] = []
|
|
self.debug_commands: list[str] = []
|
|
self.closed = False
|
|
|
|
def set_debug_command(self, command: str) -> None:
|
|
self.debug_commands.append(command)
|
|
|
|
def send(
|
|
self,
|
|
arbitration_id: int,
|
|
data: bytes,
|
|
*,
|
|
is_fd: bool = False,
|
|
bitrate_switch: bool = False,
|
|
) -> CanFrame:
|
|
self.events.append("send")
|
|
self.send_attempts.append(
|
|
(arbitration_id, bytes(data), is_fd, bitrate_switch)
|
|
)
|
|
call_number = len(self.send_attempts)
|
|
if call_number in self.fail_send_calls:
|
|
raise RuntimeError(f"injected send failure {call_number}")
|
|
frame = CanFrame(
|
|
arbitration_id,
|
|
bytes(data),
|
|
channel=1,
|
|
is_fd=is_fd,
|
|
bitrate_switch=bitrate_switch,
|
|
)
|
|
self.sent.append(frame)
|
|
return frame
|
|
|
|
def receive(self, timeout_ms: int, *, include_fd: bool = False) -> CanFrame | None:
|
|
self.events.append("receive")
|
|
return self.received.pop(0) if self.received else None
|
|
|
|
def close(self) -> None:
|
|
self.closed = True
|
|
|
|
|
|
class VerifiedBodyCanTests(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.runtime = NativeCanRuntime(REPOSITORY_ROOT)
|
|
|
|
def compile_action(
|
|
self, domain: str, capability: str, target: str, parameters: dict
|
|
) -> dict:
|
|
config = self.runtime.build_function_config(
|
|
domain,
|
|
[
|
|
{
|
|
"domain": domain,
|
|
"capability": capability,
|
|
"target": target,
|
|
"operation": "set",
|
|
"parameters": parameters,
|
|
}
|
|
],
|
|
"verified-body-can",
|
|
"local_rules",
|
|
)
|
|
return self.runtime.compile(config)["spec"]["commands"][0]
|
|
|
|
@staticmethod
|
|
def shade_feedback(position: int, motion: int = 1) -> CanFrame:
|
|
data = bytearray(8)
|
|
data[1] = position
|
|
data[3] = motion << 3
|
|
return CanFrame(0x41A, bytes(data), channel=1)
|
|
|
|
def assert_fd_brs_triplet(self, command: dict, frame_id: str) -> None:
|
|
self.assertEqual(command["signals"]["frame_id"], frame_id)
|
|
self.assertTrue(command["signals"]["is_fd"])
|
|
self.assertTrue(command["signals"]["bitrate_switch"])
|
|
self.assertEqual(command["repeat"], {"count": 3, "period_ms": 100})
|
|
|
|
def test_indoor_and_all_reading_lights_use_only_verified_4f5(self) -> None:
|
|
cases = (
|
|
("interior_light.set_indoor_power", "vehicle", "on", "00 00 03 FF FF 00 08 00"),
|
|
("interior_light.set_indoor_power", "vehicle", "off", "00 00 03 FF FF 00 04 00"),
|
|
("interior_light.set_reading_power", "all", "on", "00 00 03 FF FF 00 08 00"),
|
|
("interior_light.set_reading_power", "all", "off", "00 00 03 FF FF 00 04 00"),
|
|
)
|
|
for capability, target, state, expected_hex in cases:
|
|
with self.subTest(capability=capability, state=state):
|
|
command = self.compile_action(
|
|
"interior_light", capability, target, {"state": state}
|
|
)
|
|
self.assert_fd_brs_triplet(command, "0x4F5")
|
|
self.assertEqual(command["signals"]["data_hex"], expected_hex)
|
|
self.assertEqual(command["signals"]["feedback_ids"], "0x42E")
|
|
self.assertEqual(
|
|
command["signals"]["expected_feedback"],
|
|
f"indoor_light:{state}",
|
|
)
|
|
|
|
def test_unverified_light_controls_are_not_hardware_capabilities(self) -> None:
|
|
rejected = (
|
|
("interior_light.set_wall_power", "vehicle", {"state": "on"}),
|
|
("ambient_light.set_mode", "vehicle", {"mode": "cinema"}),
|
|
(
|
|
"ambient_light.set_illumination",
|
|
"vehicle",
|
|
{"color_hex": "#203040", "brightness_pct": 40},
|
|
),
|
|
)
|
|
for capability, target, parameters in rejected:
|
|
action = {"capability": capability, "target": target, "parameters": parameters}
|
|
with self.subTest(capability=capability):
|
|
self.assertFalse(self.runtime.supports(action))
|
|
with self.assertRaises(NativeCanCompileError):
|
|
self.runtime.build_function_config(
|
|
capability.split(".", 1)[0],
|
|
[action],
|
|
"reject-unverified-light",
|
|
"local_rules",
|
|
)
|
|
|
|
zoned = {
|
|
"capability": "interior_light.set_reading_power",
|
|
"target": "front_left",
|
|
"parameters": {"state": "on"},
|
|
}
|
|
self.assertFalse(self.runtime.supports(zoned))
|
|
with self.assertRaises(NativeCanCompileError):
|
|
config = self.runtime.build_function_config(
|
|
"interior_light", [zoned], "reject-zoned-reading", "local_rules"
|
|
)
|
|
self.runtime.compile(config)
|
|
|
|
def test_compressor_catalog_action_cannot_reuse_hvac_power_frame(self) -> None:
|
|
action = {
|
|
"capability": "climate.set_ac",
|
|
"target": "vehicle",
|
|
"parameters": {"state": "on"},
|
|
}
|
|
|
|
self.assertFalse(self.runtime.supports(action))
|
|
self.assertNotIn("climate.set_ac", self.runtime.supported_capabilities)
|
|
with self.assertRaisesRegex(NativeCanCompileError, "no HIL-verified physical"):
|
|
self.runtime._frame_specs(
|
|
{
|
|
"capability": "climate.set_ac",
|
|
"target": {"zone": "vehicle"},
|
|
"parameters": {"state": "on"},
|
|
}
|
|
)
|
|
with self.assertRaises(NativeCanCompileError):
|
|
self.runtime.build_function_config(
|
|
"climate", [action], "reject-unverified-compressor", "local_rules"
|
|
)
|
|
|
|
def test_exact_parser_exposes_only_verified_light_vocabulary(self) -> None:
|
|
self.assertEqual(_parse_exact_command("light wall on"), [])
|
|
self.assertEqual(_parse_exact_command("light reading front-left on"), [])
|
|
self.assertEqual(_parse_exact_command("light ambient scene cinema"), [])
|
|
self.assertEqual(_parse_exact_command("light ambient rgb 32 64 96 40"), [])
|
|
|
|
color = _parse_exact_command("light ambient color 70")
|
|
self.assertEqual(color[0]["capability"], "ambient_light.set_color_index")
|
|
self.assertEqual(color[0]["parameters"], {"color_index": 70})
|
|
brightness = _parse_exact_command("light ambient brightness 5")
|
|
self.assertEqual(
|
|
brightness[0]["capability"], "ambient_light.set_brightness_level"
|
|
)
|
|
self.assertEqual(brightness[0]["parameters"], {"brightness_level": 5})
|
|
|
|
def test_all_cabin_lights_off_expands_to_verified_light_controls(self) -> None:
|
|
parsed = merge_native_can_intent(
|
|
{
|
|
"actions": [],
|
|
"domain_actions": [],
|
|
"native_can_actions": [],
|
|
"deferred_requests": [],
|
|
"model_requests": [],
|
|
"detected_domains": [],
|
|
"ambiguities": [],
|
|
"decision_summary": [],
|
|
},
|
|
"灯光全部关闭",
|
|
)
|
|
|
|
actions = parsed["native_can_actions"]
|
|
self.assertEqual(
|
|
[
|
|
(
|
|
action["capability"],
|
|
action["target"],
|
|
action["parameters"],
|
|
)
|
|
for action in actions
|
|
],
|
|
[
|
|
(
|
|
"interior_light.set_indoor_power",
|
|
"vehicle",
|
|
{"state": "off"},
|
|
),
|
|
(
|
|
"interior_light.set_reading_power",
|
|
"all",
|
|
{"state": "off"},
|
|
),
|
|
(
|
|
"ambient_light.set_power",
|
|
"vehicle",
|
|
{"state": "off"},
|
|
),
|
|
],
|
|
)
|
|
self.assertTrue(all(self.runtime.supports(action) for action in actions))
|
|
self.assertEqual(
|
|
parsed["detected_domains"], ["interior_light", "ambient_light"]
|
|
)
|
|
|
|
def test_fully_exact_ac_power_is_the_only_executable_plan(self) -> None:
|
|
for requested_state in ("on", "off"):
|
|
opposite_state = "off" if requested_state == "on" else "on"
|
|
for model_capability, model_state in (
|
|
("climate.set_ac", requested_state),
|
|
("climate.set_ac", opposite_state),
|
|
("climate.set_power", opposite_state),
|
|
):
|
|
with self.subTest(
|
|
requested_state=requested_state,
|
|
model_capability=model_capability,
|
|
model_state=model_state,
|
|
):
|
|
model_request = {
|
|
"domain": "climate",
|
|
"capability": model_capability,
|
|
"target_scope": "vehicle",
|
|
"operation": "set",
|
|
"parameters": [{"name": "state", "value": model_state}],
|
|
}
|
|
routed_action = {
|
|
"domain": "climate",
|
|
"capability": model_capability,
|
|
"target": "vehicle",
|
|
"operation": "set",
|
|
"parameters": {"state": model_state},
|
|
}
|
|
parsed = {
|
|
"detected_domains": ["climate", "window", "seat"],
|
|
"actions": [
|
|
{
|
|
"target": "passenger",
|
|
"function": "ventilation",
|
|
"state": "on",
|
|
}
|
|
],
|
|
"model_requests": [model_request],
|
|
"domain_actions": [
|
|
routed_action,
|
|
{
|
|
"domain": "window",
|
|
"capability": "window.set_motion",
|
|
"target": "front_left",
|
|
"operation": "set",
|
|
"parameters": {"motion": "open"},
|
|
},
|
|
],
|
|
"native_can_actions": [routed_action],
|
|
"deferred_requests": [routed_action],
|
|
"preset_actions": [{"target": "passenger"}],
|
|
"preset_events": [{"action": "apply"}],
|
|
"ambiguities": [
|
|
{"field": "requests[0].target", "question": "hallucinated"}
|
|
],
|
|
}
|
|
merged = merge_native_can_intent(
|
|
parsed, f"ac {requested_state}"
|
|
)
|
|
|
|
self.assertEqual(merged["actions"], [])
|
|
self.assertEqual(merged["domain_actions"], [])
|
|
self.assertEqual(merged["preset_actions"], [])
|
|
self.assertEqual(merged["preset_events"], [])
|
|
self.assertEqual(merged["ambiguities"], [])
|
|
self.assertEqual(merged["detected_domains"], ["climate"])
|
|
self.assertEqual(len(merged["model_requests"]), 1)
|
|
self.assertEqual(len(merged["deferred_requests"]), 1)
|
|
self.assertEqual(len(merged["native_can_actions"]), 1)
|
|
for key in (
|
|
"model_requests",
|
|
"deferred_requests",
|
|
"native_can_actions",
|
|
):
|
|
self.assertEqual(
|
|
merged[key][0]["capability"], "climate.set_power"
|
|
)
|
|
parameters = merged[key][0]["parameters"]
|
|
if isinstance(parameters, list):
|
|
parameters = {
|
|
item["name"]: item["value"] for item in parameters
|
|
}
|
|
self.assertEqual(parameters, {"state": requested_state})
|
|
|
|
natural_model_action = {
|
|
"domain": "climate",
|
|
"capability": "climate.set_ac",
|
|
"target": "vehicle",
|
|
"operation": "set",
|
|
"parameters": {"state": "on"},
|
|
}
|
|
natural = merge_native_can_intent(
|
|
{
|
|
"model_requests": [natural_model_action],
|
|
"domain_actions": [natural_model_action],
|
|
"native_can_actions": [],
|
|
"deferred_requests": [],
|
|
},
|
|
"打开空调并关闭遮阳帘",
|
|
)
|
|
self.assertIn(natural_model_action, natural["domain_actions"])
|
|
self.assertGreaterEqual(len(natural["native_can_actions"]), 2)
|
|
|
|
mixed_shade_action = {
|
|
"domain": "roof",
|
|
"capability": "roof.set_shade_motion",
|
|
"target": "vehicle",
|
|
"operation": "set",
|
|
"parameters": {"motion": "close"},
|
|
}
|
|
mixed = merge_native_can_intent(
|
|
{
|
|
"model_requests": [],
|
|
"domain_actions": [mixed_shade_action],
|
|
"native_can_actions": [],
|
|
"deferred_requests": [],
|
|
},
|
|
"ac on，关闭遮阳帘",
|
|
)
|
|
self.assertIn(mixed_shade_action, mixed["domain_actions"])
|
|
self.assertEqual(
|
|
[item["capability"] for item in mixed["native_can_actions"]],
|
|
["climate.set_power"],
|
|
)
|
|
|
|
def test_vehicle_lock_is_observational_only_and_fails_closed(self) -> None:
|
|
action = {
|
|
"domain": "door",
|
|
"capability": "door.set_vehicle_lock",
|
|
"target": "vehicle",
|
|
"operation": "set",
|
|
"parameters": {"state": "lock"},
|
|
}
|
|
catalog_entry = self.runtime.capabilities[action["capability"]]
|
|
self.assertEqual(catalog_entry["execution_support"], "observational_only")
|
|
self.assertFalse(catalog_entry["physical_control_supported"])
|
|
self.assertNotIn(action["capability"], self.runtime.supported_capabilities)
|
|
self.assertFalse(self.runtime.supports(action))
|
|
with self.assertRaisesRegex(
|
|
NativeCanCompileError,
|
|
r"unsupported.*current CH1.*0x496.*observational only",
|
|
):
|
|
self.runtime.build_function_config(
|
|
"door", [action], "retired-vehicle-lock", "local_rules"
|
|
)
|
|
with self.assertRaisesRegex(
|
|
NativeCanCompileError, r"unsupported.*current CH1.*observational only"
|
|
):
|
|
self.runtime._frame_specs({"capability": action["capability"]})
|
|
|
|
vehicle = _parse_exact_command("door vehicle lock")
|
|
self.assertEqual(vehicle[0]["capability"], "door.set_vehicle_lock")
|
|
self.assertEqual(vehicle[0]["target"], "vehicle")
|
|
|
|
def test_rear_lock_compiles_only_the_matrix_backed_second_row_request(self) -> None:
|
|
action = {
|
|
"domain": "door",
|
|
"capability": "door.set_rear_lock",
|
|
"target": "rear",
|
|
"operation": "set",
|
|
"parameters": {"state": "lock"},
|
|
}
|
|
catalog_entry = self.runtime.capabilities[action["capability"]]
|
|
self.assertEqual(
|
|
catalog_entry["execution_support"], "matrix_candidate_with_feedback"
|
|
)
|
|
self.assertTrue(catalog_entry["physical_control_supported"])
|
|
self.assertIn(action["capability"], self.runtime.supported_capabilities)
|
|
self.assertTrue(self.runtime.supports(action))
|
|
cases = (
|
|
("lock", "0A 40 00 00 00 00 00 00"),
|
|
("unlock", "0A 80 00 00 00 00 00 00"),
|
|
)
|
|
for state, payload in cases:
|
|
with self.subTest(state=state):
|
|
command = self.compile_action(
|
|
"door", "door.set_rear_lock", "rear", {"state": state}
|
|
)
|
|
self.assertEqual(command["signals"]["frame_id"], "0x32B")
|
|
self.assertEqual(command["signals"]["data_hex"], payload)
|
|
self.assertFalse(command["signals"]["is_fd"])
|
|
self.assertFalse(command["signals"]["bitrate_switch"])
|
|
self.assertNotIn("repeat", command)
|
|
self.assertEqual(command["signals"]["feedback_ids"], "0x407")
|
|
self.assertEqual(
|
|
command["signals"]["expected_feedback"], f"rear_lock:{state}"
|
|
)
|
|
exact = _parse_exact_command(f"door rear {state}")
|
|
self.assertEqual(exact[0]["capability"], "door.set_rear_lock")
|
|
self.assertEqual(exact[0]["target"], "rear")
|
|
|
|
def test_climate_cycle_uses_verified_1de_bits_and_404_feedback(self) -> None:
|
|
cases = (
|
|
("internal", "00 00 08 00 00 00 00 00"),
|
|
("external", "00 00 10 00 00 00 00 00"),
|
|
)
|
|
for mode, payload in cases:
|
|
with self.subTest(mode=mode):
|
|
command = self.compile_action(
|
|
"climate",
|
|
"climate.set_cycle_mode",
|
|
"vehicle",
|
|
{"mode": mode},
|
|
)
|
|
self.assertEqual(command["signals"]["frame_id"], "0x1DE")
|
|
self.assertEqual(command["signals"]["data_hex"], payload)
|
|
self.assertEqual(command["repeat"], {"count": 3, "period_ms": 100})
|
|
self.assertEqual(command["signals"]["feedback_ids"], "0x404")
|
|
self.assertEqual(
|
|
command["signals"]["expected_feedback"],
|
|
f"climate_cycle:{mode}",
|
|
)
|
|
|
|
def test_ambient_controls_use_only_verified_3fc_layouts(self) -> None:
|
|
cases = (
|
|
("ambient_light.set_power", {"state": "on"}, "00 36 00 75 00 80 00 00", "ambient_power:on"),
|
|
("ambient_light.set_power", {"state": "off"}, "00 31 00 75 00 80 00 00", "ambient_power:off"),
|
|
("ambient_light.set_color_index", {"color_index": 70}, "46 30 00 75 00 00 00 00", "ambient_color:70"),
|
|
("ambient_light.set_brightness_level", {"brightness_level": 1}, "00 32 00 75 00 00 00 00", "ambient_brightness:1"),
|
|
("ambient_light.set_brightness_level", {"brightness_level": 5}, "00 36 00 75 00 00 00 00", "ambient_brightness:5"),
|
|
)
|
|
for capability, parameters, payload, expected_feedback in cases:
|
|
with self.subTest(capability=capability, parameters=parameters):
|
|
command = self.compile_action(
|
|
"ambient_light", capability, "vehicle", parameters
|
|
)
|
|
self.assert_fd_brs_triplet(command, "0x3FC")
|
|
self.assertEqual(command["signals"]["data_hex"], payload)
|
|
self.assertEqual(command["signals"]["feedback_ids"], "0x42E")
|
|
self.assertEqual(
|
|
command["signals"]["expected_feedback"], expected_feedback
|
|
)
|
|
|
|
def test_verified_ambient_function_configs_pass_domain_schema(self) -> None:
|
|
validator = ContractValidator(REPOSITORY_ROOT)
|
|
for capability, parameters in (
|
|
("ambient_light.set_color_index", {"color_index": 70}),
|
|
("ambient_light.set_brightness_level", {"brightness_level": 5}),
|
|
):
|
|
with self.subTest(capability=capability):
|
|
config = self.runtime.build_function_config(
|
|
"ambient_light",
|
|
[
|
|
{
|
|
"domain": "ambient_light",
|
|
"capability": capability,
|
|
"target": "vehicle",
|
|
"operation": "set",
|
|
"parameters": parameters,
|
|
}
|
|
],
|
|
"verified-ambient-schema",
|
|
"local_rules",
|
|
)
|
|
self.assertEqual(validator.validate(config), [])
|
|
|
|
def test_ambient_ranges_and_parameter_shapes_fail_closed(self) -> None:
|
|
for capability, parameters in (
|
|
("ambient_light.set_color_index", {"color_index": 0}),
|
|
("ambient_light.set_color_index", {"color_index": 128}),
|
|
("ambient_light.set_brightness_level", {"brightness_level": 0}),
|
|
("ambient_light.set_brightness_level", {"brightness_level": 6}),
|
|
(
|
|
"ambient_light.set_color_index",
|
|
{"color_index": 70, "brightness_level": 3},
|
|
),
|
|
):
|
|
with self.subTest(capability=capability, parameters=parameters):
|
|
with self.assertRaises(NativeCanCompileError):
|
|
self.compile_action("ambient_light", capability, "vehicle", parameters)
|
|
|
|
def test_shade_presets_and_arbitrary_position_never_use_4f5(self) -> None:
|
|
for position, payload in (
|
|
(0, "20 00 00 00 00 00 00 00"),
|
|
(50, "30 00 00 00 00 00 00 00"),
|
|
(100, "10 00 00 00 00 00 00 00"),
|
|
):
|
|
with self.subTest(position=position):
|
|
command = self.compile_action(
|
|
"roof", "roof.set_shade_position", "vehicle", {"position_pct": position}
|
|
)
|
|
self.assertEqual(command["signals"]["frame_id"], "0x431")
|
|
self.assertEqual(command["signals"]["data_hex"], payload)
|
|
self.assertFalse(command["signals"]["is_fd"])
|
|
self.assertEqual(command["signals"]["feedback_ids"], "0x41A")
|
|
self.assertEqual(command["signals"]["special_handler"], "shade_preset")
|
|
self.assertEqual(
|
|
command["signals"]["debug_command"],
|
|
f"shade position {position}",
|
|
)
|
|
|
|
arbitrary = self.compile_action(
|
|
"roof", "roof.set_shade_position", "vehicle", {"position_pct": 37}
|
|
)
|
|
self.assertEqual(arbitrary["signals"]["frame_id"], "0x431")
|
|
self.assertNotEqual(arbitrary["signals"]["frame_id"], "0x4F5")
|
|
self.assertEqual(arbitrary["signals"]["special_handler"], "shade_position")
|
|
self.assertEqual(arbitrary["signals"]["debug_command"], "shade position 37")
|
|
special = arbitrary["signals"]["special_parameters"]
|
|
self.assertIn('"fresh_feedback_timeout_ms": 2500', special)
|
|
self.assertIn('"stop_data_hex": "40 00 00 00 00 00 00 00"', special)
|
|
|
|
def test_executor_attaches_compiled_debug_command_to_mqtt_style_bus(self) -> None:
|
|
config = self.runtime.build_function_config(
|
|
"roof",
|
|
[
|
|
{
|
|
"domain": "roof",
|
|
"capability": "roof.set_shade_position",
|
|
"target": "vehicle",
|
|
"operation": "set",
|
|
"parameters": {"position_pct": 100},
|
|
}
|
|
],
|
|
"shade-debug-command",
|
|
"local_rules",
|
|
)
|
|
plan = self.runtime.compile(config)
|
|
bus = _FakeBus([self.shade_feedback(100, motion=1)])
|
|
executor = NativeCanExecutor(lambda: bus, feedback_timeout_ms=1)
|
|
|
|
report = executor.execute(self.runtime, plan, config)
|
|
|
|
self.assertEqual(report["spec"]["status"], "succeeded")
|
|
self.assertEqual(bus.debug_commands, ["shade position 100"])
|
|
|
|
def test_shade_preset_executor_waits_for_authoritative_target(self) -> None:
|
|
command = self.compile_action(
|
|
"roof", "roof.set_shade_position", "vehicle", {"position_pct": 100}
|
|
)
|
|
reached = self.shade_feedback(99, motion=1)
|
|
bus = _FakeBus([reached])
|
|
executor = NativeCanExecutor(lambda: bus, feedback_timeout_ms=1)
|
|
|
|
sent, feedback = executor._execute_shade_preset(bus, command)
|
|
|
|
self.assertIs(feedback, reached)
|
|
self.assertEqual(len(sent), 1)
|
|
self.assertEqual(
|
|
bus.send_attempts,
|
|
[(0x431, bytes.fromhex("10 00 00 00 00 00 00 00"), False, False)],
|
|
)
|
|
|
|
def test_shade_preset_forces_stop_when_target_feedback_is_still_moving(self) -> None:
|
|
command = self.compile_action(
|
|
"roof", "roof.set_shade_position", "vehicle", {"position_pct": 0}
|
|
)
|
|
reached_while_closing = self.shade_feedback(2, motion=3)
|
|
stopped = self.shade_feedback(2, motion=1)
|
|
bus = _FakeBus([reached_while_closing, stopped])
|
|
executor = NativeCanExecutor(lambda: bus, feedback_timeout_ms=1)
|
|
|
|
sent, feedback = executor._execute_shade_preset(bus, command)
|
|
|
|
self.assertIs(feedback, stopped)
|
|
self.assertEqual(len(sent), 2)
|
|
self.assertEqual(
|
|
[attempt[1] for attempt in bus.send_attempts],
|
|
[
|
|
bytes.fromhex("20 00 00 00 00 00 00 00"),
|
|
bytes.fromhex("40 00 00 00 00 00 00 00"),
|
|
],
|
|
)
|
|
|
|
def test_arbitrary_shade_position_reads_fresh_state_then_stops(self) -> None:
|
|
command = self.compile_action(
|
|
"roof", "roof.set_shade_position", "vehicle", {"position_pct": 37}
|
|
)
|
|
before = self.shade_feedback(20, motion=1)
|
|
reached = self.shade_feedback(38, motion=2)
|
|
stopped = self.shade_feedback(38, motion=1)
|
|
bus = _FakeBus([before, reached, stopped])
|
|
executor = NativeCanExecutor(lambda: bus, feedback_timeout_ms=1)
|
|
|
|
sent, feedback = executor._execute_shade_position(bus, command)
|
|
|
|
self.assertIs(feedback, stopped)
|
|
self.assertEqual(bus.events[0], "receive")
|
|
self.assertEqual(
|
|
[attempt[1] for attempt in bus.send_attempts],
|
|
[
|
|
bytes.fromhex("10 00 00 00 00 00 00 00"),
|
|
bytes.fromhex("40 00 00 00 00 00 00 00"),
|
|
],
|
|
)
|
|
self.assertTrue(all(not attempt[2] and not attempt[3] for attempt in bus.send_attempts))
|
|
self.assertEqual(len(sent), 2)
|
|
|
|
def test_arbitrary_shade_position_always_stops_after_feedback_fault(self) -> None:
|
|
command = self.compile_action(
|
|
"roof", "roof.set_shade_position", "vehicle", {"position_pct": 37}
|
|
)
|
|
bus = _FakeBus()
|
|
executor = NativeCanExecutor(lambda: bus, feedback_timeout_ms=1)
|
|
before = self.shade_feedback(20, motion=1)
|
|
|
|
with patch.object(
|
|
executor,
|
|
"_wait_for_id",
|
|
side_effect=[before, RuntimeError("injected receive failure")],
|
|
):
|
|
with self.assertRaisesRegex(RuntimeError, "injected receive failure"):
|
|
executor._execute_shade_position(bus, command)
|
|
|
|
self.assertEqual(
|
|
[attempt[1] for attempt in bus.send_attempts],
|
|
[
|
|
bytes.fromhex("10 00 00 00 00 00 00 00"),
|
|
bytes.fromhex("40 00 00 00 00 00 00 00"),
|
|
],
|
|
)
|
|
|
|
def test_arbitrary_shade_start_failure_still_attempts_stop(self) -> None:
|
|
command = self.compile_action(
|
|
"roof", "roof.set_shade_position", "vehicle", {"position_pct": 37}
|
|
)
|
|
bus = _FakeBus([self.shade_feedback(20, motion=1)], fail_send_calls={1})
|
|
executor = NativeCanExecutor(lambda: bus, feedback_timeout_ms=1)
|
|
|
|
with self.assertRaisesRegex(RuntimeError, "injected send failure"):
|
|
executor._execute_shade_position(bus, command)
|
|
|
|
self.assertEqual(
|
|
[attempt[1] for attempt in bus.send_attempts],
|
|
[
|
|
bytes.fromhex("10 00 00 00 00 00 00 00"),
|
|
bytes.fromhex("40 00 00 00 00 00 00 00"),
|
|
],
|
|
)
|
|
|
|
def test_arbitrary_shade_missing_or_invalid_feedback_never_starts_motion(self) -> None:
|
|
command = self.compile_action(
|
|
"roof", "roof.set_shade_position", "vehicle", {"position_pct": 37}
|
|
)
|
|
executor = NativeCanExecutor(lambda: _FakeBus(), feedback_timeout_ms=1)
|
|
for before in (None, self.shade_feedback(255, motion=1)):
|
|
bus = _FakeBus()
|
|
with self.subTest(before=before):
|
|
with patch.object(executor, "_wait_for_id", return_value=before):
|
|
with self.assertRaisesRegex(
|
|
RuntimeError, "0x41A|新鲜|有效|fresh|valid"
|
|
):
|
|
executor._execute_shade_position(bus, command)
|
|
self.assertEqual(bus.send_attempts, [])
|
|
|
|
def test_door_feedback_enums_and_shade_stop_match_reference(self) -> None:
|
|
door_data = bytearray(8)
|
|
# 0x407: front-left unlock, front-right lock, rear-left lock,
|
|
# rear-right unlock; whole-vehicle logical state lock.
|
|
door_data[0] = 0b01101001
|
|
door_data[1] = 0b00001000
|
|
decoded = decode_feedback(CanFrame(0x407, bytes(door_data), channel=1))
|
|
locks = decoded["domain_state"]["door"]["locks"]
|
|
self.assertEqual(locks["front_left"]["state"], "unlock")
|
|
self.assertEqual(locks["front_right"]["state"], "lock")
|
|
self.assertEqual(locks["rear_left"]["state"], "lock")
|
|
self.assertEqual(locks["rear_right"]["state"], "unlock")
|
|
self.assertEqual(
|
|
decoded["domain_state"]["door"]["vehicle_lock"]["state"], "lock"
|
|
)
|
|
|
|
all_locked = bytearray(8)
|
|
all_locked[0] = 0xAA
|
|
all_locked[1] = 0x08
|
|
locked_frame = CanFrame(0x407, bytes(all_locked), channel=1)
|
|
self.assertTrue(feedback_matches("vehicle_lock:lock", locked_frame))
|
|
|
|
logical_only = bytearray(all_locked)
|
|
logical_only[0] = 0xA9
|
|
self.assertFalse(
|
|
feedback_matches(
|
|
"vehicle_lock:lock",
|
|
CanFrame(0x407, bytes(logical_only), channel=1),
|
|
)
|
|
)
|
|
|
|
all_unlocked = bytearray(8)
|
|
all_unlocked[0] = 0x55
|
|
all_unlocked[1] = 0x04
|
|
self.assertTrue(
|
|
feedback_matches(
|
|
"vehicle_lock:unlock",
|
|
CanFrame(0x407, bytes(all_unlocked), channel=1),
|
|
)
|
|
)
|
|
|
|
shade_data = bytearray(8)
|
|
shade_data[1] = 37
|
|
shade_data[3] = 1 << 3
|
|
shade = CanFrame(0x41A, bytes(shade_data), channel=1)
|
|
self.assertTrue(feedback_matches("shade_stopped", shade))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|