vee
225 lines
9.6 KiB
Python
225 lines
9.6 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
|
|
from cognitive_agent.application.seat_service import REPOSITORY_ROOT, SeatDemoService
|
|
from cognitive_agent.infrastructure.can.hardware import (
|
|
HardwareSafetyError,
|
|
SeatHardwareSettings,
|
|
VehicleCanSettings,
|
|
)
|
|
|
|
|
|
SUITE_PATH = REPOSITORY_ROOT / "tests" / "fixtures" / "seat_frontend_hil_suite.json"
|
|
EXPECTED_CAPABILITIES = {
|
|
"heating": "seat.set_heating",
|
|
"ventilation": "seat.set_ventilation",
|
|
"longitudinal": "seat.set_longitudinal_position",
|
|
"cushion_tilt": "seat.set_cushion_tilt",
|
|
"backrest_recline": "seat.set_backrest_recline",
|
|
"legrest_height": "seat.set_legrest_height",
|
|
}
|
|
|
|
|
|
def expand_entries(entries: list[dict], group: str | None = None) -> list[dict]:
|
|
expanded = []
|
|
for entry in entries:
|
|
base = {key: value for key, value in entry.items() if key != "variants"}
|
|
if group is not None:
|
|
base["group"] = group
|
|
variants = entry.get("variants")
|
|
if not variants:
|
|
expanded.append(base)
|
|
continue
|
|
expanded.extend({**base, **variant} for variant in variants)
|
|
return expanded
|
|
|
|
|
|
class SeatFrontendInputCoverageTests(unittest.TestCase):
|
|
@classmethod
|
|
def setUpClass(cls) -> None:
|
|
cls.temporary_directory = tempfile.TemporaryDirectory()
|
|
with patch.dict(
|
|
os.environ,
|
|
{"COGNITIVE_AGENT_HARDWARE_EXECUTION_ENABLED": "false"},
|
|
clear=False,
|
|
):
|
|
cls.service = SeatDemoService(REPOSITORY_ROOT)
|
|
cls.service.artifact_root = Path(cls.temporary_directory.name)
|
|
cls.suite = json.loads(SUITE_PATH.read_text(encoding="utf-8"))
|
|
cls.positive_cases = expand_entries(cls.suite["cases"]) + expand_entries(
|
|
cls.suite["mode_cases"], "mode"
|
|
)
|
|
cls.model_cases = expand_entries(cls.suite.get("model_cases", []), "model")
|
|
cls.rejection_cases = expand_entries(cls.suite["rejection_cases"], "safety")
|
|
|
|
@classmethod
|
|
def tearDownClass(cls) -> None:
|
|
cls.temporary_directory.cleanup()
|
|
|
|
def test_suite_has_broad_equivalence_and_safety_coverage(self) -> None:
|
|
self.assertGreaterEqual(len(self.positive_cases), 160)
|
|
self.assertGreaterEqual(len(self.model_cases), 4)
|
|
self.assertGreaterEqual(len(self.rejection_cases), 15)
|
|
self.assertEqual(
|
|
{case["group"] for case in self.positive_cases},
|
|
{"thermal", "motion", "combined", "mode"},
|
|
)
|
|
self.assertEqual(
|
|
{
|
|
case["mode_id"]
|
|
for case in self.positive_cases
|
|
if case["group"] == "mode" and case.get("mode_id")
|
|
},
|
|
{
|
|
"easy_entry_exit",
|
|
"office",
|
|
"queen_passenger_driving",
|
|
"baby_nap",
|
|
"cruise_sofa",
|
|
"meditation_space",
|
|
"leisure_lounger",
|
|
"calf_decompression_float",
|
|
"electric_beach_chair",
|
|
"nap",
|
|
"zero_gravity",
|
|
"rocking_horse",
|
|
"dynamic_cradle",
|
|
},
|
|
)
|
|
|
|
def test_all_expanded_case_ids_and_utterances_are_unique(self) -> None:
|
|
cases = [*self.positive_cases, *self.model_cases, *self.rejection_cases]
|
|
self.assertEqual(len({case["id"] for case in cases}), len(cases))
|
|
self.assertEqual(len({case["utterance"] for case in cases}), len(cases))
|
|
|
|
def test_model_cases_require_gpt56_and_define_physical_targets(self) -> None:
|
|
for case in self.model_cases:
|
|
with self.subTest(case=case["id"]):
|
|
self.assertEqual(case["provider"], "openai")
|
|
self.assertEqual(
|
|
case["expected_provider_prefix"], "openai:gpt-5.6-sol"
|
|
)
|
|
self.assertEqual(
|
|
set(case["expected"]),
|
|
{
|
|
"longitudinal",
|
|
"cushion_tilt",
|
|
"backrest_recline",
|
|
"legrest_height",
|
|
},
|
|
)
|
|
def test_every_positive_utterance_generates_the_expected_dry_run(self) -> None:
|
|
for case in self.positive_cases:
|
|
with self.subTest(case=case["id"], utterance=case["utterance"]):
|
|
result = self.service.process(
|
|
case["utterance"],
|
|
provider=case.get("provider", "local"),
|
|
conversation_id="conv-coverage-positive-sequence",
|
|
execution_mode="dry_run",
|
|
vehicle_id=self.suite["vehicle_id"],
|
|
)
|
|
self.assertEqual(result["status"], "succeeded")
|
|
if case.get("mode_id"):
|
|
self.assertIn(
|
|
case["mode_id"],
|
|
{item["id"] for item in result["matched_modes"]},
|
|
)
|
|
seat_config = result["function_configs"].get("seat") or result[
|
|
"function_config"
|
|
]
|
|
self.assertIsNotNone(seat_config)
|
|
operations = seat_config["spec"]["operations"]
|
|
by_capability = {item["capability"]: item for item in operations}
|
|
passenger = result["actuator_state"]["passenger"]
|
|
for field, expected in case["expected"].items():
|
|
capability = EXPECTED_CAPABILITIES[field]
|
|
self.assertIn(capability, by_capability)
|
|
operation = by_capability[capability]
|
|
self.assertEqual(operation["target"]["instance"], "passenger")
|
|
self.assertEqual(operation["target"]["zone"], "front_right")
|
|
if field in {"heating", "ventilation"}:
|
|
if expected == 0:
|
|
self.assertEqual(operation["parameters"]["state"], "off")
|
|
else:
|
|
self.assertEqual(passenger[field]["level"], expected)
|
|
else:
|
|
self.assertEqual(operation["parameters"]["position_pct"], expected)
|
|
self.assertEqual(
|
|
passenger["position"][field]["position_pct"], expected
|
|
)
|
|
expected_motion = case.get("expected_motion")
|
|
if expected_motion:
|
|
capability = EXPECTED_CAPABILITIES[expected_motion["axis"]]
|
|
operation = by_capability[capability]
|
|
self.assertIn("motion", operation)
|
|
for field, expected in expected_motion.items():
|
|
if field != "axis":
|
|
self.assertEqual(operation["motion"][field], expected)
|
|
expected_dynamic_mode = case.get("expected_dynamic_mode")
|
|
if expected_dynamic_mode:
|
|
capability = (
|
|
"seat.stop_dynamic_mode"
|
|
if expected_dynamic_mode["mode_id"] == 0
|
|
else "seat.start_dynamic_mode"
|
|
)
|
|
self.assertIn(capability, by_capability)
|
|
if expected_dynamic_mode["mode_id"]:
|
|
self.assertEqual(
|
|
by_capability[capability]["parameters"]["mode_id"],
|
|
expected_dynamic_mode["mode_id"],
|
|
)
|
|
|
|
def test_every_rejection_utterance_stops_before_execution(self) -> None:
|
|
for case in self.rejection_cases:
|
|
with self.subTest(case=case["id"], utterance=case["utterance"]):
|
|
result = self.service.process(
|
|
case["utterance"],
|
|
provider=case.get("provider", "local"),
|
|
conversation_id=f"conv-coverage-{case['id'].lower()}",
|
|
execution_mode="dry_run",
|
|
vehicle_id=self.suite["vehicle_id"],
|
|
)
|
|
self.assertEqual(result["status"], case["expected_status"])
|
|
self.assertIsNone(result["function_config"])
|
|
self.assertEqual(result["function_configs"], {})
|
|
self.assertIsNone(result["execution_report"])
|
|
|
|
def test_mixed_ch1_driver_and_ch0_passenger_hardware_scope_fails_closed(self) -> None:
|
|
utterances = (
|
|
"打开主驾和副驾座椅通风",
|
|
"seat driver vent low, seat passenger vent low",
|
|
)
|
|
for index, utterance in enumerate(utterances):
|
|
with self.subTest(utterance=utterance):
|
|
with (
|
|
patch.object(self.service, "_interpret") as interpret,
|
|
patch.object(SeatHardwareSettings, "authorize") as seat_auth,
|
|
patch.object(VehicleCanSettings, "authorize") as vehicle_auth,
|
|
patch.object(VehicleCanSettings, "bridge") as bridge,
|
|
):
|
|
with self.assertRaisesRegex(
|
|
HardwareSafetyError, "mixed CH1.*CH0"
|
|
):
|
|
self.service.process(
|
|
utterance,
|
|
provider="local",
|
|
conversation_id=f"conv-mixed-scope-{index}",
|
|
execution_mode="hardware",
|
|
vehicle_id=self.suite["vehicle_id"],
|
|
)
|
|
|
|
interpret.assert_not_called()
|
|
seat_auth.assert_not_called()
|
|
vehicle_auth.assert_not_called()
|
|
bridge.assert_not_called()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|