diff --git a/tests/desktop/README.md b/tests/desktop/README.md index 0e11662..a8ecf3e 100644 --- a/tests/desktop/README.md +++ b/tests/desktop/README.md @@ -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)` 给个别字幕指定独立时长, + 覆盖长等待场景中间没日志输出的"字幕真空" diff --git a/tests/desktop/conftest.py b/tests/desktop/conftest.py index 2ae9e48..09811f4 100644 --- a/tests/desktop/conftest.py +++ b/tests/desktop/conftest.py @@ -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 /jiangchang-platform-kit/sdk/jiangchang-desktop-sdk\n" + " 1) pip install -e /jiangchang-platform-kit\n" " 2) 设置环境变量 JIANGCHANG_KIT_ROOT=/jiangchang-platform-kit\n" - " 3) 将 jiangchang-platform-kit 与本仓库放在同一 workspace 父级可推断的位置,例如:\n" - f" {workspace_root}/jiangchang-platform-kit/ 或 {open_claw_root}/jiangchang-platform-kit/\n" + " (SDK 新版位置:\\src;旧版:\\sdk\\jiangchang-desktop-sdk\\src)\n" + " 3) 将 jiangchang-platform-kit 与本仓库放在同一 workspace 父级可推断的位置\n" "已尝试的候选路径:\n - " + "\n - ".join(tried), returncode=4, ) diff --git a/tests/desktop/record_screencast.py.sample b/tests/desktop/record_screencast.py.sample index c39a9b9..89cfa1a 100644 --- a/tests/desktop/record_screencast.py.sample +++ b/tests/desktop/record_screencast.py.sample @@ -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 显式录主屏(多显示器场景推荐) ) diff --git a/tests/desktop/test_desktop_smoke.py.sample b/tests/desktop/test_desktop_smoke.py.sample index 095a0df..e099851 100644 --- a/tests/desktop/test_desktop_smoke.py.sample +++ b/tests/desktop/test_desktop_smoke.py.sample @@ -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 diff --git a/tests/desktop/test_desktop_smoke_with_attachment.py.sample b/tests/desktop/test_desktop_smoke_with_attachment.py.sample new file mode 100644 index 0000000..b21f1a8 --- /dev/null +++ b/tests/desktop/test_desktop_smoke_with_attachment.py.sample @@ -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}"