14 Commits
v1.0.3 ... main

19 changed files with 1895 additions and 9 deletions

View File

@@ -0,0 +1,115 @@
# Standalone reusable workflow for Node/Vite static sites.
# Does not reference reusable-release-skill.yaml; skill workflows remain unchanged.
#
# Node请在 Runner 上预装 Node 20+(勿用 actions/setup-node避免每次从 GitHub 拉包)。
# npm通过 npm_registry 输入使用国内镜像(默认 npmmirror
# Checkout不用 actions/checkouthost 模式下 JS Action 仍会进无 Node 的容器);改用 git + github.token。
# 重要act_runner 会把 run: 脚本写到 $GITHUB_WORKSPACE/workflow/*.sh
# 所以 checkout 绝对不能在 workspace 根上 rm -rf ./*,否则会把自己这个脚本一起删了。
# 解决办法:克隆到 $GITHUB_WORKSPACE/_src 子目录,后续 step 都在该子目录操作。
# 浅拉不要用「裸 SHA」作 fetch 参数Gitea 上易失败);按分支 clone 再对齐 SHA。
name: Reusable Frontend Deploy
on:
workflow_call:
inputs:
deploy_path:
description: "Target directory for static files (trailing slash optional)"
required: false
type: string
default: "/www/wwwroot/sandbox/web"
build_command:
required: false
type: string
default: "npm run build:jc2009"
npm_registry:
description: "npm registry (e.g. npmmirror for CN)"
required: false
type: string
default: "https://registry.npmmirror.com"
runs_on:
description: "Runner label; must match a registered runner"
required: false
type: string
default: "ubuntu-latest"
chown_www:
description: "Run chown -R www:www after sync (requires permission on runner)"
required: false
type: boolean
default: true
jobs:
build-and-deploy:
runs-on: ${{ inputs.runs_on }}
defaults:
run:
shell: bash
working-directory: ${{ github.workspace }}/_src
env:
NPM_CONFIG_REGISTRY: ${{ inputs.npm_registry }}
steps:
- name: Checkout
# 这一步在 workspace 根执行,不能用默认的 _src此时还不存在
working-directory: ${{ github.workspace }}
env:
GITEA_HOST: git.jc2009.com
GITEA_TOKEN: ${{ github.token }}
GITHUB_TOKEN: ${{ github.token }}
run: |
set -euo pipefail
git config --global --add safe.directory '*'
REPO="${{ github.repository }}"
SHA="${{ github.sha }}"
BRANCH="${{ github.ref_name }}"
TOKEN="${GITEA_TOKEN:-${GITHUB_TOKEN:-}}"
test -n "$TOKEN" || { echo "Checkout: empty token (github.token not available to this job)"; exit 1; }
URL="https://x-access-token:${TOKEN}@${GITEA_HOST}/${REPO}.git"
SRC_DIR="${GITHUB_WORKSPACE}/_src"
# 只清理 _src不动 workspace 根(保护 runner 写入的 workflow/*.sh
rm -rf "$SRC_DIR"
git clone --depth 1 --branch "$BRANCH" "$URL" "$SRC_DIR"
cd "$SRC_DIR"
if [ "$(git rev-parse HEAD)" != "$SHA" ]; then
git fetch --depth 200 origin "refs/heads/${BRANCH}"
git checkout --force "$SHA"
fi
echo "Checked out $(git rev-parse HEAD) to $SRC_DIR"
- name: Check Node.js (pre-installed on runner)
run: |
set -e
command -v node >/dev/null || { echo "Runner 上未找到 node请先安装 Node 20+"; exit 1; }
command -v npm >/dev/null || { echo "Runner 上未找到 npm"; exit 1; }
node -v
npm -v
NODE_MAJOR=$(node -p "process.versions.node.split('.')[0]")
test "$NODE_MAJOR" -ge 20 || { echo "需要 Node 20 及以上,当前: $(node -v)"; exit 1; }
- name: Install dependencies
run: npm ci
- name: Build
run: ${{ inputs.build_command }}
- name: Deploy (sync dist to web root)
run: |
set -euo pipefail
test -d dist
DEST="${{ inputs.deploy_path }}"
DEST="${DEST%/}"
mkdir -p "$DEST"
# 宝塔面板会自动生成 .user.ini(并加 immutable 属性)和 .htaccess,跳过它们
find "$DEST" -mindepth 1 -maxdepth 1 \
! -name ".user.ini" \
! -name ".htaccess" \
-exec rm -rf {} +
cp -a dist/. "$DEST/"
- name: Set ownership for Nginx (Baota www)
run: |
if [ "${{ inputs.chown_www }}" != "true" ]; then exit 0; fi
DEST="${{ inputs.deploy_path }}"
DEST="${DEST%/}"
chown -R www:www "$DEST" || chown -R nginx:nginx "$DEST" || true

