vee
313 lines
12 KiB
Python
313 lines
12 KiB
Python
from __future__ import annotations
|
|
|
|
import unittest
|
|
|
|
from cognitive_agent.application.conversation_state import ConversationStateStore
|
|
|
|
|
|
class NativeStateSyncTests(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.store = ConversationStateStore()
|
|
|
|
def apply(self, action: dict) -> dict:
|
|
return self.store.apply_successful_actions(
|
|
"conv-native-sync",
|
|
"test",
|
|
[{**action, "feedback_source": "physical_can_ch1"}],
|
|
[],
|
|
)
|
|
|
|
def test_driver_seat_native_feedback_never_updates_passenger(self) -> None:
|
|
state = self.apply(
|
|
{
|
|
"domain": "seat",
|
|
"capability": "seat.set_driver_heating_native",
|
|
"target": "driver",
|
|
"parameters": {"state": "low"},
|
|
}
|
|
)
|
|
self.assertEqual(state["devices"]["driver"]["heating"]["state"], "on")
|
|
self.assertEqual(state["devices"]["driver"]["heating"]["level"], 1)
|
|
self.assertEqual(
|
|
state["devices"]["driver"]["heating"]["feedback_source"],
|
|
"physical_can_ch1",
|
|
)
|
|
self.assertEqual(state["devices"]["passenger"]["heating"]["state"], "off")
|
|
|
|
def test_window_feedback_updates_frontend_domain_state(self) -> None:
|
|
state = self.apply(
|
|
{
|
|
"domain": "window",
|
|
"capability": "window.set_position",
|
|
"target": "front_left",
|
|
"parameters": {"position_pct": 25},
|
|
}
|
|
)
|
|
window = state["domain_devices"]["window"]["positions"]["front_left"]
|
|
self.assertEqual(window["position_pct"], 25)
|
|
self.assertFalse(window["closed"])
|
|
self.assertEqual(window["feedback_source"], "physical_can_ch1")
|
|
|
|
def test_rear_lock_updates_only_rear_locks(self) -> None:
|
|
state = self.apply(
|
|
{
|
|
"domain": "door",
|
|
"capability": "door.set_rear_lock",
|
|
"target": "rear",
|
|
"parameters": {"state": "lock"},
|
|
}
|
|
)
|
|
locks = state["domain_devices"]["door"]["locks"]
|
|
self.assertEqual(locks["rear_left"]["state"], "lock")
|
|
self.assertEqual(locks["rear_right"]["state"], "lock")
|
|
self.assertEqual(locks["front_left"]["state"], "unknown")
|
|
|
|
def test_vehicle_lock_projection_updates_logical_and_all_door_states(self) -> None:
|
|
state = self.apply(
|
|
{
|
|
"domain": "door",
|
|
"capability": "door.set_vehicle_lock",
|
|
"target": "vehicle",
|
|
"parameters": {"state": "lock"},
|
|
}
|
|
)
|
|
door = state["domain_devices"]["door"]
|
|
self.assertEqual(door["vehicle_lock"]["state"], "lock")
|
|
self.assertTrue(
|
|
all(item["state"] == "lock" for item in door["locks"].values())
|
|
)
|
|
|
|
def test_authoritative_ch1_feedback_overrides_requested_ui_projection(self) -> None:
|
|
state = self.store.synchronize_physical_ch1_feedback(
|
|
"conv-native-sync",
|
|
[
|
|
{
|
|
"source": "physical_can_ch1",
|
|
"channel": 1,
|
|
"frame_id": "0x404",
|
|
"feedback_frame_id": "0x404",
|
|
"valid": True,
|
|
"verification": {
|
|
"authoritative": True,
|
|
"channel": 1,
|
|
"simulated": False,
|
|
"feedback_required": True,
|
|
"feedback_observed": True,
|
|
"feedback_match": True,
|
|
"frame_id": "0x404",
|
|
"payload_size": 8,
|
|
},
|
|
"actuator_state": {
|
|
"driver": {
|
|
"ventilation": {"state": "on", "level": 3},
|
|
"massage": {"mode": "relax", "strength": "weak"},
|
|
}
|
|
},
|
|
"domain_state": {
|
|
"climate": {
|
|
"power": "on",
|
|
"ac": "on",
|
|
"control_mode": "manual",
|
|
"cycle_mode": "internal",
|
|
"fan_level": 3,
|
|
"zones": {
|
|
"driver": {
|
|
"temperature_c": 24,
|
|
"outlet_mode": "defrost",
|
|
}
|
|
},
|
|
},
|
|
"roof": {
|
|
"shade_position_pct": 2,
|
|
"motion_state": "stopped",
|
|
},
|
|
"ambient_light": {
|
|
"power": "on",
|
|
"front_brightness_level": 3,
|
|
"rear_brightness_level": 3,
|
|
"front_color_index": 32,
|
|
"rear_color_index": 32,
|
|
},
|
|
"door": {
|
|
"locks": {
|
|
"rear_left": {"state": "lock"},
|
|
"rear_right": {"state": "lock"},
|
|
}
|
|
},
|
|
"window": {
|
|
"positions": {
|
|
"front_left": {
|
|
"position_pct": 9,
|
|
"closed": False,
|
|
}
|
|
}
|
|
},
|
|
},
|
|
}
|
|
],
|
|
)
|
|
|
|
self.assertEqual(state["devices"]["driver"]["ventilation"]["level"], 3)
|
|
self.assertEqual(
|
|
state["devices"]["driver"]["ventilation"]["feedback_source"],
|
|
"physical_can_ch1",
|
|
)
|
|
climate = state["domain_devices"]["climate"]
|
|
self.assertEqual(climate["zones"]["driver"]["outlet_mode"], "defrost")
|
|
self.assertEqual(climate["cycle_mode"], "internal")
|
|
self.assertEqual(climate["feedback_source"], "physical_can_ch1")
|
|
self.assertEqual(state["domain_devices"]["roof"]["shade_position_pct"], 2)
|
|
self.assertEqual(state["domain_devices"]["ambient_light"]["brightness_pct"], 100)
|
|
self.assertEqual(
|
|
state["domain_devices"]["door"]["locks"]["rear_left"]["feedback_source"],
|
|
"physical_can_ch1",
|
|
)
|
|
self.assertEqual(
|
|
state["domain_devices"]["window"]["positions"]["front_left"]["position_pct"],
|
|
9,
|
|
)
|
|
self.assertNotIn(
|
|
"feedback_source",
|
|
state["domain_devices"]["window"]["positions"]["rear_left"],
|
|
)
|
|
self.assertNotIn(
|
|
"feedback_source",
|
|
state["domain_devices"]["door"]["locks"]["front_left"],
|
|
)
|
|
|
|
def test_non_authoritative_feedback_is_ignored(self) -> None:
|
|
before = self.store.snapshot("conv-native-sync")
|
|
after = self.store.synchronize_physical_ch1_feedback(
|
|
"conv-native-sync",
|
|
[
|
|
{
|
|
"source": "physical_can_ch0",
|
|
"channel": 0,
|
|
"valid": True,
|
|
"domain_state": {"climate": {"power": "on"}},
|
|
}
|
|
],
|
|
)
|
|
self.assertEqual(after, before)
|
|
|
|
def test_verified_read_only_ch1_feedback_updates_only_driver(self) -> None:
|
|
state = self.store.synchronize_physical_ch1_feedback(
|
|
"conv-native-sync",
|
|
[{
|
|
"source": "physical_can_ch1",
|
|
"channel": 1,
|
|
"frame_id": "0x4FA",
|
|
"feedback_frame_id": "0x4FA",
|
|
"valid": True,
|
|
"verification": {
|
|
"authoritative": True,
|
|
"channel": 1,
|
|
"simulated": False,
|
|
"read_only": True,
|
|
"frame_id": "0x4FA",
|
|
"payload_size": 8,
|
|
},
|
|
"actuator_state": {
|
|
"driver": {"ventilation": {"state": "on", "level": 3}}
|
|
},
|
|
}],
|
|
)
|
|
self.assertEqual(state["devices"]["driver"]["ventilation"]["level"], 3)
|
|
self.assertEqual(
|
|
state["devices"]["driver"]["ventilation"]["feedback_source"],
|
|
"physical_can_ch1",
|
|
)
|
|
self.assertEqual(state["devices"]["passenger"]["ventilation"]["state"], "off")
|
|
|
|
def test_verified_vehicle_safety_and_door_feedback_feed_whole_vehicle_state(self) -> None:
|
|
def feedback(frame_id: str, **state_patch: dict) -> dict:
|
|
return {
|
|
"source": "physical_can_ch1",
|
|
"channel": 1,
|
|
"frame_id": frame_id,
|
|
"feedback_frame_id": frame_id,
|
|
"valid": True,
|
|
"verification": {
|
|
"authoritative": True,
|
|
"channel": 1,
|
|
"simulated": False,
|
|
"read_only": True,
|
|
"frame_id": frame_id,
|
|
"payload_size": 8,
|
|
},
|
|
**state_patch,
|
|
}
|
|
|
|
state = self.store.synchronize_physical_ch1_feedback(
|
|
"conv-native-sync",
|
|
[
|
|
feedback("0x342", vehicle_safety={"gear": "P"}),
|
|
feedback("0x4EB", vehicle_safety={"speed_kmh": 0.0}),
|
|
feedback(
|
|
"0x407",
|
|
domain_state={
|
|
"door": {
|
|
"locks": {
|
|
"front_left": {"state": "lock"},
|
|
"front_right": {"state": "unlock"},
|
|
"rear_left": {"state": "lock"},
|
|
"rear_right": {"state": "lock"},
|
|
},
|
|
"vehicle_lock": {"state": "unknown"},
|
|
}
|
|
},
|
|
),
|
|
],
|
|
)
|
|
|
|
vehicle_status = state["domain_devices"]["vehicle_status"]
|
|
self.assertEqual(vehicle_status["gear"], "P")
|
|
self.assertEqual(vehicle_status["speed_kmh"], 0.0)
|
|
self.assertTrue(vehicle_status["stationary"])
|
|
self.assertEqual(vehicle_status["feedback_source"], "physical_can_ch1")
|
|
door = state["domain_devices"]["door"]
|
|
self.assertEqual(door["locks"]["front_right"]["state"], "unlock")
|
|
self.assertEqual(
|
|
door["locks"]["front_right"]["feedback_source"],
|
|
"physical_can_ch1",
|
|
)
|
|
|
|
def test_unverified_or_empty_feedback_never_updates_state(self) -> None:
|
|
before = self.store.snapshot("conv-native-sync")
|
|
for feedback in (
|
|
{
|
|
"source": "physical_can_ch1",
|
|
"channel": 1,
|
|
"frame_id": "0x404",
|
|
"feedback_frame_id": "0x404",
|
|
"valid": True,
|
|
"domain_state": {"climate": {"power": "on"}},
|
|
},
|
|
{
|
|
"source": "physical_can_ch1",
|
|
"channel": 1,
|
|
"frame_id": "0x404",
|
|
"feedback_frame_id": "0x404",
|
|
"valid": True,
|
|
"verification": {
|
|
"authoritative": True,
|
|
"channel": 1,
|
|
"simulated": False,
|
|
"feedback_required": True,
|
|
"feedback_observed": True,
|
|
"feedback_match": True,
|
|
"frame_id": "0x404",
|
|
"payload_size": 8,
|
|
},
|
|
},
|
|
):
|
|
with self.subTest(feedback=feedback):
|
|
after = self.store.synchronize_physical_ch1_feedback(
|
|
"conv-native-sync", [feedback]
|
|
)
|
|
self.assertEqual(after, before)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|