vee
988 lines
27 KiB
Python
988 lines
27 KiB
Python
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
|
|
import pytest
|
|
|
|
import cognitive_agent.application.seat_service as seat_service_module
|
|
from cognitive_agent.application.assistant_plan_guard import guard_assistant_plan
|
|
from cognitive_agent.application.seat_service import SeatDemoService
|
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
|
|
|
|
|
def _request(
|
|
domain: str,
|
|
capability: str,
|
|
target: str,
|
|
parameters: dict[str, object],
|
|
*,
|
|
operation: str = "set",
|
|
duration_ms: int | None = None,
|
|
) -> dict[str, object]:
|
|
return {
|
|
"domain": domain,
|
|
"capability": capability,
|
|
"target_scope": target,
|
|
"operation": operation,
|
|
"duration_ms": duration_ms,
|
|
"parameters": [{"name": name, "value": value} for name, value in parameters.items()],
|
|
"reason": "将用户明确表达的控制目标规范化为登记能力。",
|
|
}
|
|
|
|
|
|
def _plan(*requests: dict[str, object]) -> dict[str, object]:
|
|
return {
|
|
"detected_domains": list(dict.fromkeys(str(request["domain"]) for request in requests))
|
|
or ["seat"],
|
|
"requests": list(requests),
|
|
"ambiguities": [],
|
|
"decision_summary": ["仅规范化原始请求明确涉及的能力。"],
|
|
"confidence": 0.95,
|
|
}
|
|
|
|
|
|
@pytest.fixture(scope="module")
|
|
def service(tmp_path_factory: pytest.TempPathFactory) -> SeatDemoService:
|
|
instance = SeatDemoService(ROOT)
|
|
instance.artifact_root = tmp_path_factory.mktemp("assistant-plan-runs")
|
|
yield instance
|
|
instance.close()
|
|
|
|
|
|
def _executable_capabilities(service: SeatDemoService) -> set[str]:
|
|
return {
|
|
str(capability.get("name") or "")
|
|
for domain in service.capability_context["domains"]
|
|
for capability in domain.get("capabilities", [])
|
|
if domain.get("availability") == "executable"
|
|
or str(capability.get("name") or "") in service.native_can_runtime.supported_capabilities
|
|
}
|
|
|
|
|
|
def _guard(
|
|
service: SeatDemoService,
|
|
text: str,
|
|
plan: dict[str, object],
|
|
*,
|
|
trusted_local: dict[str, object] | None = None,
|
|
allow_scene_composition: bool = False,
|
|
):
|
|
return guard_assistant_plan(
|
|
plan,
|
|
fallback_text=text,
|
|
capability_context=service.capability_context,
|
|
executable_capabilities=_executable_capabilities(service),
|
|
trusted_local=trusted_local,
|
|
allow_scene_composition=allow_scene_composition,
|
|
)
|
|
|
|
|
|
def _operations(result: dict[str, object]) -> list[dict[str, object]]:
|
|
return [
|
|
operation
|
|
for config in result.get("function_configs", {}).values()
|
|
for operation in config.get("spec", {}).get("operations", [])
|
|
]
|
|
|
|
|
|
def test_scene_composition_relaxes_only_text_evidence_for_an_atomic_dry_run(
|
|
service: SeatDemoService,
|
|
) -> None:
|
|
scene_plan = _plan(
|
|
_request(
|
|
"climate",
|
|
"climate.set_temperature",
|
|
"vehicle",
|
|
{"temperature_c": 24},
|
|
),
|
|
_request(
|
|
"roof",
|
|
"roof.set_shade_position",
|
|
"vehicle",
|
|
{"position_pct": 0},
|
|
),
|
|
)
|
|
|
|
ordinary = _guard(service, "进入午休模式", scene_plan)
|
|
scene = _guard(
|
|
service,
|
|
"进入午休模式",
|
|
scene_plan,
|
|
allow_scene_composition=True,
|
|
)
|
|
|
|
assert ordinary.plan["requests"] == []
|
|
assert [item["capability"] for item in scene.plan["requests"]] == [
|
|
"climate.set_temperature",
|
|
"roof.set_shade_position",
|
|
]
|
|
assert scene.rejected_requests == ()
|
|
|
|
|
|
def test_scene_template_metadata_is_preserved_only_on_the_scene_path(
|
|
service: SeatDemoService,
|
|
) -> None:
|
|
plan = {
|
|
**_plan(
|
|
_request(
|
|
"climate",
|
|
"climate.set_temperature",
|
|
"vehicle",
|
|
{"temperature_c": 24},
|
|
)
|
|
),
|
|
"metadata": {
|
|
"source": "scene_template",
|
|
"template_id": "cinema",
|
|
"template_revision": "0123456789abcdef",
|
|
},
|
|
}
|
|
|
|
guarded = _guard(
|
|
service,
|
|
"进入观影模式",
|
|
plan,
|
|
allow_scene_composition=True,
|
|
)
|
|
assert guarded.plan["metadata"] == plan["metadata"]
|
|
|
|
with pytest.raises(ValueError, match="only valid for scene composition"):
|
|
_guard(service, "进入观影模式", plan)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"metadata",
|
|
(
|
|
{"source": "model", "template_id": "cinema", "template_revision": "0123456789abcdef"},
|
|
{"source": "scene_template", "template_id": "../cinema", "template_revision": "0123456789abcdef"},
|
|
{"source": "scene_template", "template_id": "cinema", "template_revision": "stale"},
|
|
{
|
|
"source": "scene_template",
|
|
"template_id": "cinema",
|
|
"template_revision": "0123456789abcdef",
|
|
"extra": "untrusted",
|
|
},
|
|
),
|
|
)
|
|
def test_scene_template_metadata_rejects_spoofed_or_malformed_values(
|
|
service: SeatDemoService,
|
|
metadata: dict[str, object],
|
|
) -> None:
|
|
plan = {
|
|
**_plan(
|
|
_request(
|
|
"climate",
|
|
"climate.set_temperature",
|
|
"vehicle",
|
|
{"temperature_c": 24},
|
|
)
|
|
),
|
|
"metadata": metadata,
|
|
}
|
|
|
|
with pytest.raises(ValueError, match="metadata is invalid"):
|
|
_guard(
|
|
service,
|
|
"进入观影模式",
|
|
plan,
|
|
allow_scene_composition=True,
|
|
)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"plan",
|
|
(
|
|
_plan(
|
|
_request(
|
|
"seat",
|
|
"seat.save_posture_preset",
|
|
"passenger",
|
|
{"preset_name": "午休"},
|
|
operation="save",
|
|
)
|
|
),
|
|
_plan(
|
|
_request(
|
|
"climate",
|
|
"climate.set_temperature",
|
|
"vehicle",
|
|
{"temperature_c": 24},
|
|
operation="increase",
|
|
)
|
|
),
|
|
_plan(
|
|
_request(
|
|
"climate",
|
|
"climate.set_temperature",
|
|
"vehicle",
|
|
{"temperature_c": 24},
|
|
duration_ms=3_000,
|
|
)
|
|
),
|
|
_plan(
|
|
_request(
|
|
"climate",
|
|
"climate.set_temperature",
|
|
"vehicle",
|
|
{"temperature_c": 24},
|
|
),
|
|
_request(
|
|
"climate",
|
|
"climate.set_temperature",
|
|
"vehicle",
|
|
{"temperature_c": 25},
|
|
),
|
|
),
|
|
{
|
|
**_plan(
|
|
_request(
|
|
"climate",
|
|
"climate.set_temperature",
|
|
"vehicle",
|
|
{"temperature_c": 24},
|
|
)
|
|
),
|
|
"ambiguities": [{"field": "temperature", "question": "24 or 25"}],
|
|
},
|
|
),
|
|
)
|
|
def test_scene_composition_rejects_non_repeatable_or_non_atomic_plans(
|
|
service: SeatDemoService,
|
|
plan: dict[str, object],
|
|
) -> None:
|
|
guarded = _guard(
|
|
service,
|
|
"进入午休模式",
|
|
plan,
|
|
allow_scene_composition=True,
|
|
)
|
|
|
|
assert guarded.plan["requests"] == []
|
|
assert guarded.rejected_requests
|
|
|
|
|
|
def test_scene_composition_rejects_the_whole_plan_when_one_capability_is_unknown(
|
|
service: SeatDemoService,
|
|
) -> None:
|
|
guarded = _guard(
|
|
service,
|
|
"进入午休模式",
|
|
_plan(
|
|
_request(
|
|
"climate",
|
|
"climate.set_temperature",
|
|
"vehicle",
|
|
{"temperature_c": 24},
|
|
),
|
|
_request("flight", "vehicle.fly", "vehicle", {"altitude_m": 100}),
|
|
),
|
|
allow_scene_composition=True,
|
|
)
|
|
|
|
assert guarded.plan["requests"] == []
|
|
assert len(guarded.rejected_requests) == 1
|
|
|
|
|
|
def test_scene_composition_uses_native_fan_bounds_and_rejects_siblings_atomically(
|
|
service: SeatDemoService,
|
|
) -> None:
|
|
guarded = _guard(
|
|
service,
|
|
"进入静享模式",
|
|
_plan(
|
|
_request(
|
|
"climate",
|
|
"climate.set_fan_level",
|
|
"vehicle",
|
|
{"level": 0},
|
|
),
|
|
_request(
|
|
"climate",
|
|
"climate.set_temperature",
|
|
"vehicle",
|
|
{"temperature_c": 24},
|
|
),
|
|
),
|
|
allow_scene_composition=True,
|
|
)
|
|
|
|
assert guarded.plan["requests"] == []
|
|
assert len(guarded.rejected_requests) == 1
|
|
assert "1-7" in guarded.rejected_requests[0]["question"]
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("text", "plan_request", "expected_capability", "expected_parameters"),
|
|
[
|
|
(
|
|
"空调打开",
|
|
_request(
|
|
"climate",
|
|
"climate.set_power",
|
|
"vehicle",
|
|
{"state": "on"},
|
|
operation="activate",
|
|
),
|
|
"climate.set_power",
|
|
{"state": "on"},
|
|
),
|
|
(
|
|
"天幕透光率调到30%",
|
|
_request(
|
|
"roof",
|
|
"roof.set_transmittance",
|
|
"vehicle",
|
|
{"transmittance_pct": 30},
|
|
),
|
|
"roof.set_transmittance",
|
|
{"transmittance_pct": 30},
|
|
),
|
|
(
|
|
"开启天幕自动调光",
|
|
_request(
|
|
"roof",
|
|
"roof.set_automatic_dimming",
|
|
"vehicle",
|
|
{"state": "on"},
|
|
operation="activate",
|
|
),
|
|
"roof.set_automatic_dimming",
|
|
{"state": "on"},
|
|
),
|
|
(
|
|
"氛围灯调成蓝色，亮度50%",
|
|
_request(
|
|
"ambient_light",
|
|
"ambient_light.set_illumination",
|
|
"vehicle",
|
|
{"color_hex": "#0000FF", "brightness_pct": 50},
|
|
),
|
|
"ambient_light.set_illumination",
|
|
{"color_hex": "#0000FF", "brightness_pct": 50},
|
|
),
|
|
(
|
|
"氛围灯设置为影院模式",
|
|
_request(
|
|
"ambient_light",
|
|
"ambient_light.set_mode",
|
|
"vehicle",
|
|
{"mode": "cinema"},
|
|
),
|
|
"ambient_light.set_mode",
|
|
{"mode": "cinema"},
|
|
),
|
|
(
|
|
"副驾腿托伸出到40%",
|
|
_request(
|
|
"seat",
|
|
"seat.set_legrest_extension",
|
|
"passenger",
|
|
{"position_pct": 40},
|
|
),
|
|
"seat.set_legrest_extension",
|
|
{"position_pct": 40},
|
|
),
|
|
],
|
|
)
|
|
def test_explicit_registered_plan_executes_through_existing_pipeline(
|
|
service: SeatDemoService,
|
|
text: str,
|
|
plan_request: dict[str, object],
|
|
expected_capability: str,
|
|
expected_parameters: dict[str, object],
|
|
) -> None:
|
|
with patch.object(
|
|
service.openai,
|
|
"interpret",
|
|
side_effect=AssertionError("Codex plan must not invoke the legacy hosted parser"),
|
|
):
|
|
result = service.preview_native_can(
|
|
text,
|
|
provider="local",
|
|
assistant_plan=_plan(plan_request),
|
|
)
|
|
|
|
operations = _operations(result)
|
|
assert result["status"] == "succeeded"
|
|
assert result["provider_used"] == "local_rules+codex_app_server"
|
|
assert [(item["capability"], item["parameters"]) for item in operations] == [
|
|
(expected_capability, expected_parameters)
|
|
]
|
|
assert operations[0]["parse_source"] == "cloud_model"
|
|
|
|
|
|
def test_named_preset_can_be_saved_then_applied_by_guarded_plan(
|
|
service: SeatDemoService,
|
|
) -> None:
|
|
conversation_id = "conv-assistant-plan-preset"
|
|
saved = service.preview_native_can(
|
|
"保存当前座椅姿态为通勤",
|
|
provider="local",
|
|
conversation_id=conversation_id,
|
|
assistant_plan=_plan(
|
|
_request(
|
|
"seat",
|
|
"seat.save_posture_preset",
|
|
"passenger",
|
|
{"preset_name": "通勤"},
|
|
operation="save",
|
|
)
|
|
),
|
|
)
|
|
applied = service.preview_native_can(
|
|
"应用座椅预设通勤",
|
|
provider="local",
|
|
conversation_id=conversation_id,
|
|
assistant_plan=_plan(
|
|
_request(
|
|
"seat",
|
|
"seat.apply_posture_preset",
|
|
"passenger",
|
|
{"preset_name": "通勤"},
|
|
operation="apply",
|
|
)
|
|
),
|
|
)
|
|
|
|
assert saved["status"] == "succeeded"
|
|
assert saved["saved_presets"] == [{"name": "通勤", "target": "passenger"}]
|
|
assert applied["status"] == "succeeded"
|
|
assert {item["capability"] for item in _operations(applied)} == {
|
|
"seat.set_longitudinal_position",
|
|
"seat.set_cushion_tilt",
|
|
"seat.set_backrest_recline",
|
|
"seat.set_legrest_height",
|
|
"seat.set_legrest_extension",
|
|
}
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("text", "plan_request"),
|
|
[
|
|
(
|
|
"关闭空调",
|
|
_request("climate", "climate.set_power", "vehicle", {"state": "on"}),
|
|
),
|
|
(
|
|
"打开遮阳帘",
|
|
_request(
|
|
"roof",
|
|
"roof.set_shade_motion",
|
|
"vehicle",
|
|
{"motion": "close"},
|
|
),
|
|
),
|
|
(
|
|
"遮阳帘打开30%",
|
|
_request(
|
|
"roof",
|
|
"roof.set_shade_position",
|
|
"vehicle",
|
|
{"position_pct": 70},
|
|
),
|
|
),
|
|
(
|
|
"空调温度调到23度",
|
|
_request(
|
|
"climate",
|
|
"climate.set_temperature",
|
|
"vehicle",
|
|
{"temperature_c": 26},
|
|
),
|
|
),
|
|
(
|
|
"风量调到3档",
|
|
_request(
|
|
"climate",
|
|
"climate.set_fan_level",
|
|
"vehicle",
|
|
{"level": 2},
|
|
),
|
|
),
|
|
(
|
|
"氛围灯亮度调到3档",
|
|
_request(
|
|
"ambient_light",
|
|
"ambient_light.set_brightness_level",
|
|
"vehicle",
|
|
{"brightness_level": 5},
|
|
),
|
|
),
|
|
(
|
|
"智能天幕透光率调到30%",
|
|
_request(
|
|
"roof",
|
|
"roof.set_transmittance",
|
|
"vehicle",
|
|
{"transmittance_pct": 70},
|
|
),
|
|
),
|
|
(
|
|
"后门上锁",
|
|
_request("door", "door.set_rear_lock", "rear", {"state": "unlock"}),
|
|
),
|
|
(
|
|
"把空调温度调低一点",
|
|
_request(
|
|
"climate",
|
|
"climate.set_temperature",
|
|
"vehicle",
|
|
{"temperature_c": 23},
|
|
operation="increase",
|
|
),
|
|
),
|
|
(
|
|
"把空调温度调低一点",
|
|
_request(
|
|
"climate",
|
|
"climate.set_temperature",
|
|
"vehicle",
|
|
{"temperature_c": 23},
|
|
operation="decrease",
|
|
),
|
|
),
|
|
],
|
|
)
|
|
def test_reversed_or_invented_value_is_filtered(
|
|
service: SeatDemoService,
|
|
text: str,
|
|
plan_request: dict[str, object],
|
|
) -> None:
|
|
guarded = _guard(service, text, _plan(plan_request))
|
|
|
|
assert guarded.plan["requests"] == []
|
|
assert len(guarded.rejected_requests) == 1
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("text", "plan_request"),
|
|
[
|
|
(
|
|
"整车上锁",
|
|
_request("door", "door.set_vehicle_lock", "vehicle", {"state": "lock"}),
|
|
),
|
|
(
|
|
"打开壁灯",
|
|
_request(
|
|
"interior_light",
|
|
"interior_light.set_wall_power",
|
|
"vehicle",
|
|
{"state": "on"},
|
|
),
|
|
),
|
|
],
|
|
)
|
|
def test_catalog_only_or_unimplemented_native_capability_is_filtered(
|
|
service: SeatDemoService,
|
|
text: str,
|
|
plan_request: dict[str, object],
|
|
) -> None:
|
|
guarded = _guard(service, text, _plan(plan_request))
|
|
|
|
assert guarded.plan["requests"] == []
|
|
assert len(guarded.rejected_requests) == 1
|
|
|
|
|
|
def test_model_cannot_expand_an_explicit_window_target(
|
|
service: SeatDemoService,
|
|
) -> None:
|
|
guarded = _guard(
|
|
service,
|
|
"打开左前车窗",
|
|
_plan(
|
|
_request(
|
|
"window",
|
|
"window.set_motion",
|
|
"rear_right",
|
|
{"motion": "open"},
|
|
)
|
|
),
|
|
)
|
|
|
|
assert guarded.plan["requests"] == []
|
|
assert guarded.rejected_requests[0]["field"].endswith("target_scope")
|
|
|
|
|
|
@pytest.mark.parametrize(("text", "state"), [("打开空调", "on"), ("关闭空调", "off")])
|
|
def test_hvac_power_wording_does_not_authorize_compressor_control(
|
|
service: SeatDemoService,
|
|
text: str,
|
|
state: str,
|
|
) -> None:
|
|
guarded = _guard(
|
|
service,
|
|
text,
|
|
_plan(
|
|
_request(
|
|
"climate",
|
|
"climate.set_ac",
|
|
"vehicle",
|
|
{"state": state},
|
|
operation="activate" if state == "on" else "deactivate",
|
|
)
|
|
),
|
|
)
|
|
|
|
assert guarded.plan["requests"] == []
|
|
assert guarded.rejected_requests[0]["field"].endswith("capability")
|
|
|
|
|
|
def test_local_control_wins_and_plan_only_fills_unparsed_clause(
|
|
service: SeatDemoService,
|
|
) -> None:
|
|
result = service.preview_native_can(
|
|
"打开遮阳帘，空调打开",
|
|
provider="local",
|
|
assistant_plan=_plan(
|
|
_request(
|
|
"roof",
|
|
"roof.set_shade_motion",
|
|
"vehicle",
|
|
{"motion": "close"},
|
|
),
|
|
_request(
|
|
"climate",
|
|
"climate.set_power",
|
|
"vehicle",
|
|
{"state": "on"},
|
|
operation="activate",
|
|
),
|
|
),
|
|
)
|
|
|
|
operations = _operations(result)
|
|
by_capability = {item["capability"]: item for item in operations}
|
|
assert by_capability["roof.set_shade_motion"]["parameters"] == {"motion": "open"}
|
|
assert by_capability["roof.set_shade_motion"]["parse_source"] == "local_rule"
|
|
assert by_capability["climate.set_power"]["parameters"] == {"state": "on"}
|
|
assert by_capability["climate.set_power"]["parse_source"] == "cloud_model"
|
|
|
|
|
|
def test_plan_still_passes_through_language_safety_guard(
|
|
service: SeatDemoService,
|
|
) -> None:
|
|
original_guard = seat_service_module.apply_language_safety_guards
|
|
with patch.object(
|
|
seat_service_module,
|
|
"apply_language_safety_guards",
|
|
wraps=original_guard,
|
|
) as safety_guard:
|
|
result = service.preview_native_can(
|
|
"空调打开",
|
|
provider="local",
|
|
assistant_plan=_plan(
|
|
_request(
|
|
"climate",
|
|
"climate.set_power",
|
|
"vehicle",
|
|
{"state": "on"},
|
|
operation="activate",
|
|
)
|
|
),
|
|
)
|
|
|
|
assert result["status"] == "succeeded"
|
|
safety_guard.assert_called_once()
|
|
assert safety_guard.call_args.args[1] == "空调打开"
|
|
|
|
|
|
def test_adapter_range_check_rejects_even_text_evidenced_value(
|
|
service: SeatDemoService,
|
|
) -> None:
|
|
result = service.preview_native_can(
|
|
"空调温度调到99度",
|
|
provider="local",
|
|
assistant_plan=_plan(
|
|
_request(
|
|
"climate",
|
|
"climate.set_temperature",
|
|
"vehicle",
|
|
{"temperature_c": 99},
|
|
)
|
|
),
|
|
)
|
|
|
|
assert result["status"] == "no_action"
|
|
assert _operations(result) == []
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"message",
|
|
(
|
|
"为什么要打开空调",
|
|
"请复述：打开空调",
|
|
"写一个故事，主角打开空调",
|
|
),
|
|
)
|
|
def test_non_execution_question_cannot_authorize_a_plan(
|
|
service: SeatDemoService,
|
|
message: str,
|
|
) -> None:
|
|
guarded = _guard(
|
|
service,
|
|
message,
|
|
_plan(
|
|
_request(
|
|
"climate",
|
|
"climate.set_power",
|
|
"vehicle",
|
|
{"state": "on"},
|
|
operation="activate",
|
|
)
|
|
),
|
|
)
|
|
|
|
assert guarded.plan["requests"] == []
|
|
assert len(guarded.rejected_requests) == 1
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("text", "valid_request", "reversed_request"),
|
|
[
|
|
(
|
|
"氛围灯颜色编号32",
|
|
_request(
|
|
"ambient_light",
|
|
"ambient_light.set_color_index",
|
|
"vehicle",
|
|
{"color_index": 32},
|
|
),
|
|
_request(
|
|
"ambient_light",
|
|
"ambient_light.set_color_index",
|
|
"vehicle",
|
|
{"color_index": 70},
|
|
),
|
|
),
|
|
(
|
|
"主驾左风口水平标定值-20，垂直标定值30",
|
|
_request(
|
|
"outlet",
|
|
"outlet.set_calibrated_position",
|
|
"driver_left",
|
|
{"horizontal_value": -20, "vertical_value": 30},
|
|
),
|
|
_request(
|
|
"outlet",
|
|
"outlet.set_calibrated_position",
|
|
"driver_left",
|
|
{"horizontal_value": 20, "vertical_value": 30},
|
|
),
|
|
),
|
|
(
|
|
"召回主驾座椅2号记忆位置",
|
|
_request(
|
|
"seat",
|
|
"seat.recall_driver_memory_native",
|
|
"driver",
|
|
{"slot": 2},
|
|
),
|
|
_request(
|
|
"seat",
|
|
"seat.recall_driver_memory_native",
|
|
"driver",
|
|
{"slot": 3},
|
|
),
|
|
),
|
|
(
|
|
"副驾靠背调到120度",
|
|
_request(
|
|
"seat",
|
|
"seat.set_backrest_recline",
|
|
"passenger",
|
|
{"angle_deg": 120},
|
|
),
|
|
_request(
|
|
"seat",
|
|
"seat.set_backrest_recline",
|
|
"passenger",
|
|
{"angle_deg": 130},
|
|
),
|
|
),
|
|
(
|
|
"副驾座椅向后移动20毫米",
|
|
_request(
|
|
"seat",
|
|
"seat.set_longitudinal_position",
|
|
"passenger",
|
|
{"distance_mm": 20},
|
|
),
|
|
_request(
|
|
"seat",
|
|
"seat.set_longitudinal_position",
|
|
"passenger",
|
|
{"distance_mm": 30},
|
|
),
|
|
),
|
|
(
|
|
"主驾座椅按摩强度开到3档",
|
|
_request(
|
|
"seat",
|
|
"seat.set_driver_massage_native",
|
|
"driver",
|
|
{"action": "strength", "strength": 3},
|
|
),
|
|
_request(
|
|
"seat",
|
|
"seat.set_driver_massage_native",
|
|
"driver",
|
|
{"action": "strength", "strength": 2},
|
|
),
|
|
),
|
|
(
|
|
"启动副驾木马模式",
|
|
_request(
|
|
"seat",
|
|
"seat.start_dynamic_mode",
|
|
"passenger",
|
|
{"mode_id": 1},
|
|
operation="activate",
|
|
),
|
|
_request(
|
|
"seat",
|
|
"seat.start_dynamic_mode",
|
|
"passenger",
|
|
{"mode_id": 2},
|
|
operation="activate",
|
|
),
|
|
),
|
|
],
|
|
)
|
|
def test_remaining_numeric_and_action_parameters_are_text_bound(
|
|
service: SeatDemoService,
|
|
text: str,
|
|
valid_request: dict[str, object],
|
|
reversed_request: dict[str, object],
|
|
) -> None:
|
|
accepted = _guard(service, text, _plan(valid_request))
|
|
rejected = _guard(service, text, _plan(reversed_request))
|
|
|
|
assert len(accepted.plan["requests"]) == 1
|
|
assert accepted.rejected_requests == ()
|
|
assert rejected.plan["requests"] == []
|
|
assert len(rejected.rejected_requests) == 1
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("text", "plan_request"),
|
|
[
|
|
(
|
|
"关闭副驾空调",
|
|
_request(
|
|
"seat",
|
|
"seat.stop_dynamic_mode",
|
|
"passenger",
|
|
{},
|
|
operation="deactivate",
|
|
),
|
|
),
|
|
(
|
|
"打开副驾空调加热",
|
|
_request(
|
|
"seat",
|
|
"seat.set_heating",
|
|
"passenger",
|
|
{"state": "on", "level": 2},
|
|
operation="activate",
|
|
),
|
|
),
|
|
],
|
|
)
|
|
def test_occupant_words_in_climate_requests_cannot_authorize_seat_controls(
|
|
service: SeatDemoService,
|
|
text: str,
|
|
plan_request: dict[str, object],
|
|
) -> None:
|
|
guarded = _guard(service, text, _plan(plan_request))
|
|
|
|
assert guarded.plan["requests"] == []
|
|
assert len(guarded.rejected_requests) == 1
|
|
|
|
|
|
@pytest.mark.parametrize("sentinel", ("unknown", "unspecified", "default"))
|
|
def test_placeholder_targets_are_never_treated_as_implicit_authorization(
|
|
service: SeatDemoService,
|
|
sentinel: str,
|
|
) -> None:
|
|
guarded = _guard(
|
|
service,
|
|
"打开左前车窗",
|
|
_plan(
|
|
_request(
|
|
"window",
|
|
"window.set_motion",
|
|
sentinel,
|
|
{"motion": "open"},
|
|
)
|
|
),
|
|
)
|
|
|
|
assert guarded.plan["requests"] == []
|
|
assert guarded.rejected_requests[0]["field"].endswith("target_scope")
|
|
|
|
|
|
@pytest.mark.parametrize("target", ("front_left", "all"))
|
|
def test_local_reading_light_request_cannot_expand_to_another_scope(
|
|
service: SeatDemoService,
|
|
target: str,
|
|
) -> None:
|
|
guarded = _guard(
|
|
service,
|
|
"打开主驾阅读灯",
|
|
_plan(
|
|
_request(
|
|
"interior_light",
|
|
"interior_light.set_reading_power",
|
|
target,
|
|
{"state": "on"},
|
|
operation="activate",
|
|
)
|
|
),
|
|
)
|
|
|
|
assert guarded.plan["requests"] == []
|
|
assert len(guarded.rejected_requests) == 1
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("text", "capability", "valid", "swapped"),
|
|
[
|
|
(
|
|
"主驾左风口水平25%垂直75%",
|
|
"outlet.set_position_percent",
|
|
{"horizontal_pct": 25, "vertical_pct": 75},
|
|
{"horizontal_pct": 75, "vertical_pct": 25},
|
|
),
|
|
(
|
|
"主驾左风口水平-20垂直30标定坐标",
|
|
"outlet.set_calibrated_position",
|
|
{"horizontal_value": -20, "vertical_value": 30},
|
|
{"horizontal_value": 30, "vertical_value": -20},
|
|
),
|
|
],
|
|
)
|
|
def test_outlet_axis_values_cannot_be_swapped_by_a_model(
|
|
service: SeatDemoService,
|
|
text: str,
|
|
capability: str,
|
|
valid: dict[str, object],
|
|
swapped: dict[str, object],
|
|
) -> None:
|
|
accepted = _guard(
|
|
service,
|
|
text,
|
|
_plan(_request("outlet", capability, "driver_left", valid)),
|
|
)
|
|
rejected = _guard(
|
|
service,
|
|
text,
|
|
_plan(_request("outlet", capability, "driver_left", swapped)),
|
|
)
|
|
|
|
assert len(accepted.plan["requests"]) == 1
|
|
assert accepted.rejected_requests == ()
|
|
assert rejected.plan["requests"] == []
|
|
assert len(rejected.rejected_requests) == 1
|