View File

@@ -3,6 +3,11 @@ name: Reusable Skill Release
on:
workflow_call:
inputs:
runs_on:
description: "Runner label; must match a registered runner (use host runner for pip/python on same machine as Node frontend)"
required: false
type: string
default: ubuntu-latest
artifact_platform:
required: false
type: string
@@ -30,23 +35,30 @@ on:
jobs:
build-and-deploy:
runs-on: ubuntu-latest
runs-on: ${{ inputs.runs_on }}
defaults:
run:
shell: bash
env:
ARTIFACT_PLATFORM: ${{ inputs.artifact_platform }}
PYARMOR_PLATFORM: ${{ inputs.pyarmor_platform }}
PIP_BREAK_SYSTEM_PACKAGES: "1"
# Prefer self-built Python 3.12 under /usr/local (Alibaba Cloud Linux host); keep system paths as fallback.
# 显式前缀,避免部分 Runner 未注入 env.PATH 时丢失系统路径
PATH: /usr/local/bin:/usr/local/python3.12/bin:/usr/bin:/bin:/usr/local/sbin:/usr/sbin
# PyArmor 交叉平台加密时会内部执行 pip 安装 pyarmor.cli.core.* 等包;不设则默认走 files.pythonhosted.org国内 CI 易超时。
PIP_INDEX_URL: https://pypi.tuna.tsinghua.edu.cn/simple
PIP_EXTRA_INDEX_URL: https://mirrors.aliyun.com/pypi/simple https://mirrors.cloud.tencent.com/pypi/simple https://mirrors.huaweicloud.com/repository/pypi/simple
PIP_DEFAULT_TIMEOUT: "180"
PIP_TRUSTED_HOST: pypi.tuna.tsinghua.edu.cn mirrors.aliyun.com mirrors.cloud.tencent.com mirrors.huaweicloud.com files.pythonhosted.org pypi.org
steps:
- uses: http://120.25.191.12:3000/admin/actions-checkout@v4
- uses: https://git.jc2009.com/admin/actions-checkout@v4
# Pin PyArmor 8.5.3 — matches desktop bundles; 9.x trial is stricter in CI。
# 镜像由 job envPIP_INDEX_URL / PIP_EXTRA_INDEX_URL统一指定与 Encrypt 步骤中 PyArmor 内部 pip 一致。
# 使用 python3.12 -m pip避免仅存在 python3(3.6) 或裸 pip 不在 PATH 的宿主机/容器。
- name: Setup Tools
run: pip install "pyarmor==8.5.3" requests python-frontmatter --break-system-packages
run: python3.12 -m pip install "pyarmor==8.5.3" requests python-frontmatter --break-system-packages
- name: Register PyArmor (optional)
env:
@@ -55,12 +67,13 @@ jobs:
if [ -z "${PYARMOR_REG_B64}" ]; then
echo "PyArmor: no PYARMOR_REG_B64 secret — trial mode (very large single .py modules may fail to obfuscate)."
else
python -c "import os,base64,pathlib,subprocess; p=pathlib.Path('/tmp/pyarmor-reg.zip'); p.write_bytes(base64.standard_b64decode(os.environ['PYARMOR_REG_B64'])); subprocess.run(['pyarmor','reg',str(p)],check=True); p.unlink(missing_ok=True)"
python3.12 -c "import os,base64,pathlib,subprocess; os.environ['PATH']='/usr/local/bin:/usr/local/python3.12/bin:'+os.environ.get('PATH',''); p=pathlib.Path('/tmp/pyarmor-reg.zip'); p.write_bytes(base64.standard_b64decode(os.environ['PYARMOR_REG_B64'])); subprocess.run(['pyarmor','reg',str(p)],check=True); p.unlink(missing_ok=True)"
fi
# 递归加密整个 scripts/(含 cli、service、db、util 等子包);产物保留与源码一致的 scripts/ 层级,入口为 scripts/main.py。
- name: Encrypt Source Code
run: |
export PATH="/usr/local/bin:/usr/local/python3.12/bin:${PATH:-}"
mkdir -p dist/package/scripts
set -euo pipefail
test -d scripts
@@ -72,11 +85,30 @@ jobs:
- name: Parse Metadata and Pack
id: build_task
run: |
python -c "
python3.12 -c "
import frontmatter, os, json, shutil
post = frontmatter.load('SKILL.md')
metadata = dict(post.metadata or {})
metadata['readme_md'] = (post.content or '').strip()
skill_readme_md = (post.content or '').strip()
skill_description = metadata.get('description')
metadata['readme_md'] = skill_readme_md
readme_path = os.path.join('references', 'README.md')
if os.path.isfile(readme_path):
try:
readme_post = frontmatter.load(readme_path)
body = (readme_post.content or '').strip()
rm_meta = readme_post.metadata or {}
desc = rm_meta.get('description')
if desc is not None and str(desc).strip():
metadata['description'] = str(desc).strip()
if body:
metadata['readme_md'] = body
except Exception:
pass
if not (metadata.get('readme_md') or '').strip():
metadata['readme_md'] = skill_readme_md
if skill_description is not None and not str(metadata.get('description', '') or '').strip():
metadata['description'] = skill_description
openclaw_meta = metadata.get('metadata', {}).get('openclaw', {})
slug = (openclaw_meta.get('slug') or metadata.get('slug') or metadata.get('name') or '').strip()
if not slug:
@@ -102,7 +134,7 @@ jobs:
METADATA_JSON: ${{ steps.build_task.outputs.metadata }}
SYNC_URL: ${{ inputs.sync_url }}
run: |
python -c "
python3.12 -c "
import requests, json, os
metadata = json.loads(os.environ['METADATA_JSON'])
res = requests.post(os.environ['SYNC_URL'], json=metadata)
@@ -121,7 +153,7 @@ jobs:
ARTIFACT_PLATFORM: ${{ steps.build_task.outputs.artifact_platform }}
UPLOAD_URL: ${{ inputs.upload_url }}
run: |
python -c "
python3.12 -c "
import requests, os
slug = os.environ['SLUG']
version = os.environ['VERSION']
@@ -148,7 +180,7 @@ jobs:
VERSION: ${{ steps.build_task.outputs.version }}
PRUNE_URL: ${{ inputs.prune_url }}
run: |
python -c "
python3.12 -c "
import requests, os
payload = {
'name': os.environ['SLUG'],

View File

@@ -0,0 +1,23 @@
[build-system]
requires = ["setuptools>=61.0", "wheel"]
build-backend = "setuptools.build_meta"
[project]
name = "jiangchang-desktop-sdk"
version = "0.4.0"
description = "匠厂桌面应用自动化测试 SDK + 共享 pytest plugin"
requires-python = ">=3.10"
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]
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

