提交修改

This commit is contained in:
2026-04-22 15:51:37 +08:00
parent 2174b2b573
commit d6ad90a7db
16 changed files with 1384 additions and 358 deletions

View File

@@ -4,10 +4,20 @@ build-backend = "setuptools.build_meta"
[project] [project]
name = "jiangchang-desktop-sdk" name = "jiangchang-desktop-sdk"
version = "0.1.0" version = "0.4.0"
description = "匠厂桌面应用自动化测试 SDK" description = "匠厂桌面应用自动化测试 SDK + 共享 pytest plugin"
requires-python = ">=3.10" requires-python = ">=3.10"
dependencies = ["playwright>=1.42.0"] dependencies = ["playwright>=1.42.0"]
[project.optional-dependencies]
testing = ["pytest>=7.0"]
# 让 pytest 在 pip install 后自动发现并加载 jiangchang_desktop_sdk.testing.plugin。
# 未 pip install 时(开发态 sys.path 注入skill 的 conftest 可改用
# pytest_plugins = ["jiangchang_desktop_sdk.testing.plugin"]
# 二者等价。
[project.entry-points.pytest11]
jiangchang_desktop_sdk_testing = "jiangchang_desktop_sdk.testing.plugin"
[tool.setuptools.packages.find] [tool.setuptools.packages.find]
where = ["src"] where = ["src"]

View File

@@ -0,0 +1,6 @@
Metadata-Version: 2.4
Name: jiangchang-desktop-sdk
Version: 0.1.0
Summary: 匠厂桌面应用自动化测试 SDK
Requires-Python: >=3.10
Requires-Dist: playwright>=1.42.0

View File

@@ -0,0 +1,11 @@
pyproject.toml
src/jiangchang_desktop_sdk/__init__.py
src/jiangchang_desktop_sdk/client.py
src/jiangchang_desktop_sdk/exceptions.py
src/jiangchang_desktop_sdk/types.py
src/jiangchang_desktop_sdk.egg-info/PKG-INFO
src/jiangchang_desktop_sdk.egg-info/SOURCES.txt
src/jiangchang_desktop_sdk.egg-info/dependency_links.txt
src/jiangchang_desktop_sdk.egg-info/requires.txt
src/jiangchang_desktop_sdk.egg-info/top_level.txt
tests/test_client.py

View File

@@ -0,0 +1 @@
playwright>=1.42.0

View File

@@ -0,0 +1 @@
jiangchang_desktop_sdk

View File

