ee2a4d9e22
All checks were successful
技能自动化发布 / release (push) Successful in 15s
基于 skill-template 脚手架开发，包含： - 5 张业务表（教职工、请假政策、考勤日历、请假记录、假期余额） - 请假核算引擎（扣薪计算、审批层级判定、年假抵扣） - CLI 入口（import-employees / calculate-leave / render-html） - HTML 审批与扣薪报告生成 - 示例数据与完整测试套件（295 tests passed）
138 lines
4.7 KiB
Plaintext
138 lines
4.7 KiB
Plaintext
# -*- 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 = "process-leave-attendance-default"
|
||
|
||
PROMPT = (
|
||
"请使用 /process-leave-attendance-default 处理我刚才上传的文件。"
|
||
"完成后请用清单方式汇总：[改成你技能要求的字段]。"
|
||
)
|
||
|
||
# 业务断言：根据你的技能输出修改
|
||
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}"
|