feat(template): 升级 desktop E2E 模板到本系列最佳实践
基于 disburse-payroll-icbc 实战经验沉淀,让所有未来新 skill 复制模板后直接 享受本系列全部改进: - record_screencast.py.sample: · 字幕脚本改用 "step: N/X" 模式 + ASCII 关键词锚(避免编码问题) · 展示三元组 (keyword, text, duration) 用法(依赖 platform-kit subtitle.py) · 加 TEST_FILE_NAME / MUSIC_SUBDIR 切换常量 · 文档说明绕开 SDK ask() / send_file 的最佳实践 - test_desktop_smoke.py.sample: · docstring 加"何时用此模板 / 何时改用 _with_attachment"指引 · 标注已知 SDK 限制(ask() user_idx<0 / send_file NotImplementedError) - test_desktop_smoke_with_attachment.py.sample(新增): · 完整附件场景模板,import jiangchang_desktop_sdk.e2e_helpers 6 个 API · 7 步 step: N/7 模式,跟 record_screencast.py.sample 字幕脚本对齐 · 占位符 SKILL_SLUG / PROMPT / EXPECTED_KEYWORDS_IN_REPLY - README.md: 追加 3 节 · 模板选择指南(纯问答 vs 附件场景) · 已知 SDK 坑(send_file / ask / 字幕关键词 / bring_to_front) · 录屏新能力(activate_window / monitor_index / 字幕三元组) - conftest.py: 同步到支持新版 platform-kit src/ 布局的 SDK 路径推断 Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com> Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
@@ -32,3 +32,32 @@ pytest tests/desktop/test_desktop_smoke.py -v -s
|
||||
- **桌面 E2E**:仅在本目录用 **pytest** 维护,二者互补、互不替代。
|
||||
|
||||
若仅需验证 **service 业务逻辑**、契约与 golden 回归,**不要**写 desktop E2E;desktop 只用于验证**宿主 UI、路由、`jiangchang://` 协议、真实用户交互链路**等与桌面宿主强相关的行为。
|
||||
|
||||
## 模板选择指南
|
||||
|
||||
| 场景 | 用哪个模板 |
|
||||
|---|---|
|
||||
| 纯问答 skill(不上传附件、不跑长流程) | `test_desktop_smoke.py.sample` |
|
||||
| 需要上传附件 / RPA 长流程 / Agent 多轮工具调用 | `test_desktop_smoke_with_attachment.py.sample` |
|
||||
|
||||
录屏入口都用 `record_screencast.py.sample`,里面 `TEST_FILE_NAME` 常量切换测试文件。
|
||||
|
||||
## 已知 SDK 坑(来自 disburse-payroll-icbc 实战)
|
||||
|
||||
1. **`client.send_file()` 在 v2.0.17+ 抛 NotImplementedError**——附件场景必须用
|
||||
`jiangchang_desktop_sdk.e2e_helpers.drop_file_into_composer`(DataTransfer 拖拽)。
|
||||
2. **`client.ask()` 内部 `user_msg_index<0` 死循环**——如果匠厂启动慢 / gateway 握手有
|
||||
延迟,导致 15s 内没抓到 user 节点,ask() 会锁死后续 900s 等待。建议长流程 / 附件场景
|
||||
绕开 ask(),用 `jiangchang_desktop_sdk.e2e_helpers` 里的 `send_prompt_via_composer`
|
||||
+ `wait_for_user_message` + `wait_for_agent_complete` 自己控时。
|
||||
3. **字幕关键词建议用 ASCII**——虽然 platform-kit 已修了 subprocess 编码乱码,
|
||||
保险起见字幕脚本里关键词用 `step: 1/` 这种 ASCII 锚而非中文短语。
|
||||
4. **`client.bring_to_front()` 已升级**——v?? 起会 ctypes ShowWindow(SW_MAXIMIZE) +
|
||||
SetForegroundWindow,能真正铺满桌面。dev 模式下也能用(不依赖 jiangchang:// 协议)。
|
||||
|
||||
## 录屏新能力(platform-kit v??)
|
||||
|
||||
- `run_screencast(..., activate_window=True)` 自动激活+最大化匠厂窗口
|
||||
- `run_screencast(..., monitor_index=1)` 多显示器场景显式录主屏
|
||||
- 字幕脚本支持三元组 `(keyword, text, duration)` 给个别字幕指定独立时长,
|
||||
覆盖长等待场景中间没日志输出的"字幕真空"
|
||||
|
||||
@@ -2,11 +2,13 @@
|
||||
"""
|
||||
可选桌面 E2E:pytest 收集时自动加载,负责把 jiangchang_desktop_sdk 加入 sys.path。
|
||||
|
||||
查找顺序(与 content-manager 对齐思路,去掉业务仓专属路径假设):
|
||||
查找顺序:
|
||||
1. 已 pip install 则直接 import;
|
||||
2. 环境变量 JIANGCHANG_KIT_ROOT → sdk/jiangchang-desktop-sdk/src;
|
||||
3. 自本文件上溯:skill 根 → workspace 根 → OpenClaw 根,拼 jiangchang-platform-kit/sdk/.../src;
|
||||
4. 都找不到则 pytest.exit 给出可操作的排障说明。
|
||||
2. 环境变量 JIANGCHANG_KIT_ROOT → sdk/jiangchang-desktop-sdk/src(旧版布局);
|
||||
3. 环境变量 JIANGCHANG_KIT_ROOT → src(新版 platform-kit 布局);
|
||||
4. 自本文件上溯:skill 根 → workspace 根 → OpenClaw 根,拼路径;
|
||||
5. 兜底 D:\\AI\\jiangchang-platform-kit\\src(若存在);
|
||||
6. 都找不到则 pytest.exit 给出可操作的排障说明。
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
@@ -31,12 +33,22 @@ def _bootstrap_jiangchang_desktop_sdk() -> None:
|
||||
candidates = []
|
||||
if env_kit_root:
|
||||
candidates.append(os.path.join(env_kit_root, "sdk", "jiangchang-desktop-sdk", "src"))
|
||||
candidates.append(os.path.join(env_kit_root, "src"))
|
||||
candidates.append(
|
||||
os.path.join(workspace_root, "jiangchang-platform-kit", "sdk", "jiangchang-desktop-sdk", "src")
|
||||
)
|
||||
candidates.append(
|
||||
os.path.join(open_claw_root, "jiangchang-platform-kit", "sdk", "jiangchang-desktop-sdk", "src")
|
||||
)
|
||||
candidates.append(
|
||||
os.path.join(workspace_root, "jiangchang-platform-kit", "src")
|
||||
)
|
||||
candidates.append(
|
||||
os.path.join(open_claw_root, "jiangchang-platform-kit", "src")
|
||||
)
|
||||
default_kit_src = r"D:\AI\jiangchang-platform-kit\src"
|
||||
if os.path.isdir(default_kit_src):
|
||||
candidates.append(default_kit_src)
|
||||
|
||||
tried = []
|
||||
for cand in candidates:
|
||||
@@ -49,10 +61,10 @@ def _bootstrap_jiangchang_desktop_sdk() -> None:
|
||||
|
||||
pytest.exit(
|
||||
"无法定位 jiangchang_desktop_sdk。请任选一种方式解决:\n"
|
||||
" 1) pip install -e <path-to>/jiangchang-platform-kit/sdk/jiangchang-desktop-sdk\n"
|
||||
" 1) pip install -e <path-to>/jiangchang-platform-kit\n"
|
||||
" 2) 设置环境变量 JIANGCHANG_KIT_ROOT=<path-to>/jiangchang-platform-kit\n"
|
||||
" 3) 将 jiangchang-platform-kit 与本仓库放在同一 workspace 父级可推断的位置,例如:\n"
|
||||
f" {workspace_root}/jiangchang-platform-kit/ 或 {open_claw_root}/jiangchang-platform-kit/\n"
|
||||
" (SDK 新版位置:<root>\\src;旧版:<root>\\sdk\\jiangchang-desktop-sdk\\src)\n"
|
||||
" 3) 将 jiangchang-platform-kit 与本仓库放在同一 workspace 父级可推断的位置\n"
|
||||
"已尝试的候选路径:\n - " + "\n - ".join(tried),
|
||||
returncode=4,
|
||||
)
|
||||
|
||||
@@ -3,14 +3,21 @@
|
||||
Screencast 录制入口(模板)。
|
||||
|
||||
启用方式:
|
||||
复制为 record_screencast.py,修改 SKILL_SLUG 和 SUBTITLE_SCRIPT,然后运行:
|
||||
python tests/desktop/record_screencast.py
|
||||
1. 复制为 record_screencast.py
|
||||
2. 修改 SKILL_SLUG 和 SUBTITLE_SCRIPT
|
||||
3. 决定 test_file 指向哪个测试(默认 test_desktop_smoke.py;
|
||||
有附件场景的用 test_desktop_smoke_with_attachment.py)
|
||||
4. 运行:python tests/desktop/record_screencast.py
|
||||
|
||||
依赖:
|
||||
1. pip install mss
|
||||
2. 系统已安装 ffmpeg 并在 PATH 中
|
||||
3. jiangchang-platform-kit 在本机可推断路径(见 _bootstrap 逻辑)
|
||||
3. jiangchang-platform-kit 可推断路径(见 _bootstrap 逻辑)
|
||||
4. MEDIA_ASSETS_ROOT 环境变量指向 media-assets 仓库(或使用默认路径)
|
||||
|
||||
平台 kit v?? 之后 run_screencast 自动激活并最大化匠厂窗口、支持指定显示器、
|
||||
字幕支持每条独立时长。本模板已经利用这些能力,新 skill 复制后无需再内嵌
|
||||
ctypes 激活代码。
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
@@ -22,9 +29,9 @@ from pathlib import Path
|
||||
def _bootstrap_screencast() -> None:
|
||||
"""将 jiangchang-platform-kit/tools 加入 sys.path,使 screencast 包可导入。"""
|
||||
here = Path(__file__).resolve()
|
||||
skill_root = here.parents[2] # tests/desktop → tests → skill_root
|
||||
workspace_root = skill_root.parent # skill_root → workspace
|
||||
open_claw_root = workspace_root.parent # workspace → OpenClaw
|
||||
skill_root = here.parents[2]
|
||||
workspace_root = skill_root.parent
|
||||
open_claw_root = workspace_root.parent
|
||||
|
||||
candidates = [
|
||||
workspace_root / "jiangchang-platform-kit" / "tools",
|
||||
@@ -43,7 +50,8 @@ def _bootstrap_screencast() -> None:
|
||||
|
||||
raise RuntimeError(
|
||||
"找不到 jiangchang-platform-kit/tools/screencast。\n"
|
||||
"请设置 JIANGCHANG_KIT_ROOT 环境变量,或将 platform-kit 放在 workspace 同级目录。"
|
||||
"请设置 JIANGCHANG_KIT_ROOT 环境变量,或将 platform-kit 放在 workspace "
|
||||
"同级目录。"
|
||||
)
|
||||
|
||||
|
||||
@@ -52,25 +60,48 @@ _bootstrap_screencast()
|
||||
from screencast import run_screencast # noqa: E402
|
||||
|
||||
# ============================================================
|
||||
# ↓↓↓ 复制后只需修改这两处 ↓↓↓
|
||||
# ↓↓↓ 复制后只需修改下面这些 ↓↓↓
|
||||
# ============================================================
|
||||
|
||||
SKILL_SLUG = "your-skill-slug" # 替换为本技能 slug
|
||||
|
||||
# 字幕脚本设计要点(来自 disburse-payroll-icbc 实战经验):
|
||||
# 1. **关键词全用 ASCII**:保险起见绕开 Windows cp936/UTF-8 编码偶发问题。
|
||||
# 虽然 platform-kit 已修了 subprocess 编码,ASCII 关键词仍是最稳的。
|
||||
# 2. **用 "step: N/X" 模式作触发锚**:在测试代码里加 _logger.info("step: 1/5 ...")
|
||||
# 日志作为字幕触发点,能精确控制字幕出现时机,比依赖 SDK 内部日志更可靠。
|
||||
# 3. **关键词每条只触发一次**,所以"step: 1/" 这种唯一前缀比"准备数据"更稳。
|
||||
# 4. **三元组支持每条独立时长**:(关键词, 文案, 显示秒数);
|
||||
# long wait 场景可以让某条字幕停留更久(如 30s)覆盖中间无日志输出的等待期。
|
||||
SUBTITLE_SCRIPT = [
|
||||
# (pytest stdout 中出现的关键词, 字幕显示文案)
|
||||
("ensure_app_running", "正在启动匠厂智能体平台..."),
|
||||
("wait_gateway_ready", "网关就绪,准备发起任务"),
|
||||
("ask(", "向助手发送指令..."),
|
||||
("PASSED", "任务完成"),
|
||||
# (keyword, text) 二元组用默认时长(5s)
|
||||
("ensure_app_running", "启动匠厂智能体平台"),
|
||||
("wait_gateway_ready", "网关就绪"),
|
||||
("step: 1/", "准备测试数据"),
|
||||
("step: 2/", "执行技能操作"),
|
||||
# (keyword, text, duration) 三元组指定独立时长
|
||||
("step: 3/", "技能在后端持续执行,请稍候...", 30.0),
|
||||
("step: 4/", "结果汇总返回"),
|
||||
("PASSED", "测试通过 ✓"),
|
||||
]
|
||||
|
||||
# 测试文件选择(按需切换):
|
||||
# - test_desktop_smoke.py: 纯问答 skill 用这个
|
||||
# - test_desktop_smoke_with_attachment.py: 需要上传附件的 skill 用这个
|
||||
TEST_FILE_NAME = "test_desktop_smoke.py"
|
||||
|
||||
# 音乐子目录(按 skill 风格选):
|
||||
# - "music/corporate": 商务严肃(金融、办公场景)
|
||||
# - "music/calm": 舒缓(流程演示、配置类)
|
||||
# - "music/upbeat": 活力(成功汇报、营销视频)
|
||||
# - "music"(默认): 三个子目录混选
|
||||
MUSIC_SUBDIR = "music/corporate"
|
||||
|
||||
# ============================================================
|
||||
|
||||
if __name__ == "__main__":
|
||||
_here = Path(__file__).resolve()
|
||||
skill_root = _here.parents[2]
|
||||
test_file = str(_here.parent / "test_desktop_smoke.py")
|
||||
test_file = str(_here.parent / TEST_FILE_NAME)
|
||||
output_dir = str(_here.parent / "artifacts" / "recordings")
|
||||
|
||||
run_screencast(
|
||||
@@ -78,4 +109,8 @@ if __name__ == "__main__":
|
||||
subtitle_script=SUBTITLE_SCRIPT,
|
||||
pytest_args=[test_file, "-v", "-s"],
|
||||
output_dir=output_dir,
|
||||
music_subdir=MUSIC_SUBDIR,
|
||||
# platform-kit 自动激活+最大化窗口(Windows ctypes / 其他系统协议),
|
||||
# 想关掉传 activate_window=False
|
||||
# monitor_index=1 显式录主屏(多显示器场景推荐)
|
||||
)
|
||||
|
||||
@@ -1,12 +1,29 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""
|
||||
可选桌面 E2E 示例(默认不执行:文件名后缀为 .sample)。
|
||||
可选桌面 E2E 示例(默认不执行:文件名后缀为 .sample)—— **简单问答场景**。
|
||||
|
||||
启用方式:
|
||||
复制本文件为 test_desktop_smoke.py,按目标技能修改提问文案与断言。
|
||||
运行:pytest tests/desktop/test_desktop_smoke.py -v -s
|
||||
|
||||
依赖:pytest、jiangchang_desktop_sdk、匠厂桌面宿主已就绪。
|
||||
|
||||
【何时用此模板】
|
||||
- 技能只需要发一句话给 Main Agent,等回复,做断言(纯问答)
|
||||
- 不涉及上传附件 / 不涉及 RPA 长流程
|
||||
|
||||
【何时不要用此模板,改用 test_desktop_smoke_with_attachment.py.sample】
|
||||
- 需要上传文件(Excel / PDF / 图片)作为附件
|
||||
- 技能内部跑 RPA / 长流程(> 2 分钟),需要精细等待
|
||||
|
||||
【已知 SDK 限制(来自 disburse-payroll-icbc 实战)】
|
||||
- client.ask() 内部对"找到 user 消息节点"只给 15 秒窗口,找不到就锁死
|
||||
user_msg_index=-1,后续等待 assistant 永远 return None。如果你的场景里
|
||||
匠厂启动较慢 / gateway 握手有延迟,client.ask() 可能假死到 timeout。
|
||||
此模板仍用 ask(),因为它对**简单场景**够用。复杂场景请用 _with_attachment.py.sample
|
||||
里的"绕开 ask()"模式。
|
||||
- client.send_file() 在 v2.0.17+ 抛 NotImplementedError,**绝不要调用它**。
|
||||
需要附件请用 _with_attachment.py.sample 里的拖拽方案。
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
|
||||
137
tests/desktop/test_desktop_smoke_with_attachment.py.sample
Normal file
137
tests/desktop/test_desktop_smoke_with_attachment.py.sample
Normal file
@@ -0,0 +1,137 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""
|
||||
可选桌面 E2E 示例(默认不执行:文件名后缀为 .sample)—— **附件上传 + 长流程场景**。
|
||||
|
||||
启用方式:
|
||||
1. 复制为 test_desktop_smoke_with_attachment.py
|
||||
2. 修改 SKILL_SLUG / PROMPT / 断言为你技能的真实业务
|
||||
3. 准备样例附件,或动态生成(参考 _prepare_sample_attachment)
|
||||
4. 运行:pytest tests/desktop/test_desktop_smoke_with_attachment.py -v -s
|
||||
|
||||
依赖:
|
||||
- pytest / jiangchang_desktop_sdk / 匠厂桌面宿主已就绪
|
||||
- 通过 conftest.py 自动 bootstrap,jiangchang_desktop_sdk.e2e_helpers 模块在 sys.path
|
||||
|
||||
【为什么用此模板而不是 test_desktop_smoke.py.sample】
|
||||
- client.send_file() 在 v2.0.17+ 抛 NotImplementedError,必须用拖拽方式
|
||||
- client.ask() 在 user_idx<0 时会锁死(已知 SDK 缺陷),本模板绕开 ask(),
|
||||
在测试侧自己用 jiangchang_desktop_sdk.e2e_helpers 做等待
|
||||
|
||||
【字幕配合】
|
||||
- 关键节点都打了 _logger.info("step: N/7 ..."),对应 record_screencast.py 的
|
||||
SUBTITLE_SCRIPT 用 step: 关键词触发字幕
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
import os
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
from jiangchang_desktop_sdk import JiangchangDesktopClient
|
||||
from jiangchang_desktop_sdk.e2e_helpers import (
|
||||
drop_file_into_composer,
|
||||
send_prompt_via_composer,
|
||||
wait_for_agent_complete,
|
||||
wait_for_attachment_ready,
|
||||
wait_for_composer_enabled,
|
||||
wait_for_user_message,
|
||||
)
|
||||
|
||||
# ============================================================
|
||||
# ↓↓↓ 复制后修改这些 ↓↓↓
|
||||
# ============================================================
|
||||
|
||||
SKILL_SLUG = "your-skill-slug"
|
||||
|
||||
PROMPT = (
|
||||
"请使用 /your-skill-slug 处理我刚才上传的文件。"
|
||||
"完成后请用清单方式汇总:[改成你技能要求的字段]。"
|
||||
)
|
||||
|
||||
# 业务断言:根据你的技能输出修改
|
||||
EXPECTED_KEYWORDS_IN_REPLY = ["关键词1", "关键词2"]
|
||||
|
||||
# ============================================================
|
||||
|
||||
_logger = logging.getLogger("desktop-e2e-with-attachment-sample")
|
||||
if not _logger.handlers:
|
||||
_logger.setLevel(logging.INFO)
|
||||
_h = logging.StreamHandler()
|
||||
_h.setFormatter(logging.Formatter("[E2E] %(message)s"))
|
||||
_logger.addHandler(_h)
|
||||
|
||||
_ARTIFACT_DIR = os.path.join(os.path.dirname(__file__), "artifacts")
|
||||
|
||||
|
||||
def _prepare_sample_attachment(tmp_path: Path) -> Path:
|
||||
"""在 tmp_path 下生成一个样例附件(按 skill 真实需要修改)。
|
||||
|
||||
示例:生成一个 5 行的 xlsx。你的 skill 可能需要 CSV / PDF / 图片等其他格式。
|
||||
"""
|
||||
from openpyxl import Workbook # noqa: WPS433
|
||||
xlsx = tmp_path / "sample.xlsx"
|
||||
wb = Workbook()
|
||||
ws = wb.active
|
||||
ws.append(["列1", "列2", "列3"])
|
||||
for i in range(1, 6):
|
||||
ws.append([f"值{i}A", f"值{i}B", i * 100])
|
||||
wb.save(str(xlsx))
|
||||
return xlsx
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def client(request):
|
||||
c = JiangchangDesktopClient()
|
||||
_logger.info("ensure_app_running")
|
||||
c.ensure_app_running(wait_timeout_s=45)
|
||||
c.bring_to_front()
|
||||
c.wait_gateway_ready(timeout_ms=30000)
|
||||
yield c
|
||||
try:
|
||||
rep = getattr(request.node, "rep_call", None)
|
||||
if rep is not None and rep.failed:
|
||||
os.makedirs(_ARTIFACT_DIR, exist_ok=True)
|
||||
paths = c.snapshot(_ARTIFACT_DIR, tag=request.node.name)
|
||||
_logger.error("failure snapshot: %s", paths)
|
||||
finally:
|
||||
c.disconnect()
|
||||
|
||||
|
||||
def test_desktop_with_attachment(client: JiangchangDesktopClient, tmp_path: Path) -> None:
|
||||
_logger.info("step: 1/7 准备样例附件")
|
||||
attachment_path = _prepare_sample_attachment(tmp_path)
|
||||
|
||||
client.new_task()
|
||||
page = client.get_page()
|
||||
client.bring_to_front()
|
||||
|
||||
_logger.info("step: 2/7 拖拽附件到聊天框")
|
||||
drop_file_into_composer(page, attachment_path)
|
||||
|
||||
_logger.info("step: 3/7 等附件 stage-buffer 完成")
|
||||
wait_for_attachment_ready(page, attachment_path.name)
|
||||
|
||||
wait_for_composer_enabled(page)
|
||||
client.bring_to_front()
|
||||
_logger.info("step: 4/7 键入并发送指令")
|
||||
send_prompt_via_composer(page, PROMPT)
|
||||
|
||||
_logger.info("step: 5/7 Main Agent 接收,等待路由到技能")
|
||||
user_idx = wait_for_user_message(page, PROMPT, timeout_s=60.0)
|
||||
|
||||
_logger.info("step: 6/7 技能在后端执行(可能 2-3 分钟)")
|
||||
client.bring_to_front()
|
||||
reply = wait_for_agent_complete(
|
||||
page, user_idx,
|
||||
overall_timeout_s=1200.0,
|
||||
stable_seconds=5.0,
|
||||
)
|
||||
|
||||
_logger.info("step: 7/7 校验回复内容")
|
||||
_logger.info("reply length=%d", len(reply))
|
||||
|
||||
assert reply, "assistant 回复为空"
|
||||
for kw in EXPECTED_KEYWORDS_IN_REPLY:
|
||||
assert kw in reply, f"期望关键词 {kw!r} 未出现在回复中;片段:{reply[:400]!r}"
|
||||
Reference in New Issue
Block a user