@@ -0,0 +1,25 @@
from .client import JiangchangDesktopClient
from .types import JiangchangMessage, AskOptions, LaunchOptions, AssertOptions
from .exceptions import (
AppNotFoundError,
ConnectionError,
TimeoutError,
AssertError,
LaunchError,
GatewayDownError,
)
__all__ = [
"JiangchangDesktopClient",
"JiangchangMessage",
"AskOptions",
"LaunchOptions",
"AssertOptions",
"AppNotFoundError",
"ConnectionError",
"TimeoutError",
"AssertError",
"LaunchError",
"GatewayDownError",
]
__version__ = "0.4.0"

File diff suppressed because it is too large Load Diff

View File

@@ -0,0 +1,27 @@
class JiangchangDesktopError(Exception):
"""基类"""
pass
class AppNotFoundError(JiangchangDesktopError):
"""Electron 可执行文件找不到时抛出"""
pass
class ConnectionError(JiangchangDesktopError, ConnectionError):
"""连接桌面应用失败时抛出"""
pass
class TimeoutError(JiangchangDesktopError, TimeoutError):
"""操作超时"""
pass
class AssertError(JiangchangDesktopError):
"""断言失败时抛出,消息要包含期望值和实际值"""
pass
class LaunchError(JiangchangDesktopError):
"""应用启动失败时抛出"""
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

