Files
Cognitive_Agent/tests/unit/test_mode_gallery_grouping.py
Miayui c4efdbe22f merge: sync feature/N7Web into feature/N7Web-Tech
# Conflicts:
#	apps/web-console/service-worker.js
#	src/cognitive_agent/application/mode_profile.py
#	src/cognitive_agent/application/mode_scene_catalog.py
#	src/cognitive_agent/application/seat_service.py
#	tests/e2e/test_seat_web_api.py
#	tests/unit/test_assistant_frontend_contract.py
#	tests/unit/test_vue_frontend_contract.py
2026-09-17 11:16:48 +08:00

547 lines
20 KiB
Python

from __future__ import annotations
import json
import shutil
import subprocess
from pathlib import Path
import pytest
from cognitive_agent.application.mode_profile import ModeProfileRegistry
ROOT = Path(__file__).resolve().parents[2]
WEB = ROOT / "apps" / "web-console"
def test_runtime_mode_catalog_exposes_expected_static_and_dynamic_membership() -> None:
registry = ModeProfileRegistry(ROOT / "config" / "modes")
modes = registry.public_modes(vehicle_id="denza_n7")
grouped = {
behavior: {mode["id"] for mode in modes if mode["behavior"] == behavior}
for behavior in ("static", "cyclic")
}
assert grouped["static"] == {
"zero_gravity",
"office",
"nap",
"queen_passenger_driving",
"leisure_lounger",
"cruise_sofa",
"easy_entry_exit",
"electric_beach_chair",
"calf_decompression_float",
"baby_nap",
"meditation_space",
}
assert grouped["cyclic"] == {"rocking_horse", "dynamic_cradle"}
assert len(modes) == len(grouped["static"] | grouped["cyclic"])
def test_runtime_mode_catalog_exposes_safe_executable_action_details() -> None:
registry = ModeProfileRegistry(ROOT / "config" / "modes")
office = next(
mode
for mode in registry.public_modes(vehicle_id="denza_n7")
if mode["id"] == "office"
)
assert len(office["operations"]) == office["operation_count"] == 12
assert office["operations"][0] == {
"domain": "seat",
"capability": "seat.set_driver_ventilation_native",
"target_scope": "driver",
"operation": "set",
"parameters": {"state": "on", "level": 2},
}
assert next(
operation for operation in office["operations"]
if operation["capability"] == "seat.set_longitudinal_position"
) == {
"domain": "seat",
"capability": "seat.set_longitudinal_position",
"target_scope": "passenger",
"operation": "set",
"parameters": {"position_pct": 72},
}
assert all(
set(operation) == {
"domain",
"capability",
"target_scope",
"operation",
"parameters",
}
for operation in office["operations"]
)
assert all("reason" not in operation for operation in office["operations"])
def node_runtime() -> str:
candidates = [
shutil.which("node"),
Path.home()
/ ".cache"
/ "codex-runtimes"
/ "codex-primary-runtime"
/ "dependencies"
/ "node"
/ "bin"
/ "node.exe",
Path.home()
/ ".cache"
/ "codex-runtimes"
/ "codex-primary-runtime"
/ "dependencies"
/ "node"
/ "bin"
/ "node",
]
for candidate in candidates:
if candidate and Path(candidate).is_file():
return str(candidate)
raise AssertionError("Node.js is required to execute the mode gallery grouping contract")
def test_mode_gallery_classifies_profiles_into_exclusive_custom_static_dynamic_groups() -> None:
app = (WEB / "app.js").read_text(encoding="utf-8")
start = app.index("const MODE_GALLERY_GROUPS")
end = app.index("\n\nfunction createModeGalleryCard", start)
snippet = app[start:end]
script = "const snippet = " + json.dumps(snippet) + ";\n" + r"""
const vm = require("node:vm");
const modes = [
{ id: "user-static", origin: "user", deletable: true, behavior: "static" },
{ id: "office", origin: "preset", behavior: "static" },
{ id: "future-cycle", origin: "user", deletable: true, behavior: "cyclic" },
{ id: "timeline-only", origin: "preset", timeline: { repeat: "once" } },
{ id: "rocking_horse", origin: "preset" },
{ id: "unknown", origin: "preset" },
{ id: "future-behavior", origin: "preset", behavior: "scheduled", timeline: { repeat: "loop" } },
];
const context = { state: { runtime: { modes } } };
vm.createContext(context);
vm.runInContext(
"function modeProfiles() { return state.runtime.modes; }\n"
+ snippet
+ "\nthis.result = {"
+ "groups: modeGalleryBuckets().map((group) => ({ id: group.id, title: group.title, modeIds: group.modes.map((mode) => mode.id) })),"
+ "dynamicFlags: state.runtime.modes.map((mode) => isDynamicModeProfile(mode))"
+ "};",
context,
);
console.log(JSON.stringify(context.result));
"""
completed = subprocess.run(
[node_runtime(), "-e", script],
check=True,
capture_output=True,
text=True,
encoding="utf-8",
)
result = json.loads(completed.stdout)
assert [(group["id"], group["title"]) for group in result["groups"]] == [
("custom", "自定义模式"),
("static", "静态模式"),
("dynamic", "动态模式"),
]
assert result["groups"][0]["modeIds"] == ["user-static", "future-cycle"]
assert result["groups"][1]["modeIds"] == ["office", "unknown"]
assert result["groups"][2]["modeIds"] == [
"rocking_horse",
"timeline-only",
"future-behavior",
]
assert result["dynamicFlags"] == [False, False, True, True, True, False, True]
def test_mode_gallery_layout_and_copy_expose_three_management_sections() -> None:
app = (WEB / "app.js").read_text(encoding="utf-8")
template = (WEB / "src" / "App.vue").read_text(encoding="utf-8")
styles = (WEB / "styles.css").read_text(encoding="utf-8")
definitions = app[
app.index("const MODE_GALLERY_GROUPS")
: app.index("const MODE_DOMAIN_LABELS")
]
assert definitions.count('title: "自定义模式"') == 1
assert definitions.count('title: "静态模式"') == 1
assert definitions.count('title: "动态模式"') == 1
for legacy_group in ('id: "personal"', 'id: "frequent"', 'id: "comfort"'):
assert legacy_group not in definitions
assert 'if (group.id === "custom") grid.append(createModeGalleryAddCard())' in app
assert "function createModeGalleryEmpty(group)" in app
assert 'section.setAttribute("aria-labelledby", headingId)' in app
assert "自定义模式 · 静态模式 · 动态模式" in template
assert "按来源与运行方式整理,快速定位并管理模式" in template
assert "grid-template-columns: repeat(auto-fit, minmax(220px, 1fr));" in styles
def test_top_mode_cards_render_real_actions_in_a_body_level_hover_and_focus_preview() -> None:
app = (WEB / "app.js").read_text(encoding="utf-8")
styles = (WEB / "console-v2.css").read_text(encoding="utf-8")
assert "function modeExecutionActionViews(mode)" in app
assert "Array.isArray(mode?.operations)" in app
assert "function renderModeActionPopover(mode)" in app
assert "modeEditorReadableCommands(mode?.command_flow)" in app
assert 'eyebrow.dataset.modeActionEyebrow = "true"' in app
assert '"最新控制指令流"' in app
assert 'popover.setAttribute("role", "tooltip")' in app
assert "document.body.append(popover)" in app
assert 'appendMetadata("语义", action.semantic || action.operation_ref)' in app
assert 'appendMetadata("实际", action.command)' in app
assert 'appendMetadata("通道", action.channel)' in app
assert 'appendMetadata("设备", action.device)' in app
assert 'appendMetadata("目标", action.target)' in app
assert 'appendMetadata("参数", action.parameters)' in app
assert 'button.setAttribute("aria-describedby", popover.id)' in app
assert 'group.addEventListener("pointerover"' in app
assert 'group.addEventListener("pointerout"' in app
assert 'group.addEventListener("focusin"' in app
assert 'group.addEventListener("focusout"' in app
assert 'group.addEventListener("scroll", () => hideModeActionPopover({ immediate: true })' in app
assert 'window.addEventListener("resize", () => hideModeActionPopover({ immediate: true })' in app
assert "installModeActionPreview();" in app
assert ".mode-chip-action-popover" in styles
popover_styles = styles.split(".mode-chip-action-popover {", 1)[1].split("}", 1)[0]
assert "position: fixed" in popover_styles
assert "max-height:" in popover_styles
assert "overflow: hidden" in popover_styles
assert ".mode-chip-action-popover ol" in styles
assert "overflow-y: auto" in styles.split(".mode-chip-action-popover ol", 1)[1].split("}", 1)[0]
def test_top_mode_action_formatter_uses_device_target_action_and_parameter_values() -> None:
registry = ModeProfileRegistry(ROOT / "config" / "modes")
office = next(
mode
for mode in registry.public_modes(vehicle_id="denza_n7")
if mode["id"] == "office"
)
app = (WEB / "app.js").read_text(encoding="utf-8")
start = app.index("const MODE_TARGET_LABELS")
end = app.index("\n\nfunction isUserModeProfile", start)
snippet = app[start:end]
script = (
'const MODE_DOMAIN_LABELS = { seat: "座椅", climate: "空调", roof: "天幕", '
'ambient_light: "氛围灯" };\n'
'const CONFIG_PARAMETER_LABELS = { position_pct: "目标位置", '
'temperature_c: "目标温度", level: "档位", mode: "工作模式", '
'shade_position_pct: "遮阳帘位置", brightness_pct: "目标亮度", '
'color_hex: "目标颜色" };\n'
'const CONFIG_CAPABILITY_LABELS = { '
'"seat.set_longitudinal_position": "调节座椅前后位置", '
'"climate.set_temperature": "设置空调温度" };\n'
'function formatConfigValue(name, value) { '
'if (name === "temperature_c") return `${value} ℃`; '
'if (name.endsWith("_pct")) return `${value}%`; return String(value); }\n'
'function modeSecondsLabel(value) { return String(Number(value) / 1000); }\n'
+ snippet
+ "\nconst mode = "
+ json.dumps(office, ensure_ascii=False)
+ "; console.log(JSON.stringify(modeExecutionActionViews(mode)));"
)
completed = subprocess.run(
[node_runtime(), "-e", script],
check=True,
capture_output=True,
text=True,
encoding="utf-8",
)
actions = json.loads(completed.stdout)
assert len(actions) == office["operation_count"]
assert next(action for action in actions if action["action"] == "调节座椅前后位置") == {
"device": "座椅",
"target": "副驾",
"action": "调节座椅前后位置",
"parameters": "目标位置:72%",
}
temperature = next(action for action in actions if action["action"] == "设置空调温度")
assert temperature == {
"device": "空调",
"target": "副驾",
"action": "设置空调温度",
"parameters": "目标温度:25 ℃",
}
def test_mode_chip_scroller_keeps_clicks_and_only_captures_real_drags() -> None:
app = (WEB / "app.js").read_text(encoding="utf-8")
start = app.index("function installModeChipScroller")
end = app.index("\n\nfunction ensureModeActionPopover", start)
snippet = app[start:end]
threshold = snippet.index('if (!dragged && Math.abs(distance) < 5) return;')
capture = snippet.index("group.setPointerCapture(pointerId);")
assert capture > threshold
assert "let pointerCaptured = false;" in snippet
assert "if (pointerCaptured) group.releasePointerCapture?.(pointerId);" in snippet
assert 'event.stopImmediatePropagation();' in snippet
def test_top_mode_triangle_is_an_explicit_execution_target() -> None:
app = (WEB / "app.js").read_text(encoding="utf-8")
vue_ui = (WEB / "vue-ui.js").read_text(encoding="utf-8")
handlers = app[app.index('document.addEventListener("DOMContentLoaded"') :]
assert '"data-mode-run": "true"' in vue_ui
assert 'runIcon.dataset.modeRun = "true";' in app
assert 'const runControl = target?.closest("[data-mode-run]");' in handlers
assert '(runControl || target)?.closest("button[data-mode-command]")' in handlers
def test_saved_mode_cards_open_assistant_and_execute_the_saved_command_flow() -> None:
app = (WEB / "app.js").read_text(encoding="utf-8")
start = app.index("function applyModeCommand")
end = app.index("\n\nfunction applyRuntimeTargetPolicy", start)
snippet = app[start:end]
script = "const snippet = " + json.dumps(snippet) + ";\n" + r"""
const vm = require("node:vm");
const calls = [];
const runButton = { disabled: false };
const savedMode = {
id: "reading-user",
name: "阅读模式",
command_flow: { source_request_id: "saved-request" },
};
const context = {
state: { pendingModeProfileId: null },
Event: class Event {
constructor(type, options = {}) { this.type = type; this.bubbles = options.bubbles; }
},
window: {
CognitiveAgentAssistant: {
isPending() { return false; },
open() { calls.push(["open"]); },
},
},
modeProfiles() { return [savedMode]; },
setModeRailSelection(mode) { calls.push(["select", mode?.id]); },
closeModeGallery(options) { calls.push(["close", options.restoreFocus]); },
hideModeActionPopover() {},
showToast(message) { calls.push(["toast", message]); },
hasExecutableModeCommandFlow(mode) {
const flow = mode?.command_flow;
return Boolean(flow?.source_request_id)
|| Boolean(flow?.plans && Object.keys(flow.plans).length);
},
executeSavedModeProfile(mode, options) {
calls.push(["execute", mode.id, options.command]);
return Promise.resolve(true);
},
$(selector) {
return { "#runButton": runButton }[selector];
},
};
vm.createContext(context);
vm.runInContext(snippet + '\nthis.result = applyModeCommand(" 进入阅读模式 ", "reading-user");', context);
console.log(JSON.stringify({
result: context.result,
calls,
pendingModeProfileId: context.state.pendingModeProfileId,
}));
"""
completed = subprocess.run(
[node_runtime(), "-e", script],
check=True,
capture_output=True,
text=True,
encoding="utf-8",
)
result = json.loads(completed.stdout)
assert result["result"] is True
assert result["calls"] == [
["select", "reading-user"],
["close", False],
["open"],
["execute", "reading-user", "进入阅读模式"],
]
assert result["pendingModeProfileId"] is None
assert "executeSavedModeProfile" in snippet
def test_n7_mode_command_flow_is_directly_executable_without_source_request_id() -> None:
app = (WEB / "app.js").read_text(encoding="utf-8")
start = app.index("function hasExecutableModeCommandFlow")
end = app.index("\n\nasync function executeSavedModeProfile", start)
helper = app[start:end]
script = helper + r"""
console.log(JSON.stringify({
imported: hasExecutableModeCommandFlow({
id: "cinema",
command_flow: {
source_topic: "from_n7/scenes",
plans: { from_n7_scene: { spec: { commands: [] } } },
},
}),
generated: hasExecutableModeCommandFlow({
id: "reading-user",
command_flow: { source_request_id: "req-reading" },
}),
missing: hasExecutableModeCommandFlow({ id: "office" }),
}));
"""
completed = subprocess.run(
[node_runtime(), "-e", script],
check=True,
capture_output=True,
text=True,
encoding="utf-8",
)
assert json.loads(completed.stdout) == {
"imported": True,
"generated": True,
"missing": False,
}
def test_mode_card_without_saved_flow_falls_back_to_assistant_submission() -> None:
app = (WEB / "app.js").read_text(encoding="utf-8")
start = app.index("function applyModeCommand")
end = app.index("\n\nfunction applyRuntimeTargetPolicy", start)
snippet = app[start:end]
script = "const snippet = " + json.dumps(snippet) + ";\n" + r"""
const vm = require("node:vm");
const calls = [];
const input = {
value: "",
dataset: { speechStage: "final", speechUtteranceId: "old", speechValue: "old" },
dispatchEvent(event) { calls.push(["input", event.type, event.bubbles, this.value]); },
};
const form = { requestSubmit() { calls.push(["submit", input.value]); } };
const runButton = { disabled: false };
const unsavedMode = { id: "future-mode", name: "未来模式" };
const context = {
state: { pendingModeProfileId: null },
Event: class Event {
constructor(type, options = {}) { this.type = type; this.bubbles = options.bubbles; }
},
window: {
CognitiveAgentAssistant: {
isPending() { return false; },
open() { calls.push(["open"]); },
},
},
modeProfiles() { return [unsavedMode]; },
setModeRailSelection(mode) { calls.push(["select", mode?.id]); },
setModeExecutionState(modeId, status) { calls.push(["state", modeId, status]); },
closeModeGallery(options) { calls.push(["close", options.restoreFocus]); },
showToast(message) { calls.push(["toast", message]); },
hasExecutableModeCommandFlow(mode) {
const flow = mode?.command_flow;
return Boolean(flow?.source_request_id)
|| Boolean(flow?.plans && Object.keys(flow.plans).length);
},
$(selector) {
return { "#utterance": input, "#requestForm": form, "#runButton": runButton }[selector];
},
};
vm.createContext(context);
vm.runInContext(snippet + '\nthis.result = applyModeCommand(" 进入未来模式 ", "future-mode");', context);
console.log(JSON.stringify({ result: context.result, calls, value: input.value }));
"""
completed = subprocess.run(
[node_runtime(), "-e", script],
check=True,
capture_output=True,
text=True,
encoding="utf-8",
)
result = json.loads(completed.stdout)
assert result["result"] is True
assert result["calls"] == [
["select", "future-mode"],
["close", False],
["open"],
["input", "input", True, "进入未来模式"],
["state", "future-mode", "running"],
["submit", "进入未来模式"],
]
assert result["value"] == "进入未来模式"
def test_mode_cards_do_not_submit_while_assistant_is_busy() -> None:
app = (WEB / "app.js").read_text(encoding="utf-8")
start = app.index("function applyModeCommand")
end = app.index("\n\nfunction applyRuntimeTargetPolicy", start)
snippet = app[start:end]
script = "const snippet = " + json.dumps(snippet) + ";\n" + r"""
const vm = require("node:vm");
const calls = [];
const context = {
window: { CognitiveAgentAssistant: { isPending() { return true; } } },
showToast(message) { calls.push(message); },
};
vm.createContext(context);
vm.runInContext(snippet + '\nthis.result = applyModeCommand("进入办公模式", "office");', context);
console.log(JSON.stringify({ result: context.result, calls }));
"""
completed = subprocess.run(
[node_runtime(), "-e", script],
check=True,
capture_output=True,
text=True,
encoding="utf-8",
)
result = json.loads(completed.stdout)
assert result == {
"result": False,
"calls": ["AI助手正在处理上一条指令,请稍候"],
}
@pytest.mark.parametrize(("phase", "input_mode"), (
("capturing", "manual"),
("capturing", "wake"),
("finalizing", "manual"),
("finalizing", "wake"),
("arming", "manual"),
))
def test_mode_click_during_dictation_does_not_create_stuck_running_state(
phase: str, input_mode: str,
) -> None:
app = (WEB / "app.js").read_text(encoding="utf-8")
start = app.index("function applyModeCommand")
end = app.index("\n\nfunction applyRuntimeTargetPolicy", start)
script = "const snippet = " + json.dumps(app[start:end]) + ";\n"
script += "const speech = " + json.dumps({"phase": phase, "inputMode": input_mode}) + ";\n"
script += r"""
const vm = require("node:vm");
const calls = [];
const context = {
window: {
CognitiveAgentAssistant: { isPending() { return false; } },
CognitiveAgentSpeech: speech,
},
showToast(message) { calls.push(message); },
};
vm.createContext(context);
vm.runInContext(snippet + '\nthis.result = applyModeCommand("进入办公模式", "office");', context);
console.log(JSON.stringify({ result: context.result, calls }));
"""
completed = subprocess.run(
[node_runtime(), "-e", script], check=True, capture_output=True, encoding="utf-8",
)
assert json.loads(completed.stdout) == {
"result": False,
"calls": ["请等待语音整理完成后再发送"],
}
def test_top_and_management_mode_cards_share_the_assistant_submission_entrypoint() -> None:
app = (WEB / "app.js").read_text(encoding="utf-8")
install_start = app.index('document.addEventListener("DOMContentLoaded"')
handlers = app[install_start:]
assert handlers.count("if (button) applyModeCommand(") == 2
assert '$(".mode-chips").addEventListener("click"' in handlers
assert '$("#modeGalleryGroups").addEventListener("click"' in handlers
assert app.count('button.dataset.modeCommand = `进入${mode.name}`;') == 2