Miayui
166 lines
8.0 KiB
Python
166 lines
8.0 KiB
Python
from copy import deepcopy
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
from unittest.mock import patch
|
|
|
|
import pytest
|
|
|
|
from cognitive_agent.application.codex_assistant import CodexAssistantService
|
|
from cognitive_agent.application.romantic_lighting import romantic_light_actions
|
|
from cognitive_agent.application.scene_composer import scene_required_peripheral_refs
|
|
from cognitive_agent.application.mode_profile import ModeProfileRegistry
|
|
from cognitive_agent.application.seat_service import REPOSITORY_ROOT, SeatDemoService
|
|
from cognitive_agent.application.self_extension import (
|
|
ExtensionIntegrationError, GatewayError, SelfExtensionService, SimulationGateway,
|
|
)
|
|
from cognitive_agent.server import SeatDemoHandler
|
|
|
|
|
|
def light(device, name, action, slot, **extra):
|
|
return {
|
|
"id": f"ir-slot:{device}:{slot}", "status": "learned", "enabled": True,
|
|
"device_id": device, "device_name": name, "device_type": "other",
|
|
"card_uid": f"{int(device[-1]):08X}", "slot": slot, "action": action,
|
|
"object_names": [name], "action_names": [action], "risk": "low", **extra,
|
|
}
|
|
|
|
|
|
def catalog():
|
|
return [
|
|
light("light1", "灯笼", "关闭", 1), light("light1", "灯笼", "打开", 2),
|
|
light("light1", "灯笼", "切换为红色", 3), light("light1", "灯笼", "切换为蓝色", 5),
|
|
light("light2", "灯带", "打开", 22), light("light2", "灯带", "设置为红色", 23),
|
|
light("light3", "蜡烛", "打开", 41), light("light3", "蜡烛", "关闭", 42),
|
|
light("light4", "球形灯", "打开", 51), light("light4", "球形灯", "设置为红色", 52),
|
|
]
|
|
|
|
|
|
def test_romantic_lights_only_use_power_on():
|
|
entries = catalog()
|
|
entries += [
|
|
light("light5", "落地灯", "红色", 61), # No known power-on command.
|
|
light("light6", "投影灯", "打开", 71),
|
|
light("light6", "投影灯", "蓝色", 72), # Color capability never becomes a mode action.
|
|
light("light7", "灯笼", "打开", 81, enabled=False),
|
|
light("light7", "灯笼", "红色", 82, enabled=False),
|
|
]
|
|
before = deepcopy(entries)
|
|
assert [item["slot"] for item in romantic_light_actions(entries)] == [2, 22, 41, 51, 71]
|
|
assert entries == before
|
|
|
|
|
|
@pytest.mark.parametrize("name", ["浪漫模式", "浪漫影院模式", "约会模式", "Romantic cinema"])
|
|
def test_creation_removes_colors_and_only_opens_required_lights(name):
|
|
actions = [{
|
|
"ref": f"ir:{item['slot']}", "device_key": item["device_id"],
|
|
"device_label": item["device_name"], "action_label": item["action"],
|
|
"object_names": item["object_names"], "action_names": item["action_names"],
|
|
} for item in catalog()]
|
|
context = {"peripheral_actions": actions}
|
|
required = scene_required_peripheral_refs(name, context)
|
|
assert required == ["ir:2", "ir:22", "ir:41", "ir:51"]
|
|
context["scene_semantics"] = {"mode_name": name, "peripheral_policy": {"required_refs": required}}
|
|
plan = {"requests": [{
|
|
"domain": "self_extension", "capability": "infrared.emit",
|
|
"target_scope": ref, "operation": "emit",
|
|
"parameters": [{"name": "capability_ref", "value": ref}],
|
|
} for ref in ["ir:5", "ir:3", "ir:2"]], "detected_domains": ["self_extension"]}
|
|
result = CodexAssistantService._with_required_scene_peripherals(plan, context)
|
|
assert [item["parameters"][0]["value"] for item in result["requests"]] == required
|
|
|
|
|
|
def test_same_named_lights_remain_separate_devices():
|
|
actions = [{
|
|
"ref": f"ir:{device}-{action}", "device_key": device,
|
|
"device_label": "灯笼", "action_label": label,
|
|
} for device in ["one", "two"] for action, label in [("on", "打开"), ("red", "红色")]]
|
|
assert [item["ref"] for item in romantic_light_actions(actions)] == [
|
|
"ir:one-on", "ir:two-on",
|
|
]
|
|
|
|
|
|
def test_mode_creation_deduplicates_colors_into_authorized_activation():
|
|
capabilities = catalog()
|
|
handler = object.__new__(SeatDemoHandler)
|
|
handler.extension = SimpleNamespace(list_peripheral_capabilities=lambda: capabilities)
|
|
entries = [{"capability": capabilities[index]} for index in [3, 2, 1, 5, 4, 6]]
|
|
selected = handler._mode_creation_activation_entries(entries)
|
|
assert [entry["capability"]["slot"] for entry in selected] == [2, 22, 41]
|
|
|
|
|
|
def test_mode_creation_rejects_color_only_device_or_rebound_power_on():
|
|
red = light("light1", "灯笼", "红色", 3)
|
|
handler = object.__new__(SeatDemoHandler)
|
|
for available in [
|
|
[red],
|
|
[red, light("light1", "灯笼", "打开", 2, enabled=False)],
|
|
[red, light("light1", "灯笼", "打开", 2, card_uid="00000002")],
|
|
]:
|
|
handler.extension = SimpleNamespace(list_peripheral_capabilities=lambda: available)
|
|
with pytest.raises(ExtensionIntegrationError, match="开启指令"):
|
|
handler._mode_creation_activation_entries([{"capability": red}])
|
|
|
|
|
|
def test_scene_defaults_never_fall_back_to_color_only_refs():
|
|
context = {"peripheral_actions": [{
|
|
"ref": "ir:strip-red", "device_label": "灯带", "action_label": "红色",
|
|
}]}
|
|
assert scene_required_peripheral_refs("中秋模式", context) == []
|
|
|
|
|
|
def test_romantic_import_adds_ordered_lights_without_requiring_cinema_base(tmp_path: Path):
|
|
service = SeatDemoService(REPOSITORY_ROOT)
|
|
service.mode_profiles = ModeProfileRegistry(tmp_path / "presets", user_root=tmp_path / "modes")
|
|
service.set_mode_scene_peripheral_catalog_provider(catalog)
|
|
climate = {"type": "control", "at_ms": 0, "domain": "climate",
|
|
"capability": "climate.set_temperature", "target_scope": "vehicle",
|
|
"operation": "set", "parameters": {"temperature_c": 25}}
|
|
scene = {"id": "romantic-date", "name": "浪漫约会模式", "label": "浪漫约会模式",
|
|
"origin": "headunit", "actions": [climate], "scope": {"vehicle_ids": ["denza_n7"]}}
|
|
try:
|
|
for _ in range(2):
|
|
service.import_n7_mode_scene(scene)
|
|
profile = service.get_mode_profile_card("romantic-date", "denza_n7")
|
|
actions = profile.command_flow["source_scene"]["actions"]
|
|
assert actions[0] == climate
|
|
assert [item["parameters"]["slot"] for item in actions[1:]] == [2, 22, 41, 51]
|
|
scene = profile.command_flow["source_scene"]
|
|
assert len(profile.operations) == 5
|
|
finally:
|
|
service.close()
|
|
|
|
|
|
@pytest.mark.parametrize("fail_power", [False, True])
|
|
def test_saved_romantic_mode_orders_real_dispatch_and_stops_on_power_failure(tmp_path: Path, fail_power):
|
|
extension = SelfExtensionService(SimulationGateway(), catalog_db_path=tmp_path / "ir.sqlite3")
|
|
extension.sync_peripheral_catalog(catalog())
|
|
handler = object.__new__(SeatDemoHandler)
|
|
handler.extension = extension
|
|
profile = SimpleNamespace(name="浪漫影院模式", command_flow={"function_configs": {
|
|
"self_extension": {"spec": {"domain": "self_extension", "operations": [{
|
|
"capability": "infrared.emit", "target": {"device": "light1", "card_uid": "00000001"},
|
|
"parameters": {"action": "切换为蓝色", "slot": 5},
|
|
}]}}
|
|
}})
|
|
entries = handler._saved_mode_infrared_entries(profile)
|
|
capabilities = [entry["capability"] for entry in entries]
|
|
assert [item["slot"] for item in capabilities] == [2, 22, 41, 51]
|
|
sent = []
|
|
|
|
def send(_command, parameters):
|
|
sent.append(parameters["slot"])
|
|
if fail_power:
|
|
raise GatewayError("power-on transport failed")
|
|
return {"transport": "tcp", "gateway_acknowledged": True, "request": "mock emit"}
|
|
|
|
with patch.object(extension, "execute_gateway", side_effect=send), patch(
|
|
"cognitive_agent.application.self_extension.time.sleep"
|
|
):
|
|
result = extension.execute_peripheral_plan_request(
|
|
"执行浪漫影院模式", {"steps": [
|
|
{"kind": "peripheral", "capability_id": item["id"]} for item in capabilities
|
|
]}, capabilities, conversation_id="romantic-order-test", execution_mode="hardware",
|
|
)
|
|
assert sent == ([2] if fail_power else [2, 22, 41, 51])
|
|
assert result["status"] == ("failed" if fail_power else "succeeded")
|