@@ -0,0 +1,48 @@
from dataclasses import dataclass
from typing import Optional, List
@dataclass
class JiangchangMessage:
"""单条会话消息"""
id: str
role: str # 'user' | 'assistant' | 'system' | 'tool'
content: str
timestamp: float
is_error: bool = False
tool_call_id: Optional[str] = None
tool_name: Optional[str] = None
tool_status: Optional[str] = None # 'running' | 'completed' | 'error'
thinking_content: Optional[str] = None
@dataclass
class AskOptions:
"""ask() 方法的选项"""
timeout: int = 120000 # 毫秒
wait_for_tools: bool = True
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
class LaunchOptions:
"""launch_app() 方法的选项"""
executable_path: Optional[str] = None # 默认从 JIANGCHANG_E2E_APP_PATH 环境变量读取
cdp_port: int = 9222
startup_timeout: int = 30000
headless: bool = False # 是否无头模式运行
@dataclass
class AssertOptions:
"""assert_contains() 方法的选项"""
timeout: int = 5000
match_mode: str = "contains" # 'contains' | 'regex' | 'exact'
message_index: int = -1 # -1 表示最后一条 assistant 消息
include_tools: bool = False

View File

@@ -0,0 +1,36 @@
import pytest
from jiangchang_desktop_sdk import JiangchangDesktopClient, JiangchangMessage, AskOptions, LaunchOptions, AssertOptions
from jiangchang_desktop_sdk.exceptions import JiangchangDesktopError, AppNotFoundError, ConnectionError, TimeoutError, AssertError
def test_client_init():
"""实例化 JiangchangDesktopClient验证 _connected = False"""
client = JiangchangDesktopClient()
assert client._connected is False
assert client._browser is None
assert client._page is None
def test_types_dataclasses():
"""验证 JiangchangMessage, AskOptions 等 dataclass 可以正常实例化"""
msg = JiangchangMessage(id="1", role="user", content="hello", timestamp=123.456)
assert msg.id == "1"
assert msg.role == "user"
ask_opts = AskOptions(timeout=1000)
assert ask_opts.timeout == 1000
launch_opts = LaunchOptions(headless=True)
assert launch_opts.headless is True
assert_opts = AssertOptions(match_mode="exact")
assert assert_opts.match_mode == "exact"
def test_exceptions_are_exceptions():
"""验证所有异常类都是 Exception 的子类"""
assert issubclass(JiangchangDesktopError, Exception)
assert issubclass(AppNotFoundError, JiangchangDesktopError)
assert issubclass(ConnectionError, JiangchangDesktopError)
assert issubclass(ConnectionError, ConnectionError)
assert issubclass(TimeoutError, JiangchangDesktopError)
# Note: in Python 3.10+, TimeoutError is a built-in.
# Our definition: class TimeoutError(JiangchangDesktopError, TimeoutError):
assert issubclass(AssertError, JiangchangDesktopError)