Miayui
868 lines
29 KiB
Python
868 lines
29 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import time
|
|
from copy import deepcopy
|
|
from pathlib import Path
|
|
from unittest.mock import MagicMock
|
|
|
|
import pytest
|
|
|
|
from cognitive_agent.application.assistant_capabilities import (
|
|
AssistantCapabilityCatalog,
|
|
)
|
|
from cognitive_agent.application.mode_profile import ModeProfileRegistry
|
|
from cognitive_agent.application.mode_scene_ingress import (
|
|
ModeSceneIngressConsumer,
|
|
ModeSceneIngressError,
|
|
parse_mode_scene_ingress_payload,
|
|
)
|
|
from cognitive_agent.application.seat_service import REPOSITORY_ROOT, SeatDemoService
|
|
|
|
|
|
def scene_card(*, temperature: int = 24) -> dict[str, object]:
|
|
return {
|
|
"id": "n7-reading",
|
|
"label": "N7 阅读模式",
|
|
"name": "N7 阅读模式",
|
|
"background_time": "evening",
|
|
"description": "由车端同步的阅读场景。",
|
|
"icon": "book-open",
|
|
"aliases": ["N7 阅读模式", "车端阅读"],
|
|
"origin": "user",
|
|
"deletable": True,
|
|
"behavior": "static",
|
|
"domains": ["climate"],
|
|
"operation_count": 1,
|
|
"scope": {"common": False, "vehicle_ids": ["denza_n7"]},
|
|
"profile_revision": f"remote-{temperature}",
|
|
"actions": [
|
|
{
|
|
"type": "climate_temperature",
|
|
"at_ms": 250,
|
|
"domain": "climate",
|
|
"capability": "climate.set_temperature",
|
|
"target_scope": "passenger",
|
|
"operation": "set",
|
|
"parameters": {"temperature_c": temperature},
|
|
"zone": "passenger",
|
|
"celsius": temperature,
|
|
}
|
|
],
|
|
}
|
|
|
|
|
|
def cinema_scene_card() -> dict[str, object]:
|
|
"""Representative N7 card containing local and head-unit-only actions."""
|
|
|
|
actions: list[dict[str, object]] = [
|
|
*[
|
|
{
|
|
"type": "window_position",
|
|
"at_ms": 0,
|
|
"domain": "window",
|
|
"capability": "window.set_position",
|
|
"target_scope": target,
|
|
"operation": "set",
|
|
"parameters": {"position_pct": 0},
|
|
}
|
|
for target in ("front_left", "front_right", "rear_left", "rear_right")
|
|
],
|
|
{
|
|
"type": "roof_shade_position",
|
|
"at_ms": 0,
|
|
"domain": "roof",
|
|
"capability": "roof.set_shade_position",
|
|
"target_scope": "vehicle",
|
|
"operation": "set",
|
|
"parameters": {"position_pct": 0},
|
|
},
|
|
{
|
|
"type": "seat_preset",
|
|
"at_ms": 0,
|
|
"domain": "local_vehicle",
|
|
"capability": "local_scene.seat_preset",
|
|
"target_scope": "vehicle",
|
|
"operation": "set",
|
|
"parameters": {"targets_pulse": [246, 112, 405, 342]},
|
|
},
|
|
{
|
|
"type": "seat_ventilation",
|
|
"at_ms": 0,
|
|
"domain": "seat",
|
|
"capability": "seat.set_ventilation",
|
|
"target_scope": "passenger",
|
|
"operation": "set",
|
|
"parameters": {"state": "off"},
|
|
},
|
|
{
|
|
"type": "climate_power",
|
|
"at_ms": 0,
|
|
"domain": "climate",
|
|
"capability": "climate.set_power",
|
|
"target_scope": "vehicle",
|
|
"operation": "set",
|
|
"parameters": {"state": "on"},
|
|
},
|
|
{
|
|
"type": "climate_fan",
|
|
"at_ms": 0,
|
|
"domain": "climate",
|
|
"capability": "climate.set_fan_level",
|
|
"target_scope": "vehicle",
|
|
"operation": "set",
|
|
"parameters": {"level": 2},
|
|
},
|
|
{
|
|
"type": "climate_outlet_mode",
|
|
"at_ms": 0,
|
|
"domain": "climate",
|
|
"capability": "climate.set_outlet_mode",
|
|
"target_scope": "vehicle",
|
|
"operation": "set",
|
|
"parameters": {"mode": "face"},
|
|
},
|
|
{
|
|
"type": "climate_cycle_mode",
|
|
"at_ms": 0,
|
|
"domain": "climate",
|
|
"capability": "climate.set_cycle_mode",
|
|
"target_scope": "vehicle",
|
|
"operation": "set",
|
|
"parameters": {"mode": "internal"},
|
|
},
|
|
{
|
|
"type": "climate_control_mode",
|
|
"at_ms": 0,
|
|
"domain": "climate",
|
|
"capability": "climate.set_control_mode",
|
|
"target_scope": "vehicle",
|
|
"operation": "set",
|
|
"parameters": {"mode": "manual"},
|
|
},
|
|
{
|
|
"type": "climate_temperature",
|
|
"at_ms": 0,
|
|
"domain": "climate",
|
|
"capability": "climate.set_temperature",
|
|
"target_scope": "vehicle",
|
|
"operation": "set",
|
|
"parameters": {"temperature_c": 25},
|
|
},
|
|
{
|
|
"type": "interior_light_reading_power",
|
|
"at_ms": 0,
|
|
"domain": "interior_light",
|
|
"capability": "interior_light.set_reading_power",
|
|
"target_scope": "all",
|
|
"operation": "set",
|
|
"parameters": {"state": "off"},
|
|
},
|
|
{
|
|
"type": "ambient_light_power",
|
|
"at_ms": 0,
|
|
"domain": "ambient_light",
|
|
"capability": "ambient_light.set_power",
|
|
"target_scope": "vehicle",
|
|
"operation": "set",
|
|
"parameters": {"state": "on"},
|
|
},
|
|
{
|
|
"type": "ambient_light_color",
|
|
"at_ms": 0,
|
|
"domain": "ambient_light",
|
|
"capability": "ambient_light.set_color_index",
|
|
"target_scope": "vehicle",
|
|
"operation": "set",
|
|
"parameters": {"color_index": 36},
|
|
},
|
|
{
|
|
"type": "ambient_light_brightness",
|
|
"at_ms": 0,
|
|
"domain": "ambient_light",
|
|
"capability": "ambient_light.set_brightness_level",
|
|
"target_scope": "vehicle",
|
|
"operation": "set",
|
|
"parameters": {"brightness_level": 4},
|
|
},
|
|
]
|
|
return {
|
|
"id": "cinema",
|
|
"label": "Cinema mode",
|
|
"name": "Cinema mode",
|
|
"background_time": "night",
|
|
"description": "Representative N7 mixed-domain scene.",
|
|
"icon": "bookmark-plus",
|
|
"aliases": ["Cinema mode"],
|
|
"origin": "headunit",
|
|
"deletable": True,
|
|
"behavior": "static",
|
|
"domains": list(dict.fromkeys(str(item["domain"]) for item in actions)),
|
|
"operation_count": len(actions),
|
|
"scope": {"common": False, "vehicle_ids": ["denza_n7"]},
|
|
"actions": actions,
|
|
"profile_revision": "cinema-test-revision",
|
|
}
|
|
|
|
|
|
def cinema_peripheral_catalog() -> list[dict[str, object]]:
|
|
def capability(
|
|
identity: str,
|
|
device_id: str,
|
|
device_name: str,
|
|
action: str,
|
|
slot: int,
|
|
) -> dict[str, object]:
|
|
return {
|
|
"id": identity,
|
|
"status": "learned",
|
|
"enabled": True,
|
|
"device_id": device_id,
|
|
"device_name": device_name,
|
|
"device_aliases": [device_name],
|
|
"object_names": [device_name],
|
|
"card_uid": "2CC8B103" if "flash" in device_id else "1A3AB103",
|
|
"action": action,
|
|
"aliases": [action],
|
|
"action_names": [action],
|
|
"risk": "low",
|
|
"slot": slot,
|
|
}
|
|
|
|
return [
|
|
capability("flash-on", "flash-light", "闪烁灯", "打开", 1),
|
|
capability("flash-blue", "flash-light", "闪烁灯", "切换为蓝色", 3),
|
|
capability("flash-red", "flash-light", "闪烁灯", "切换为红色", 4),
|
|
capability("strip-on", "light-strip", "灯带", "打开", 21),
|
|
capability("strip-blue", "light-strip", "灯带", "切换为蓝色", 23),
|
|
capability("strip-red", "light-strip", "灯带", "切换为红色", 24),
|
|
]
|
|
|
|
|
|
def cinema_video_launch_action() -> dict[str, object]:
|
|
return {
|
|
"type": "android_app_launch",
|
|
"at_ms": 3000,
|
|
"domain": "android",
|
|
"capability": "android.launch_app",
|
|
"target_scope": "headunit",
|
|
"operation": "launch",
|
|
"parameters": {
|
|
"activity_name": "com.byd.videoplay.activity.WelComeActivity",
|
|
"package_name": "com.byd.videoplay",
|
|
},
|
|
}
|
|
|
|
|
|
def wait_until(predicate, timeout: float = 1.5) -> bool:
|
|
deadline = time.monotonic() + timeout
|
|
while time.monotonic() < deadline:
|
|
if predicate():
|
|
return True
|
|
time.sleep(0.01)
|
|
return bool(predicate())
|
|
|
|
|
|
def configured_service(tmp_path: Path) -> SeatDemoService:
|
|
service = SeatDemoService(REPOSITORY_ROOT)
|
|
service.mode_profiles = ModeProfileRegistry(
|
|
tmp_path / "presets",
|
|
user_root=tmp_path / "profiles",
|
|
)
|
|
service.artifact_root = tmp_path / "runs"
|
|
return service
|
|
|
|
|
|
def test_parser_uses_exact_single_card_contract_and_rejects_duplicate_keys() -> None:
|
|
scene = scene_card()
|
|
parsed = parse_mode_scene_ingress_payload(
|
|
json.dumps(scene, ensure_ascii=False).encode("utf-8")
|
|
)
|
|
assert parsed == scene
|
|
|
|
duplicate = (
|
|
'{"id":"one","id":"two","label":"x","name":"x",'
|
|
'"background_time":"noon","description":"",'
|
|
'"icon":"x","aliases":[],"origin":"user",'
|
|
'"deletable":true,"behavior":"static","domains":["x"],'
|
|
'"operation_count":1,"scope":{},"profile_revision":"x",'
|
|
'"actions":[]}'
|
|
)
|
|
with pytest.raises(ModeSceneIngressError, match="duplicate key"):
|
|
parse_mode_scene_ingress_payload(duplicate)
|
|
|
|
|
|
def test_three_identical_frames_are_imported_once() -> None:
|
|
imports: list[dict[str, object]] = []
|
|
consumer = ModeSceneIngressConsumer(
|
|
lambda scene: imports.append(dict(scene)) or {"status": "imported"},
|
|
deduplication_window_s=1.0,
|
|
)
|
|
payload = json.dumps(scene_card(), ensure_ascii=False).encode("utf-8")
|
|
consumer.start()
|
|
try:
|
|
for _ in range(3):
|
|
consumer.submit(payload)
|
|
assert wait_until(lambda: consumer.snapshot()["processed_count"] == 3)
|
|
assert imports == [scene_card()]
|
|
status = consumer.snapshot()
|
|
assert status["imported_count"] == 1
|
|
assert status["duplicate_count"] == 2
|
|
assert status["rejected_count"] == 0
|
|
finally:
|
|
consumer.close()
|
|
|
|
|
|
def test_failed_import_does_not_suppress_identical_retry() -> None:
|
|
importer = MagicMock(
|
|
side_effect=[
|
|
RuntimeError("temporary persistence failure"),
|
|
{"status": "imported"},
|
|
]
|
|
)
|
|
consumer = ModeSceneIngressConsumer(
|
|
importer,
|
|
deduplication_window_s=1.0,
|
|
)
|
|
payload = json.dumps(scene_card(), ensure_ascii=False).encode("utf-8")
|
|
consumer.start()
|
|
try:
|
|
consumer.submit(payload)
|
|
assert wait_until(lambda: consumer.snapshot()["processed_count"] == 1)
|
|
first_status = consumer.snapshot()
|
|
assert first_status["rejected_count"] == 1
|
|
assert first_status["imported_count"] == 0
|
|
|
|
consumer.submit(payload)
|
|
assert wait_until(lambda: consumer.snapshot()["processed_count"] == 2)
|
|
retry_status = consumer.snapshot()
|
|
assert retry_status["rejected_count"] == 1
|
|
assert retry_status["imported_count"] == 1
|
|
assert retry_status["duplicate_count"] == 0
|
|
assert importer.call_count == 2
|
|
|
|
consumer.submit(payload)
|
|
assert wait_until(lambda: consumer.snapshot()["processed_count"] == 3)
|
|
final_status = consumer.snapshot()
|
|
assert final_status["duplicate_count"] == 1
|
|
assert importer.call_count == 2
|
|
finally:
|
|
consumer.close()
|
|
|
|
|
|
def test_import_upserts_yaml_refreshes_catalog_and_does_not_echo(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
service = configured_service(tmp_path)
|
|
service.configure_mode_scene_catalog(tmp_path / "scenes.json")
|
|
outbound = MagicMock()
|
|
service._vehicle_mqtt_transport.publish_mode_scene = outbound
|
|
|
|
first = service.import_n7_mode_scene(scene_card(temperature=24))
|
|
updated = service.import_n7_mode_scene(scene_card(temperature=26))
|
|
|
|
assert first["scene_id"] == "n7-reading"
|
|
assert updated["scene_id"] == "n7-reading"
|
|
profiles = service.mode_profiles.public_modes("denza_n7")
|
|
assert len(profiles) == 1
|
|
profile = profiles[0]
|
|
assert profile["origin"] == "n7_mqtt"
|
|
assert profile["source_topic"] == "from_n7/scenes"
|
|
assert profile["upstream_origin"] == "user"
|
|
assert profile["operations"][0]["parameters"] == {"temperature_c": 26}
|
|
|
|
catalog = service.mode_scene_catalog("denza_n7")
|
|
assert catalog["scene_count"] == 1
|
|
card = catalog["scenes"][0]
|
|
assert card["origin"] == "n7_mqtt"
|
|
assert card["source_topic"] == "from_n7/scenes"
|
|
assert card["actions"] == scene_card(temperature=26)["actions"]
|
|
assert card["actions"][0]["at_ms"] == 250
|
|
assert outbound.call_count == 0
|
|
|
|
yaml_path = Path(service.mode_profiles.get_profile("n7-reading", "denza_n7").source)
|
|
yaml_text = yaml_path.read_text(encoding="utf-8")
|
|
assert "sync_origin: from_n7/scenes" in yaml_text
|
|
assert "upstream_origin: user" in yaml_text
|
|
|
|
|
|
def test_later_n7_cinema_mode_inherits_base_and_adds_external_light_colors(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
service = configured_service(tmp_path)
|
|
service.set_mode_scene_peripheral_catalog_provider(cinema_peripheral_catalog)
|
|
base = cinema_scene_card()
|
|
base_action_count = len(base["actions"])
|
|
base_result = service.import_n7_mode_scene(base)
|
|
assert "cinema_augmentation" not in base_result
|
|
|
|
variant = cinema_scene_card()
|
|
variant.update(
|
|
id="blue-cinema",
|
|
label="蓝色影院模式",
|
|
name="蓝色影院模式",
|
|
aliases=["蓝色观影模式"],
|
|
description="在车机影院模式基础上增加蓝色外设灯光。",
|
|
domains=["climate"],
|
|
operation_count=1,
|
|
actions=[
|
|
{
|
|
"type": "climate_temperature",
|
|
"at_ms": 0,
|
|
"domain": "climate",
|
|
"capability": "climate.set_temperature",
|
|
"target_scope": "vehicle",
|
|
"operation": "set",
|
|
"parameters": {"temperature_c": 23},
|
|
}
|
|
],
|
|
profile_revision="blue-cinema-revision",
|
|
)
|
|
|
|
imported = service.import_n7_mode_scene(variant)
|
|
augmentation = imported["cinema_augmentation"]
|
|
assert augmentation == {
|
|
"base_scene_id": "cinema",
|
|
"base_action_count": base_action_count,
|
|
"incoming_action_count": 1,
|
|
"peripheral_action_count": 4,
|
|
"merged_action_count": base_action_count + 4,
|
|
}
|
|
|
|
profile = service.get_mode_profile_card("blue-cinema", "denza_n7")
|
|
source_scene = profile.command_flow["source_scene"]
|
|
actions = source_scene["actions"]
|
|
assert source_scene["operation_count"] == base_action_count + 4
|
|
assert "self_extension" in source_scene["domains"]
|
|
temperatures = [
|
|
action["parameters"]["temperature_c"]
|
|
for action in actions
|
|
if action["capability"] == "climate.set_temperature"
|
|
]
|
|
assert temperatures == [23]
|
|
peripheral_actions = [
|
|
action
|
|
for action in actions
|
|
if action["capability"] == "infrared.emit"
|
|
]
|
|
assert [
|
|
(
|
|
action["target_scope"],
|
|
action["parameters"]["action"],
|
|
action["parameters"]["slot"],
|
|
)
|
|
for action in peripheral_actions
|
|
] == [
|
|
("flash-light", "打开", 1),
|
|
("flash-light", "切换为蓝色", 3),
|
|
("light-strip", "打开", 21),
|
|
("light-strip", "切换为蓝色", 23),
|
|
]
|
|
|
|
repeated = service.import_n7_mode_scene(variant)
|
|
assert (
|
|
repeated["cinema_augmentation"]["merged_action_count"]
|
|
== base_action_count + 4
|
|
)
|
|
reloaded = service.get_mode_profile_card("blue-cinema", "denza_n7")
|
|
assert (
|
|
len(reloaded.command_flow["source_scene"]["actions"])
|
|
== base_action_count + 4
|
|
)
|
|
|
|
|
|
def test_n7_cinema_variant_is_not_augmented_before_base_arrives(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
service = configured_service(tmp_path)
|
|
service.set_mode_scene_peripheral_catalog_provider(cinema_peripheral_catalog)
|
|
variant = cinema_scene_card()
|
|
variant.update(
|
|
id="family-movie",
|
|
name="家庭影院模式",
|
|
label="家庭影院模式",
|
|
profile_revision="family-movie-revision",
|
|
)
|
|
|
|
imported = service.import_n7_mode_scene(variant)
|
|
|
|
assert "cinema_augmentation" not in imported
|
|
profile = service.get_mode_profile_card("family-movie", "denza_n7")
|
|
assert not any(
|
|
operation.capability == "infrared.emit"
|
|
for operation in profile.operations
|
|
)
|
|
|
|
|
|
def test_created_romantic_cinema_persists_and_transfers_n7_video_launch(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
service = configured_service(tmp_path)
|
|
base = cinema_scene_card()
|
|
base["actions"] = [*base["actions"], cinema_video_launch_action()]
|
|
base["operation_count"] = len(base["actions"])
|
|
base["domains"] = [*base["domains"], "android"]
|
|
service.import_n7_mode_scene(base)
|
|
|
|
request_id = "req-romantic-cinema-video"
|
|
instruction = "进入浪漫观影模式"
|
|
result_dir = service.artifact_root / request_id
|
|
result_dir.mkdir(parents=True)
|
|
(result_dir / "result.json").write_text(
|
|
json.dumps(
|
|
{
|
|
"request_id": request_id,
|
|
"status": "succeeded",
|
|
"execution_mode": "dry_run",
|
|
"control_transport": "wireless_mqtt",
|
|
"utterance": instruction,
|
|
"vehicle": {"id": "denza_n7"},
|
|
"function_configs": {
|
|
"climate": {
|
|
"spec": {
|
|
"domain": "climate",
|
|
"operations": [
|
|
{
|
|
"id": "vehicle-climate-power",
|
|
"capability": "climate.set_power",
|
|
"target": {"zone": "vehicle"},
|
|
"parameters": {"state": "on"},
|
|
}
|
|
],
|
|
}
|
|
}
|
|
},
|
|
"command_plans": {
|
|
"climate": {
|
|
"spec": {
|
|
"commands": [
|
|
{
|
|
"operation_ref": "vehicle-climate-power",
|
|
"message": "climate.set_power",
|
|
}
|
|
]
|
|
}
|
|
}
|
|
},
|
|
},
|
|
ensure_ascii=False,
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
|
|
saved = service.create_mode_profile_card(
|
|
"浪漫观影模式",
|
|
instruction,
|
|
request_id,
|
|
"denza_n7",
|
|
)
|
|
profile = service.get_mode_profile_card(
|
|
str(saved["mode_profile"]["id"]),
|
|
"denza_n7",
|
|
)
|
|
|
|
assert [
|
|
operation.capability for operation in profile.operations
|
|
] == ["climate.set_power", "android.launch_app"]
|
|
assert profile.command_flow["n7_transfer_actions"] == [
|
|
cinema_video_launch_action()
|
|
]
|
|
local_requests, transfer_actions = service._mode_requests_and_n7_transfers(
|
|
profile
|
|
)
|
|
assert [request["capability"] for request in local_requests] == [
|
|
"climate.set_power"
|
|
]
|
|
assert transfer_actions == [cinema_video_launch_action()]
|
|
|
|
exported = next(
|
|
scene
|
|
for scene in service.mode_scene_catalog("denza_n7")["scenes"]
|
|
if scene["id"] == profile.id
|
|
)
|
|
video = next(
|
|
action
|
|
for action in exported["actions"]
|
|
if action["capability"] == "android.launch_app"
|
|
)
|
|
assert video == cinema_video_launch_action()
|
|
|
|
|
|
def test_existing_local_cinema_mode_publishes_video_action_only_in_hardware(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
service = configured_service(tmp_path)
|
|
base = cinema_scene_card()
|
|
base["actions"] = [*base["actions"], cinema_video_launch_action()]
|
|
base["operation_count"] = len(base["actions"])
|
|
base["domains"] = [*base["domains"], "android"]
|
|
service.import_n7_mode_scene(base)
|
|
profile = service.mode_profiles.save_mode(
|
|
"浪漫观影模式",
|
|
[
|
|
{
|
|
"domain": "android",
|
|
"capability": "android.launch_app",
|
|
"target_scope": "headunit",
|
|
"operation": "launch",
|
|
"parameters": deepcopy(cinema_video_launch_action()["parameters"]),
|
|
}
|
|
],
|
|
vehicle_id="denza_n7",
|
|
instruction="进入浪漫观影模式",
|
|
command_flow={
|
|
"schema": "cognitive-agent-mode-command-flow/v1",
|
|
"plans": {"saved": {"spec": {"commands": [{"message": "saved"}]}}},
|
|
},
|
|
)
|
|
publisher = MagicMock(
|
|
return_value={
|
|
"topic": "transto_n7/ctl",
|
|
"qos": 1,
|
|
"retain": False,
|
|
"delivery_status": "published",
|
|
"message_id": 19,
|
|
}
|
|
)
|
|
service._vehicle_mqtt_transport.publish_n7_control_transfer = publisher
|
|
|
|
preview = service.execute_mode_profile_card(
|
|
profile.id,
|
|
execution_mode="dry_run",
|
|
control_transport="wireless_mqtt",
|
|
)
|
|
publisher.assert_not_called()
|
|
assert preview["n7_transfer_candidates"] == [cinema_video_launch_action()]
|
|
|
|
executed = service.execute_mode_profile_card(
|
|
profile.id,
|
|
execution_mode="hardware",
|
|
control_transport="wireless_mqtt",
|
|
)
|
|
|
|
assert executed["n7_control_transfer"]["state"] == "published"
|
|
publisher.assert_called_once()
|
|
payload = publisher.call_args.args[0]
|
|
assert payload["scene_id"] == profile.id
|
|
assert payload["scene_name"] == "浪漫观影模式"
|
|
assert payload["operation_count"] == 1
|
|
assert payload["actions"] == [cinema_video_launch_action()]
|
|
|
|
|
|
def test_unknown_n7_action_is_forwarded_once_only_during_hardware_execution(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
service = configured_service(tmp_path)
|
|
unknown = scene_card()
|
|
unknown.update(
|
|
domains=["media"],
|
|
operation_count=1,
|
|
actions=[
|
|
{
|
|
"type": "media_play_favorite",
|
|
"at_ms": 0,
|
|
"domain": "media",
|
|
"capability": "media.play_favorite",
|
|
"target_scope": "vehicle",
|
|
"operation": "play",
|
|
"parameters": {"playlist": "focus"},
|
|
}
|
|
],
|
|
)
|
|
service.import_n7_mode_scene(unknown)
|
|
publisher = MagicMock(
|
|
return_value={
|
|
"topic": "transto_n7/ctl",
|
|
"qos": 1,
|
|
"retain": False,
|
|
"delivery_status": "published",
|
|
"message_id": 7,
|
|
}
|
|
)
|
|
service._vehicle_mqtt_transport.publish_n7_control_transfer = publisher
|
|
|
|
dry_run = service.execute_mode_profile_card(
|
|
"n7-reading",
|
|
execution_mode="dry_run",
|
|
control_transport="wireless_mqtt",
|
|
)
|
|
assert dry_run["status"] == "capability_unavailable"
|
|
publisher.assert_not_called()
|
|
|
|
result = service.execute_mode_profile_card(
|
|
"n7-reading",
|
|
execution_mode="hardware",
|
|
control_transport="wireless_mqtt",
|
|
)
|
|
|
|
assert result["status"] == "capability_unavailable"
|
|
assert result["n7_control_transfer"]["state"] == "published"
|
|
publisher.assert_called_once()
|
|
payload = publisher.call_args.args[0]
|
|
assert payload == {
|
|
"scene_id": "n7-reading",
|
|
"scene_name": "N7 阅读模式",
|
|
"request_id": result["request_id"],
|
|
"operation_count": 1,
|
|
"actions": deepcopy(unknown["actions"]),
|
|
}
|
|
assert result["deferred_requests"][0]["transport_submitted"] is False
|
|
|
|
|
|
def test_mixed_n7_scene_compiles_every_local_action_and_preserves_transfer(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
service = configured_service(tmp_path)
|
|
scene = cinema_scene_card()
|
|
service.import_n7_mode_scene(scene)
|
|
|
|
result = service.execute_mode_profile_card(
|
|
"cinema",
|
|
execution_mode="dry_run",
|
|
control_transport="wireless_mqtt",
|
|
)
|
|
|
|
assert result["status"] == "partially_succeeded"
|
|
assert not [
|
|
issue
|
|
for issue in result["validation"]
|
|
if issue.get("severity") == "error"
|
|
]
|
|
compiled_operations = [
|
|
operation
|
|
for config in result["function_configs"].values()
|
|
for operation in config["spec"]["operations"]
|
|
]
|
|
assert len(compiled_operations) == 16
|
|
assert all(
|
|
operation.get("parse_source") == "saved_mode_command_flow"
|
|
for operation in compiled_operations
|
|
)
|
|
assert {
|
|
operation["target"]["zone"]
|
|
for operation in compiled_operations
|
|
if operation["capability"] == "window.set_position"
|
|
} == {"front_left", "front_right", "rear_left", "rear_right"}
|
|
transferred = scene["actions"][5]
|
|
assert result["n7_transfer_candidates"] == [transferred]
|
|
assert result["deferred_requests"] == [
|
|
{
|
|
"domain": "local_vehicle",
|
|
"capability": "local_scene.seat_preset",
|
|
"target_scope": "vehicle",
|
|
"desired_change": "set",
|
|
"parameters": {"targets_pulse": [246, 112, 405, 342]},
|
|
"reason": "当前 PC 未注册该能力，将由 N7 端继续处理。",
|
|
"execution_backend": "n7_vehicle_transfer",
|
|
"rejection_code": "PC_CAPABILITY_UNAVAILABLE",
|
|
"transport_submitted": False,
|
|
}
|
|
]
|
|
assert len(compiled_operations) + len(result["n7_transfer_candidates"]) == len(
|
|
scene["actions"]
|
|
)
|
|
|
|
text_result = service.process(
|
|
"进入Cinema mode模式",
|
|
provider="local",
|
|
conversation_id="conv-n7-cinema-text",
|
|
execution_mode="dry_run",
|
|
vehicle_id="denza_n7",
|
|
control_transport="wireless_mqtt",
|
|
)
|
|
assert text_result["status"] == "partially_succeeded"
|
|
assert text_result["mode_profile"] is None
|
|
assert [item["id"] for item in text_result["matched_modes"]] == ["cinema"]
|
|
assert len(text_result["n7_transfer_candidates"]) == 1
|
|
assert service.mode_profiles.match_text(
|
|
"氛围灯设置为Cinema mode模式",
|
|
vehicle_id="denza_n7",
|
|
) == []
|
|
|
|
catalog_mode = next(
|
|
mode
|
|
for mode in AssistantCapabilityCatalog(service).snapshot("denza_n7")["modes"]
|
|
if mode["id"] == "cinema"
|
|
)
|
|
assert catalog_mode["assistant_callable"] is True
|
|
assert catalog_mode["n7_transfer_operation_ids"] == [
|
|
"local_scene.seat_preset"
|
|
]
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("domain", "target_scope", "parameters"),
|
|
[
|
|
("media", "passenger", {"temperature_c": 24}),
|
|
("climate", "unsupported_zone", {"temperature_c": 24}),
|
|
("climate", "passenger", {"unsupported_parameter": 24}),
|
|
],
|
|
)
|
|
def test_registered_but_invalid_action_is_not_forwarded_to_n7(
|
|
tmp_path: Path,
|
|
domain: str,
|
|
target_scope: str,
|
|
parameters: dict[str, object],
|
|
) -> None:
|
|
service = configured_service(tmp_path)
|
|
invalid = scene_card()
|
|
invalid["actions"] = [
|
|
{
|
|
"type": "climate_temperature",
|
|
"at_ms": 0,
|
|
"domain": domain,
|
|
"capability": "climate.set_temperature",
|
|
"target_scope": target_scope,
|
|
"operation": "set",
|
|
"parameters": parameters,
|
|
}
|
|
]
|
|
service.import_n7_mode_scene(invalid)
|
|
profile = service.get_mode_profile_card("n7-reading", "denza_n7")
|
|
|
|
local_requests, transfer_actions = service._mode_requests_and_n7_transfers(profile)
|
|
|
|
assert len(local_requests) == 1
|
|
assert transfer_actions == []
|
|
|
|
|
|
def test_frontend_marks_n7_synchronized_cards() -> None:
|
|
script = (REPOSITORY_ROOT / "apps" / "web-console" / "app.js").read_text(
|
|
encoding="utf-8"
|
|
)
|
|
styles = (
|
|
REPOSITORY_ROOT / "apps" / "web-console" / "styles.css"
|
|
).read_text(encoding="utf-8")
|
|
assert 'mode?.origin === "n7_mqtt"' in script
|
|
assert 'sourceBadge.textContent = "N7 同步"' in script
|
|
assert ".mode-gallery-source-badge" in styles
|
|
|
|
|
|
def test_mode_profile_sync_status_exposes_catalog_and_ingress_counters(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
service = configured_service(tmp_path)
|
|
service.configure_mode_scene_catalog(tmp_path / "scenes.json")
|
|
|
|
status = service.mode_profile_sync_status()
|
|
|
|
assert len(status["catalog_revision"]) == 16
|
|
assert status["scene_count"] == 0
|
|
assert status["catalog_state"] == "ready"
|
|
assert status["mode_scene_ingress"]["imported_count"] == 0
|
|
|
|
|
|
def test_frontend_polls_lightweight_status_and_refreshes_on_n7_import() -> None:
|
|
script = (REPOSITORY_ROOT / "apps" / "web-console" / "app.js").read_text(
|
|
encoding="utf-8"
|
|
)
|
|
|
|
assert 'apiRequest("/api/mode-profiles/status"' in script
|
|
assert "MODE_PROFILE_SYNC_POLL_INTERVAL_MS = 1000" in script
|
|
assert 'document.visibilityState === "hidden"' in script
|
|
assert "if (catalogChanged || (imported && !sceneAlreadyVisible))" in script
|
|
assert "await loadRuntime()" in script
|
|
assert "revealModeProfile(sceneId)" in script
|
|
assert "已从 N7 同步模式卡片" in script
|
|
assert "N7 模式卡片导入失败" in script
|