@@ -1,6 +1,13 @@
from .client import JiangchangDesktopClient from .client import JiangchangDesktopClient
from .types import JiangchangMessage, AskOptions, LaunchOptions, AssertOptions from .types import JiangchangMessage, AskOptions, LaunchOptions, AssertOptions
from .exceptions import AppNotFoundError, TimeoutError, AssertError from .exceptions import (
AppNotFoundError,
ConnectionError,
TimeoutError,
AssertError,
LaunchError,
GatewayDownError,
)
__all__ = [ __all__ = [
"JiangchangDesktopClient", "JiangchangDesktopClient",
@@ -9,7 +16,10 @@ __all__ = [
"LaunchOptions", "LaunchOptions",
"AssertOptions", "AssertOptions",
"AppNotFoundError", "AppNotFoundError",
"ConnectionError",
"TimeoutError", "TimeoutError",
"AssertError", "AssertError",
"LaunchError",
"GatewayDownError",
] ]
__version__ = "0.1.0" __version__ = "0.4.0"

File diff suppressed because it is too large Load Diff

View File

@@ -21,3 +21,7 @@ class AssertError(JiangchangDesktopError):
class LaunchError(JiangchangDesktopError): class LaunchError(JiangchangDesktopError):
"""应用启动失败时抛出""" """应用启动失败时抛出"""
pass pass
class GatewayDownError(JiangchangDesktopError):
"""Gateway 在等待过程中被检测到已停止/退出时抛出,便于测试立刻失败而不是空等超时。"""
pass

View File

@@ -0,0 +1,28 @@
# -*- coding: utf-8 -*-
"""匠厂桌面 SDK 的测试工具子包。
提供 skill 桌面 E2E 测试所需的共享工具:
- ``SkillInfo`` / ``discover_skill_root`` / ``parse_skill_md``
- ``skill_healthcheck``
- ``HostAPIClient`` / ``HostAPIError``
- ``clear_main_agent_history``
pytest plugin 本体位于子模块 ``plugin``,由 pyproject.toml 的
``[project.entry-points.pytest11]`` 自动注册;未 pip install 场景下,
skill 的 conftest 也可通过 ``pytest_plugins`` 变量手动加载。
"""
from .config import SkillInfo, discover_skill_root, parse_skill_md
from .healthcheck import HealthCheckError, skill_healthcheck
from .host_api import HostAPIClient, HostAPIError
from .session_cleanup import clear_main_agent_history
__all__ = [
"SkillInfo",
"discover_skill_root",
"parse_skill_md",
"HealthCheckError",
"skill_healthcheck",
"HostAPIClient",
"HostAPIError",
"clear_main_agent_history",
]

View File

@@ -0,0 +1,137 @@
# -*- coding: utf-8 -*-
"""Skill 元信息读取。
负责:
- 定位 skill 根目录(含 SKILL.md
- 解析 SKILL.md 的 YAML frontmatter抽出 slug / version / name 等关键字段。
为避免硬性依赖 PyYAML这里用一个**足够用**的轻量正则 parser只抽取
我们真正关心的字段(顶层 `name` / `version` / `author`,嵌套 `openclaw.slug`
与 `openclaw.category`)。更复杂的 YAML 结构请自行用 PyYAML 解析。
"""
from __future__ import annotations
import os
import re
from dataclasses import dataclass
from pathlib import Path
from typing import Optional
SKILL_MD = "SKILL.md"
SKILL_ROOT_ENV = "JIANGCHANG_E2E_SKILL_ROOT"
@dataclass
class SkillInfo:
root: str
slug: str
name: str
version: str
author: str = ""
category: str = ""
def discover_skill_root(start: Optional[str] = None) -> str:
"""定位 skill 根目录。
优先级:
1. 环境变量 ``JIANGCHANG_E2E_SKILL_ROOT``(由各 skill 的 conftest 提前注入);
2. 从 ``start``(默认 cwd向上回溯直到找到含 ``SKILL.md`` 的目录。
"""
env = (os.environ.get(SKILL_ROOT_ENV) or "").strip()
if env:
env_abs = os.path.abspath(env)
if os.path.isfile(os.path.join(env_abs, SKILL_MD)):
return env_abs
cur = Path(start or os.getcwd()).resolve()
for parent in [cur, *cur.parents]:
if (parent / SKILL_MD).exists():
return str(parent)
raise FileNotFoundError(
f"未能在 {start or os.getcwd()} 及其父目录中找到 {SKILL_MD}"
f"请设置环境变量 {SKILL_ROOT_ENV}=<skill 根目录绝对路径>。"
)
_FRONTMATTER_RE = re.compile(r"^---\s*\n(.*?)\n---\s*\n", re.DOTALL)
def _extract_frontmatter(text: str) -> str:
match = _FRONTMATTER_RE.match(text)
if not match:
raise ValueError("SKILL.md 缺少 YAML frontmatter--- ... ---)。")
return match.group(1)
def _strip_quotes(value: str) -> str:
v = value.strip()
if len(v) >= 2 and v[0] == v[-1] and v[0] in ("'", '"'):
return v[1:-1]
return v
def parse_skill_md(skill_root: str) -> SkillInfo:
"""解析 SKILL.md 的 frontmatter返回 SkillInfo。"""
md_path = Path(skill_root) / SKILL_MD
if not md_path.exists():
raise FileNotFoundError(f"SKILL.md 不存在:{md_path}")
text = md_path.read_text(encoding="utf-8")
fm = _extract_frontmatter(text)
top: dict[str, str] = {}
openclaw: dict[str, str] = {}
lines = fm.splitlines()
i = 0
while i < len(lines):
line = lines[i]
if not line.strip() or line.lstrip().startswith("#"):
i += 1
continue
# 顶层 key: value
m = re.match(r"^([A-Za-z_][\w-]*):\s*(.*)$", line)
if m:
key, rest = m.group(1), m.group(2)
if rest.strip() == "":
# 嵌套块,收集 2 空格缩进的子项
i += 1
sub: dict[str, str] = {}
while i < len(lines):
sub_line = lines[i]
if sub_line.strip() == "" or sub_line.lstrip().startswith("#"):
i += 1
continue
if not sub_line.startswith(" "):
break
sm = re.match(r"^\s+([A-Za-z_][\w-]*):\s*(.*)$", sub_line)
if sm:
sub[sm.group(1)] = _strip_quotes(sm.group(2))
i += 1
if key == "metadata":
# openclaw 在 metadata 下再下一层:
# metadata:
# openclaw:
# slug: ...
# 上面 sub 已经拿到的只是 "openclaw" 这个 key。为稳妥起见回头
# 用一个更宽松的扫描:直接找所有含 `slug:` `category:` 的行。
pass
continue
top[key] = _strip_quotes(rest)
i += 1
# openclaw.slug / category 用宽松扫描兜底(无论嵌套几层)
for line in lines:
stripped = line.lstrip()
if stripped.startswith("slug:") and "slug" not in openclaw:
openclaw["slug"] = _strip_quotes(stripped.split(":", 1)[1])
elif stripped.startswith("category:") and "category" not in openclaw:
openclaw["category"] = _strip_quotes(stripped.split(":", 1)[1])
return SkillInfo(
root=os.path.abspath(skill_root),
slug=openclaw.get("slug", "") or Path(skill_root).name,
name=top.get("name", Path(skill_root).name),
version=top.get("version", "0.0.0"),
author=top.get("author", ""),
category=openclaw.get("category", ""),
)

View File

@@ -0,0 +1,73 @@
# -*- coding: utf-8 -*-
"""技能健康检查。
检查项:
1. SKILL.md 成功解析,含 slug / version。
2. ``python scripts/main.py health`` 命令退出码为 0每个 skill 都约定实现)。
CLI 耗时通常 <2s一个 pytest session 里只跑一次session-scope autouse
失败时 fail-fast 终止整个测试 session。开发时可设
``JIANGCHANG_E2E_SKIP_HEALTHCHECK=1`` 临时跳过。
"""
from __future__ import annotations
import logging
import os
import subprocess
import sys
from .config import SkillInfo
logger = logging.getLogger("jiangchang-desktop-sdk.healthcheck")
class HealthCheckError(RuntimeError):
"""健康检查失败。"""
def skill_healthcheck(
skill_info: SkillInfo,
*,
run_cli: bool = True,
timeout_s: float = 15.0,
) -> None:
"""对一个 skill 跑本地健康检查。失败抛 ``HealthCheckError``。"""
if not skill_info.slug:
raise HealthCheckError(f"SKILL.md 未定义 openclaw.slug{skill_info.root}")
if not skill_info.version or skill_info.version == "0.0.0":
raise HealthCheckError(f"SKILL.md 未定义 version{skill_info.root}")
if not run_cli:
return
main_py = os.path.join(skill_info.root, "scripts", "main.py")
if not os.path.isfile(main_py):
logger.warning("skill_healthcheck: %s 不存在,跳过 CLI health 检查", main_py)
return
cmd = [sys.executable, main_py, "health"]
logger.debug("skill_healthcheck: 运行 %s (cwd=%s)", cmd, skill_info.root)
try:
proc = subprocess.run(
cmd,
capture_output=True,
text=True,
timeout=timeout_s,
cwd=skill_info.root,
encoding="utf-8",
errors="replace",
)
except subprocess.TimeoutExpired as exc:
raise HealthCheckError(
f"skill_healthcheck: '{' '.join(cmd)}' 超时 {timeout_s}s"
) from exc
if proc.returncode != 0:
raise HealthCheckError(
f"skill_healthcheck: '{' '.join(cmd)}' 退出码 {proc.returncode}\n"
f"stdout: {proc.stdout[:600]}\n"
f"stderr: {proc.stderr[:600]}"
)
logger.info(
"skill_healthcheck: %s v%s OK", skill_info.slug, skill_info.version
)

View File

@@ -0,0 +1,111 @@
# -*- coding: utf-8 -*-
"""匠厂 Host HTTP API 的 Python 轻封装。
使用 stdlib urllib避免为 SDK 引入新运行时依赖。只暴露本模块当前
真正需要的端点;其余接口按需添加。
Host API 默认端口:`13210`(见 jiangchang/electron/utils/config.ts 中 `CLAWX_HOST_API`)。
可通过环境变量 ``CLAWX_PORT_CLAWX_HOST_API`` 或本类的 ``port`` 参数覆盖。
"""
from __future__ import annotations
import json
import logging
import os
import urllib.error
import urllib.request
from typing import Any, Dict, Optional
DEFAULT_PORT = 13210
DEFAULT_HOST = "127.0.0.1"
logger = logging.getLogger("jiangchang-desktop-sdk.host_api")
class HostAPIError(RuntimeError):
"""Host API 调用失败(网络错误或非 2xx 响应)。"""
class HostAPIClient:
def __init__(
self,
host: str = DEFAULT_HOST,
port: Optional[int] = None,
timeout: float = 10.0,
) -> None:
self.host = host
env_port = os.environ.get("CLAWX_PORT_CLAWX_HOST_API")
self.port = port or (int(env_port) if env_port else DEFAULT_PORT)
self.timeout = timeout
@property
def base_url(self) -> str:
return f"http://{self.host}:{self.port}"
# ── HTTP helpers ─────────────────────────────────────────
def _request(self, method: str, path: str, body: Any = None) -> Any:
url = f"{self.base_url}{path}"
data = None
headers = {"Accept": "application/json"}
if body is not None:
data = json.dumps(body).encode("utf-8")
headers["Content-Type"] = "application/json"
req = urllib.request.Request(url, data=data, method=method, headers=headers)
try:
with urllib.request.urlopen(req, timeout=self.timeout) as resp:
raw = resp.read().decode("utf-8")
if not raw:
return None
try:
return json.loads(raw)
except json.JSONDecodeError:
return raw
except urllib.error.HTTPError as exc:
payload: Any
try:
payload = json.loads(exc.read().decode("utf-8"))
except Exception: # noqa: BLE001
payload = str(exc)
raise HostAPIError(f"{method} {path} -> HTTP {exc.code}: {payload}") from exc
except urllib.error.URLError as exc:
raise HostAPIError(f"{method} {path} -> {exc}") from exc
# ── High-level API ───────────────────────────────────────
def ping(self) -> bool:
try:
self.list_agents()
return True
except HostAPIError:
return False
def list_agents(self) -> Dict[str, Any]:
return self._request("GET", "/api/agents") or {}
def get_main_session_key(self) -> str:
"""返回主 Agent 的 mainSessionKey形如 ``agent:main:{mainKey}``)。
若多种形式都取不到,退化为 ``agent:main:default``——这通常仍能被
后端 ``POST /api/sessions/delete`` 正确识别(会走 sessions.json 查找)。
"""
snapshot = self.list_agents()
entries = snapshot.get("entries") or []
for entry in entries:
if entry.get("id") == "main":
key = entry.get("mainSessionKey") or entry.get("sessionKey")
if key:
return key
key = snapshot.get("mainSessionKey")
if key:
return key
return "agent:main:default"
def delete_session(self, session_key: str) -> None:
self._request("POST", "/api/sessions/delete", {"sessionKey": session_key})
def sessions_dir(self, agent_id: str = "main") -> str:
"""返回指定 agent 的 sessions 目录绝对路径(供文件系统回退使用)。"""
return os.path.join(
os.path.expanduser("~"), ".openclaw", "agents", agent_id, "sessions"
)

View File

@@ -0,0 +1,117 @@
# -*- coding: utf-8 -*-
"""jiangchang_desktop_sdk.testing 的 pytest plugin。
提供的 fixture / hook所有 skill 通用):
session 级:
- ``skill_info`` 当前 skill 的 SkillInfo从 SKILL.md 解析)
- ``host_api`` 匠厂 Host API 的 Python 客户端
- ``jc_desktop_session_setup`` (autouse) 跑一次健康检查 + 清空主 Agent 历史
function 级:
- ``failure_snapshot_dir`` 失败现场落盘目录tests/desktop/artifacts/
hook
- ``pytest_runtest_makereport`` 把 setup/call/teardown 结果挂到
``item.rep_setup`` / ``item.rep_call`` / ``item.rep_teardown``
环境变量:
- ``JIANGCHANG_E2E_SKILL_ROOT`` 显式指定 skill 根目录
- ``JIANGCHANG_E2E_SKIP_HEALTHCHECK`` 置 1 可临时跳过健康检查
- ``JIANGCHANG_E2E_SKIP_CLEAR_HISTORY`` 置 1 可临时跳过清空历史
加载方式(二选一):
A) ``pip install -e`` 安装本 SDK 后,通过 ``pyproject.toml`` 的
``[project.entry-points.pytest11]`` 自动加载;
B) 在 skill 的 ``tests/desktop/conftest.py`` 里声明
``pytest_plugins = ["jiangchang_desktop_sdk.testing.plugin"]``。
"""
from __future__ import annotations
import logging
import os
import pytest
from .config import SkillInfo, discover_skill_root, parse_skill_md
from .healthcheck import HealthCheckError, skill_healthcheck
from .host_api import HostAPIClient
from .session_cleanup import clear_main_agent_history
logger = logging.getLogger("jiangchang-desktop-sdk.plugin")
# ── Fixtures ──────────────────────────────────────────────────────────────
@pytest.fixture(scope="session")
def skill_info(request) -> SkillInfo:
rootpath = str(request.config.rootpath)
root = discover_skill_root(start=rootpath)
info = parse_skill_md(root)
logger.info(
"[skill_info] root=%s slug=%s version=%s", info.root, info.slug, info.version
)
return info
@pytest.fixture(scope="session")
def host_api() -> HostAPIClient:
return HostAPIClient()
@pytest.fixture(scope="session", autouse=True)
def jc_desktop_session_setup(skill_info, host_api):
"""Desktop 测试 session 级一次性准备。
1. ``skill_healthcheck``SKILL.md + ``python scripts/main.py health``
失败时 **fail-fast** 终止整个 session
2. ``clear_main_agent_history``:通过 Host API 清空主 Agent 的当前 session
规避跨 case 的上下文污染。API 不可达则回退文件系统,不影响测试启动。
"""
if not os.environ.get("JIANGCHANG_E2E_SKIP_HEALTHCHECK"):
try:
skill_healthcheck(skill_info)
except HealthCheckError as exc:
pytest.exit(
f"[jc-healthcheck] 技能健康检查失败,终止测试 session\n{exc}\n"
"临时跳过:设置 JIANGCHANG_E2E_SKIP_HEALTHCHECK=1。",
returncode=5,
)
else:
logger.warning("[jc-healthcheck] 已通过环境变量跳过健康检查")
if not os.environ.get("JIANGCHANG_E2E_SKIP_CLEAR_HISTORY"):
try:
cleared = clear_main_agent_history(host_api)
logger.info("[jc-cleanup] session 启动前已清空 %d 个历史 session", cleared)
except Exception as exc: # noqa: BLE001
logger.warning("[jc-cleanup] 清空历史失败(非致命):%s", exc)
else:
logger.warning("[jc-cleanup] 已通过环境变量跳过清空历史")
yield
@pytest.fixture
def failure_snapshot_dir(request) -> str:
"""失败现场落盘目录:与测试文件同级的 artifacts/ 目录。"""
test_dir = os.path.dirname(str(request.node.fspath))
artifacts = os.path.join(test_dir, "artifacts")
os.makedirs(artifacts, exist_ok=True)
return artifacts
# ── Hooks ─────────────────────────────────────────────────────────────────
@pytest.hookimpl(tryfirst=True, hookwrapper=True)
def pytest_runtest_makereport(item, call):
outcome = yield
rep = outcome.get_result()
setattr(item, f"rep_{rep.when}", rep)
def pytest_configure(config):
config.addinivalue_line(
"markers",
"multi_turn: 多轮追问测试用例(同类用例共享同一会话上下文)",
)

View File

@@ -0,0 +1,73 @@
# -*- coding: utf-8 -*-
"""清空主 Agent 的 chat 会话历史。
主路径:`POST /api/sessions/delete`(匠厂官方 Host API软删除改名 `.deleted.jsonl`)。
回退路径:直接把 `~/.openclaw/agents/main/sessions/*.jsonl` 重命名为 `.deleted.jsonl`。
"""
from __future__ import annotations
import logging
import os
from typing import Optional
from .host_api import HostAPIClient, HostAPIError
logger = logging.getLogger("jiangchang-desktop-sdk.session_cleanup")
def clear_main_agent_history(
api: Optional[HostAPIClient] = None,
*,
agent_id: str = "main",
strict: bool = False,
) -> int:
"""清空主 Agent 的当前 session。返回被删除的 session 数量。
``strict=True`` 时 API 失败立刻抛出;默认 ``False`` 下 API 失败会降级到
文件系统批量重命名,尽最大努力保证下一个测试看到的是干净环境。
"""
api = api or HostAPIClient()
# ── Path A: Host API ───────────────────────────────────
try:
session_key = api.get_main_session_key()
logger.debug("clear_main_agent_history: 主 session key = %s", session_key)
api.delete_session(session_key)
logger.info(
"clear_main_agent_history: HTTP API 已软删除 %s", session_key
)
return 1
except HostAPIError as exc:
if strict:
raise
logger.warning(
"clear_main_agent_history: HTTP API 失败(%s),回退到文件系统", exc
)
# ── Path B: 文件系统回退 ────────────────────────────────
sessions_dir = api.sessions_dir(agent_id)
if not os.path.isdir(sessions_dir):
logger.info(
"clear_main_agent_history: %s 不存在,无历史可清", sessions_dir
)
return 0
deleted = 0
for name in os.listdir(sessions_dir):
if not name.endswith(".jsonl") or name.endswith(".deleted.jsonl"):
continue
src = os.path.join(sessions_dir, name)
dst = src[: -len(".jsonl")] + ".deleted.jsonl"
try:
os.replace(src, dst)
deleted += 1
except OSError as exc:
logger.warning(
"clear_main_agent_history: 重命名 %s 失败:%s", src, exc
)
if strict:
raise
logger.info(
"clear_main_agent_history: 文件系统回退共软删除 %d 个 session", deleted
)
return deleted

View File

@@ -20,6 +20,16 @@ class AskOptions:
timeout: int = 120000 # 毫秒 timeout: int = 120000 # 毫秒
wait_for_tools: bool = True wait_for_tools: bool = True
agent_id: str = "main" agent_id: str = "main"
# 拟人化输入每个字符间延迟毫秒。0 = 直接 fill()。
typing_delay_ms: int = 25
# 发送方式True=按 Enter 键False=点击发送按钮
use_enter_key: bool = True
# 每次 ask 前是否自动新建任务(避免上下文污染)
new_task: bool = True
# 流式输出判稳阈值(秒):助手消息文本连续该秒数无增长视为完成
stable_seconds: float = 3.0
# 轮询间隔(秒)
poll_interval: float = 0.5
@dataclass @dataclass
class LaunchOptions: class LaunchOptions: