Compare commits
6 Commits
v1.0.10
...
4730cd527d
| Author | SHA1 | Date | |
|---|---|---|---|
| 4730cd527d | |||
| dd6236866e | |||
| 07fa0b0038 | |||
| b1e93a323f | |||
| f214598470 | |||
| 5abe67f340 |
@@ -17,6 +17,8 @@
|
||||
4. 在 `scripts/service/` 中补业务 service 与真正的发布/执行逻辑。
|
||||
5. 用 `python scripts/main.py health` 和 `python scripts/main.py version` 做最小验证。
|
||||
|
||||
如果你的技能在平台里默认是非公开的(`access_scope = 0`),建议在 `SKILL.md` 的 `metadata.openclaw.developer_ids` 中填写开发者用户 ID 列表。这样发布后,平台会自动给这些开发者补可见权限,避免“技能已发布但开发者自己在市场中看不到”。
|
||||
|
||||
开发教程入口:
|
||||
|
||||
- <a href="references/REQUIREMENTS.md" target="_blank" rel="noopener noreferrer">需求文档模板</a>:给技术人员编写和查看研发需求的标准模板
|
||||
@@ -46,3 +48,4 @@ python scripts/main.py version
|
||||
|
||||
- 不要再往模板里引入旧式 `docs/` 或 `optional/` 目录。
|
||||
- 新技能若不需要某些目录,也建议先保留结构,再按实际业务填充内容。
|
||||
- `metadata.openclaw.developer_ids` 是发布元数据,不是用户展示文案。它的作用是让发布后的非公开技能自动授权给指定开发者查看。
|
||||
|
||||
8
SKILL.md
8
SKILL.md
@@ -1,13 +1,16 @@
|
||||
---
|
||||
name: 技能开发模板(复制后请修改)
|
||||
description: "这是 OpenClaw 技能开发模板仓库,不直接作为业务技能发布。复制为新技能仓库后,按本模板替换 slug、名称、说明、CLI 子命令与 service 实现。"
|
||||
version: 1.0.0
|
||||
version: 1.0.13
|
||||
author: 深圳匠厂科技有限公司
|
||||
metadata:
|
||||
openclaw:
|
||||
slug: your-skill-slug
|
||||
emoji: "📦"
|
||||
category: "通用"
|
||||
developer_ids:
|
||||
- 10032
|
||||
- 12428
|
||||
allowed-tools:
|
||||
- bash
|
||||
---
|
||||
@@ -39,4 +42,7 @@ python {baseDir}/scripts/main.py version
|
||||
|
||||
- 复制后请同步修改 `scripts/util/constants.py` 中的 `SKILL_SLUG` / `SKILL_VERSION`。
|
||||
- 如技能无需持久化,可保留 `db/` 目录但不主动调用。
|
||||
- `metadata.openclaw.developer_ids` 用于声明技能发布后的默认开发者可见用户 ID 列表。
|
||||
- 当技能在平台中 `access_scope = 0`(不公开)时,发布流程会把 `developer_ids` 中的用户自动补写到 `skill_user_access`,使这些开发者仍可在技能市场中查看该技能。
|
||||
- `developer_ids` 建议写为正整数数组;第一个 ID 会作为主开发者同步到 `skills.developer_id`。
|
||||
- 面向用户与编排的文档写在 `references/`,不要再新增旧式 `docs/` / `optional/` 结构。
|
||||
|
||||
@@ -186,6 +186,8 @@ scripts/
|
||||
- 平台内部键
|
||||
- 日志 logger 名
|
||||
|
||||
此外,如果该技能发布后默认不公开(`access_scope = 0`),建议一开始就把 `SKILL.md` 中的 `metadata.openclaw.developer_ids` 配好。这样后续发布到平台时,开发者本人仍能在技能市场中看到并验证该技能。
|
||||
|
||||
## 5. 哪些占位内容必须替换
|
||||
|
||||
复制后,至少要全局检查并替换下面这些内容:
|
||||
@@ -212,9 +214,23 @@ scripts/
|
||||
|
||||
- 技能名称
|
||||
- 技能描述
|
||||
- slug
|
||||
- category
|
||||
- dependencies
|
||||
- `slug`
|
||||
作用:技能的唯一英文标识,通常用于仓库名、发布包名、运行时目录名、平台主键匹配等
|
||||
示例:`weibo-publisher`、`douyin-publisher`、`xiaohongshu-publisher`
|
||||
- `category`
|
||||
作用:技能在平台中的分类,用于市场展示与归类,不是代码内部键
|
||||
示例:`通用`、`内容`、`办公协作`、`物流`、`抖音`、`小红书`
|
||||
- `developer_ids`(如需给非公开技能自动补开发者可见权限)
|
||||
作用:声明发布后默认拥有可见权限的开发者用户 ID 列表
|
||||
- `dependencies`
|
||||
作用:声明该技能依赖的兄弟技能或运行前置能力,便于平台或编排层识别依赖关系
|
||||
示例:
|
||||
```yaml
|
||||
dependencies:
|
||||
required:
|
||||
- account-manager
|
||||
- content-manager
|
||||
```
|
||||
- 何时使用本技能
|
||||
- 对用户的引导话术
|
||||
- CLI 使用原则
|
||||
@@ -226,6 +242,33 @@ scripts/
|
||||
- 代码注释
|
||||
- `service/` 实现里
|
||||
|
||||
### 关于 `metadata.openclaw.developer_ids`
|
||||
|
||||
这是一个平台发布元数据字段,用于解决下面这个问题:
|
||||
|
||||
- 技能发布后若平台记录中的 `access_scope = 0`,技能默认不公开
|
||||
- 如果不额外授权,连开发者自己也可能在技能市场里看不到这个技能
|
||||
|
||||
因此可以在 `SKILL.md` 中声明:
|
||||
|
||||
```yaml
|
||||
metadata:
|
||||
openclaw:
|
||||
slug: your-skill-slug
|
||||
category: 通用
|
||||
developer_ids:
|
||||
- 1032
|
||||
- 12428
|
||||
```
|
||||
|
||||
约定如下:
|
||||
|
||||
- 只允许填写正整数用户 ID
|
||||
- 推荐使用数组,即使当前只有 1 个开发者
|
||||
- 发布时平台会把这些用户自动补写到 `skill_user_access`
|
||||
- 第一个 ID 会同步到 `skills.developer_id`
|
||||
- 一期只做“补授权”,不会因为你 later 修改数组而自动撤销旧授权
|
||||
|
||||
## 7. `references/` 应该放什么
|
||||
|
||||
`references/` 是当前规范 skill 的文档中心,建议至少有这些:
|
||||
@@ -350,27 +393,27 @@ scripts/
|
||||
3. 用子进程调用
|
||||
4. 机器可读输出优先 JSON
|
||||
|
||||
## 13. 如何开发发布型 skill
|
||||
## 13. 如何开发一个新 skill
|
||||
|
||||
如果你开发的是 publisher 类 skill,建议按这个顺序做:
|
||||
不管你开发的是发布类、采集类、分析类还是知识库类 skill,建议都先按下面这个顺序推进:
|
||||
|
||||
1. 先把目录结构搭完整
|
||||
2. 先让 `health` / `version` 跑通
|
||||
3. 再让 `publish_service.py` 的骨架跑通
|
||||
4. 再接 `sibling_bridge.py`
|
||||
5. 最后再写 `*_playwright.py`
|
||||
3. 再把核心 `service` 骨架跑通
|
||||
4. 再接兄弟技能桥接、数据库或外部系统
|
||||
5. 最后再补浏览器自动化、复杂流程编排或高风险集成
|
||||
|
||||
不要一开始就直接写页面选择器。
|
||||
不要一开始就直接写页面选择器、复杂接口编排或深层业务逻辑。
|
||||
|
||||
推荐先确保这些基础能力正常:
|
||||
|
||||
- 能取到账号
|
||||
- 能取到文章
|
||||
- 能写日志
|
||||
- CLI 子命令通了
|
||||
- CLI 入口能跑通
|
||||
- 基础命令输出稳定
|
||||
- 关键依赖能取到
|
||||
- 日志或本地状态能落下来
|
||||
- 错误返回值格式定好了
|
||||
|
||||
然后再进浏览器自动化。
|
||||
如果你的 skill 恰好是 publisher 类,可以把上面的“核心 `service`”具体落成 `publish_service.py`,再逐步接 `sibling_bridge.py`、`*_playwright.py`。但这只是示例,不代表模板只适合发布类技能。
|
||||
|
||||
## 14. 本地开发的最小验证顺序
|
||||
|
||||
@@ -387,12 +430,19 @@ python scripts/main.py version
|
||||
|
||||
```bash
|
||||
python scripts/main.py -h
|
||||
python scripts/main.py publish -h
|
||||
python scripts/main.py <your-command> -h
|
||||
```
|
||||
|
||||
### 3. 验证本地日志与数据库
|
||||
|
||||
如果是发布型 skill,再继续:
|
||||
如果你的 skill 需要本地日志或数据库,再继续:
|
||||
|
||||
```bash
|
||||
python scripts/main.py <your-log-command>
|
||||
python scripts/main.py <your-detail-command> <id>
|
||||
```
|
||||
|
||||
如果你沿用了模板中的发布型骨架,那么这里可以具体对应成:
|
||||
|
||||
```bash
|
||||
python scripts/main.py logs
|
||||
@@ -404,7 +454,7 @@ python scripts/main.py log-get 1
|
||||
比如:
|
||||
|
||||
```bash
|
||||
python scripts/main.py publish
|
||||
python scripts/main.py <your-command>
|
||||
```
|
||||
|
||||
## 15. 发布到正式环境验证
|
||||
@@ -419,6 +469,8 @@ python scripts/main.py publish
|
||||
.\release.ps1
|
||||
```
|
||||
|
||||
如果你的技能使用了 `metadata.openclaw.developer_ids`,那么这一步触发的发布工作流除了同步 `skills` / `skill_versions` 外,还会在平台侧自动补开发者可见权限。测试非公开技能时,建议重点验证这部分是否生效。
|
||||
|
||||
这一步会自动完成标准发布动作,包括:
|
||||
|
||||
1. 检查当前仓库状态
|
||||
|
||||
@@ -17,10 +17,26 @@ description: "这是规范化的新技能模板说明,不直接作为业务技
|
||||
## 复制后你需要改什么
|
||||
|
||||
- `SKILL.md` 中的名称、描述、slug、触发说明
|
||||
- `SKILL.md` 中 `metadata.openclaw.developer_ids`(如需让非公开技能默认授权给开发者查看)
|
||||
- `references/CLI.md` 里的命令示例
|
||||
- `scripts/util/constants.py` 中的 slug / 版本 / logger 名
|
||||
- `scripts/service/` 下的真实业务实现
|
||||
|
||||
## `developer_ids` 是做什么的
|
||||
|
||||
`metadata.openclaw.developer_ids` 是平台发布元数据,不是终端用户文案。
|
||||
|
||||
它用于声明:当技能发布后,如果平台侧将该技能设置为 `access_scope = 0`(不公开),哪些开发者用户仍应自动获得可见权限。
|
||||
|
||||
约定如下:
|
||||
|
||||
- 字段位置:`SKILL.md` -> `metadata.openclaw.developer_ids`
|
||||
- 推荐格式:正整数数组,例如 `[1032, 12428]`
|
||||
- 发布效果:发布接口会把这些用户补写到 `skill_user_access`
|
||||
- 第一个 ID 会作为主开发者同步到 `skills.developer_id`
|
||||
|
||||
如果你的技能本来就是公开技能,或暂时不需要开发者专属可见性,这个字段可以留空数组。
|
||||
|
||||
## 不建议再保留的旧结构
|
||||
|
||||
- 旧模板里的 `docs/`
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
"""技能标识与版本(复制后请修改)。"""
|
||||
|
||||
SKILL_SLUG = "your-skill-slug"
|
||||
SKILL_VERSION = "1.0.0"
|
||||
SKILL_VERSION = "1.0.13"
|
||||
LOG_LOGGER_NAME = "openclaw.skill.your_skill_slug"
|
||||
|
||||
10
tests/.gitignore
vendored
Normal file
10
tests/.gitignore
vendored
Normal file
@@ -0,0 +1,10 @@
|
||||
# 测试运行产物与本地密钥,勿提交
|
||||
__pycache__/
|
||||
*.py[cod]
|
||||
.pytest_cache/
|
||||
desktop/artifacts/
|
||||
integration/artifacts/
|
||||
*.log
|
||||
.env
|
||||
.env.*
|
||||
!.env.sample
|
||||
173
tests/README.md
173
tests/README.md
@@ -1,8 +1,171 @@
|
||||
# tests
|
||||
# skill-template 测试说明(企业数字员工技能)
|
||||
|
||||
这里放模板复制后的单元测试或最小回归测试。
|
||||
本目录提供面向 **ERP / TMS / WMS / 船司 / 报关 / 邮件 / 浏览器 / LLM / RPA** 等外联场景的**分层测试骨架**:默认仅内存与本地 SQLite;**真实 API / 真实 RPA** 必须通过环境变量显式授权,且仅以 ``*.sample`` 或复制后的文件提供范式。
|
||||
|
||||
模板仓库本身不强制附带业务测试,但建议新 skill 至少补:
|
||||
复制为新技能后,请保留隔离数据根与 profile 策略,再按域扩展用例。
|
||||
|
||||
- `health` / `version` 最小冒烟测试
|
||||
- 关键 `service` 层函数的单元测试
|
||||
---
|
||||
|
||||
## 我该把测试写在哪里?
|
||||
|
||||
人类开发者与 Cursor 等 AI 工具在加用例前,先对照下表决定**文件落点**与**是否默认执行**:
|
||||
|
||||
| 场景 | 放在哪里 | 默认是否运行 | 说明 |
|
||||
|------|----------|--------------|------|
|
||||
| CLI / health / version / metadata / runtime 路径 | `tests/test_*.py` | 是 | 默认必跑,保持轻量、无外联 |
|
||||
| 纯函数 / 业务规则 | `tests/test_*.py` | 是 | 不依赖真实外部系统 |
|
||||
| service 层契约 | 从 `tests/samples/test_service_contract.py.sample` 复制到 `tests/test_service_contract.py` | 是,复制后 | 用 `FakeAdapter` / stub 注入外部依赖 |
|
||||
| golden fixture 回归 | 从 `tests/samples/test_golden_cases.py.sample` 复制到 `tests/test_golden_cases.py` | 是,复制后 | 输入样例 + 期望输出,适合解析、计算、校验类技能 |
|
||||
| 仿真 API | `tests/integration/test_simulator_api_contract.py.sample` | 否 | 复制并设置 `OPENCLAW_TEST_TARGET=simulator_api`(或兼容键名 `OPENCLOW_TEST_TARGET`)后按需运行 |
|
||||
| 仿真 RPA / 页面操作 | `tests/integration/test_simulator_rpa_contract.py.sample` | 否 | 用 localhost / 仿真页面,不碰真实系统 |
|
||||
| 真实 API | `tests/integration/test_real_api_contract.py.sample` | 否 | 必须 `OPENCLAW_TEST_TARGET=real_api` 且 `ALLOW_REAL_API=1` |
|
||||
| 真实 RPA | `tests/integration/test_real_rpa_contract.py.sample` | 否 | 最高风险,只能手动触发 |
|
||||
| 桌面宿主 E2E | `tests/desktop/test_desktop_smoke.py.sample` | 否 | 需要 pytest + `jiangchang_desktop_sdk` |
|
||||
|
||||
说明:`python tests/run_tests.py` **只**收集 `tests/` **根目录**下的 `test_*.py`;子目录里的 `*.sample` 与 desktop 用例**不会**被默认发现。
|
||||
|
||||
---
|
||||
|
||||
## 新技能最小测试清单
|
||||
|
||||
从本模板复制出新技能仓库后,建议至少逐项确认:
|
||||
|
||||
- [ ] `python tests/run_tests.py -v` 能通过。
|
||||
- [ ] `python scripts/main.py health` 能通过。
|
||||
- [ ] `python scripts/main.py version` 输出 JSON,且 `skill` 与目录名 / `SKILL.md` / `constants.SKILL_SLUG` 一致。
|
||||
- [ ] 所有 DB / 文件写入都在 `IsolatedDataRoot`(或等价隔离)下测试,不写真实数据目录。
|
||||
- [ ] 外部系统默认使用 mock / `FakeAdapter`,不访问真实 API,不打开真实 RPA。
|
||||
- [ ] 至少有 1 个成功路径测试。
|
||||
- [ ] 至少有 1 个缺必填字段 / 非法输入测试。
|
||||
- [ ] 如果有 adapter,至少覆盖 timeout / unauthorized / invalid response(可用 fake 模拟)。
|
||||
- [ ] 如果有解析 / 计算 / 校验类业务,至少保留 1 组 golden fixture(脱敏)。
|
||||
- [ ] 真实 API / 真实 RPA 测试只放 `tests/integration/`,并且默认 `.sample`,不默认运行。
|
||||
|
||||
---
|
||||
|
||||
## 1. 默认必跑(unittest)
|
||||
|
||||
在**技能仓库根目录**执行:
|
||||
|
||||
```powershell
|
||||
$env:PYTHONDONTWRITEBYTECODE = "1"
|
||||
python tests/run_tests.py
|
||||
python tests/run_tests.py -v
|
||||
python tests/run_tests.py cli_smoke
|
||||
python tests/run_tests.py test_cli_smoke
|
||||
```
|
||||
|
||||
### 1.1 收集范围
|
||||
|
||||
- ``run_tests.py`` **只**发现 ``tests/`` **根目录**下 ``test_*.py``。
|
||||
- **不递归**:``tests/samples/``、``tests/integration/``、``tests/desktop/``(其中文件为 ``*.sample`` 或需 pytest 单独跑)。
|
||||
- 启动时把 ``scripts/`` 与 ``tests/`` 加入 ``sys.path``。
|
||||
- Windows 下将 **stdout/stderr** 包装为 **UTF-8**。
|
||||
|
||||
### 1.2 默认套件覆盖
|
||||
|
||||
| 能力 | 文件 |
|
||||
|------|------|
|
||||
| CLI 冒烟(import ``cli.app.main``) | ``test_cli_smoke.py`` |
|
||||
| **subprocess 真实入口** ``python scripts/main.py`` | ``test_entrypoint_subprocess.py`` |
|
||||
| 运行路径与 **CLAW_*/JIANGCHANG_*** 隔离 | ``test_runtime_paths.py`` |
|
||||
| ``SKILL.md`` slug 与 ``constants.SKILL_SLUG`` | ``test_skill_metadata.py`` |
|
||||
| DB / ``publish_logs`` 骨架冒烟 | ``test_db_smoke.py`` |
|
||||
| **adapter profile / 外呼授权策略** | ``test_adapter_profile_policy.py`` + ``adapter_test_utils.py``(非 ``test_`` 前缀,不自动当用例收集) |
|
||||
|
||||
### 1.3 数据隔离(``_support.IsolatedDataRoot``)
|
||||
|
||||
同时设置(进入同一临时目录 / 用户):
|
||||
|
||||
| ``CLAW_DATA_ROOT`` | ``JIANGCHANG_DATA_ROOT`` |
|
||||
|--------------------|---------------------------|
|
||||
| 同一 ``tempfile.mkdtemp(prefix="skill-test-")`` | 同上 |
|
||||
|
||||
| ``CLAW_USER_ID`` | ``JIANGCHANG_USER_ID`` |
|
||||
|------------------|-------------------------|
|
||||
| ``_test`` | ``_test`` |
|
||||
|
||||
退出时四个键均恢复(原不存在则 ``pop``),并删除临时目录。
|
||||
|
||||
### 1.4 产物与密钥
|
||||
|
||||
- ``tests/.gitignore`` 忽略 ``__pycache__``、``.pytest_cache``、``*.pyc``、``desktop/artifacts/``、``integration/artifacts/``、``.env`` 等。
|
||||
- 推荐运行前设置 ``PYTHONDONTWRITEBYTECODE=1`` 减少 ``.pyc``。
|
||||
|
||||
---
|
||||
|
||||
## 2. 可选:Service / Golden(``tests/samples/*.sample``)
|
||||
|
||||
- 展示 **service 契约** 与 **fixture 回归** 写法;**默认不执行**。
|
||||
- 启用:复制 ``tests/samples/test_service_contract.py.sample`` 等为 ``tests/test_*.py``(去掉 ``.sample``),按业务修改后再 ``python tests/run_tests.py``。
|
||||
- 脱敏样例数据见 ``tests/fixtures/README.md`` 与 ``sample_request.json`` / ``expected_response.json``;profile 结构见 ``adapter_profiles.sample.yaml``。
|
||||
|
||||
---
|
||||
|
||||
## 3. 可选:Integration(``tests/integration/*.sample``)
|
||||
|
||||
- **默认不跑**;文件均为 ``*.py.sample``。
|
||||
- 说明与开关见 **`tests/integration/README.md`**。
|
||||
- **禁止**在样例中硬编码真实 token、密码、生产 URL;凭证须来自密钥库或受控环境变量。
|
||||
|
||||
---
|
||||
|
||||
## 4. 可选:Desktop E2E(pytest)
|
||||
|
||||
见 **`tests/desktop/README.md`**。示例为 `test_desktop_smoke.py.sample`,需复制为 `test_desktop_smoke.py` 后用 pytest 运行。
|
||||
|
||||
---
|
||||
|
||||
## 5. 环境变量:测试档位与授权
|
||||
|
||||
### 5.1 测试档位(二选一变量名,后者为历史拼写兼容)
|
||||
|
||||
- ``OPENCLAW_TEST_TARGET`` 或 ``OPENCLOW_TEST_TARGET``
|
||||
|
||||
**合法取值**(非法值将使 ``get_test_target()`` 抛 ``ValueError``):
|
||||
|
||||
| 取值 | 含义 |
|
||||
|------|------|
|
||||
| ``unit`` | **默认**:单元 / 内存 / mock,不跑仿真与真实外联 |
|
||||
| ``mock`` | 与 ``unit`` 类似的安全档位(显式语义) |
|
||||
| ``simulator_api`` | 仅允许 **仿真 HTTP** 类集成(如 localhost) |
|
||||
| ``simulator_rpa`` | 仅允许 **仿真页面 / 录播 RPA** |
|
||||
| ``real_api`` | 真实 API(另需 ``ALLOW_REAL_API=1``) |
|
||||
| ``real_rpa`` | 真实 RPA(另需 ``ALLOW_REAL_RPA=1``) |
|
||||
|
||||
未设置时等价 ``unit``。
|
||||
|
||||
### 5.2 授权开关(显式 ``1``)
|
||||
|
||||
| 变量 | 作用 |
|
||||
|------|------|
|
||||
| ``ALLOW_REAL_API=1`` | 允许 ``real_api`` profile 访问真实 HTTP 通道 |
|
||||
| ``ALLOW_REAL_RPA=1`` | 允许 ``real_rpa`` profile 驱动真实浏览器/RPA |
|
||||
| ``ALLOW_WRITE_ACTIONS=1`` | 在 ``real_*`` 下允许**写**操作(提交表单、下单等) |
|
||||
|
||||
### 5.3 默认安全策略(``adapter_test_utils``)
|
||||
|
||||
- 默认 ``unit``:不访问真实 API、不跑真实 RPA、不做写操作、不读取真实密钥、不写真实数据根。
|
||||
- ``assert_no_real_network_env()`` 断言默认未误开 ``ALLOW_REAL_*``(用于必跑用例自检)。
|
||||
|
||||
---
|
||||
|
||||
## AI 编程工具注意事项(Cursor 等)
|
||||
|
||||
以下约束适用于为本仓库或复制出的技能编写/修改测试时,**避免误伤业务与生产**:
|
||||
|
||||
- **不要为了**让测试通过去改业务代码,除非用户明确要求修改实现。
|
||||
- **不要**把真实外部系统调用写进默认的 `tests/test_*.py`(默认套件须无外联、无真实浏览器)。
|
||||
- **不要**把 `*.sample` 改名为可执行测试或移入根目录并去掉 `.sample`,除非用户明确要求启用 integration / 真实联调。
|
||||
- **不要**硬编码真实凭证、token、cookie、生产 URL;占位请用明显虚构域名(如 `example.invalid`)或文档约定的 `credential_ref`。
|
||||
- 默认测试只能使用 **mock、stub、`FakeAdapter`、fixtures、隔离 SQLite** 等安全手段。
|
||||
- 业务逻辑应优先在 **service 层** 编写与测试,不要只断言 CLI 文案。
|
||||
- 异常路径应返回**结构化错误**(如统一错误码/字段),避免裸异常直接穿透到 CLI。
|
||||
- 有真实系统联调需求时,测试应写在 `tests/integration/`,并配合 `OPENCLAW_TEST_TARGET` / `ALLOW_REAL_*` / `ALLOW_WRITE_ACTIONS` 等**显式开关**;不得默认开启。
|
||||
|
||||
---
|
||||
|
||||
## 6. 与 content-manager 的关系
|
||||
|
||||
- **未**复制 article / media / prompt / NotebookLM 等业务测试。
|
||||
- 本骨架面向**通用外联技能**;业务契约请放在各自技能的 ``tests/samples`` / ``tests/integration`` 复制件中实现。
|
||||
|
||||
77
tests/_support.py
Normal file
77
tests/_support.py
Normal file
@@ -0,0 +1,77 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""
|
||||
测试公共支撑。
|
||||
|
||||
1. 将 scripts/ 加入 sys.path,便于 `from cli…` / `from db…` / `from util…` 等导入。
|
||||
2. `IsolatedDataRoot`:同时设置 CLAW_* 与 JIANGCHANG_* 数据根与用户 ID,
|
||||
与 `jiangchang_skill_core.runtime_env` 的优先级一致(CLAW_* 优先),
|
||||
避免只设 JIANGCHANG_* 时仍读到真实 CLAW_DATA_ROOT。
|
||||
退出时四个变量均恢复;原不存在的键会删除;并 rmtree 临时目录。
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import shutil
|
||||
import sys
|
||||
import tempfile
|
||||
from typing import Optional
|
||||
|
||||
_TESTS_DIR = os.path.dirname(os.path.abspath(__file__))
|
||||
_SKILL_ROOT = os.path.abspath(os.path.join(_TESTS_DIR, ".."))
|
||||
_SCRIPTS_DIR = os.path.join(_SKILL_ROOT, "scripts")
|
||||
|
||||
if _SCRIPTS_DIR not in sys.path:
|
||||
sys.path.insert(0, _SCRIPTS_DIR)
|
||||
|
||||
_ISOLATION_KEYS = (
|
||||
"CLAW_DATA_ROOT",
|
||||
"JIANGCHANG_DATA_ROOT",
|
||||
"CLAW_USER_ID",
|
||||
"JIANGCHANG_USER_ID",
|
||||
)
|
||||
|
||||
|
||||
def get_scripts_dir() -> str:
|
||||
return _SCRIPTS_DIR
|
||||
|
||||
|
||||
def get_skill_root() -> str:
|
||||
return _SKILL_ROOT
|
||||
|
||||
|
||||
class IsolatedDataRoot:
|
||||
"""
|
||||
在独立临时目录下模拟数据根,避免污染本机默认数据目录。
|
||||
|
||||
进入上下文时(同一套隔离值写入 CLAW_* 与 JIANGCHANG_*):
|
||||
- CLAW_DATA_ROOT / JIANGCHANG_DATA_ROOT -> 同一 tempfile.mkdtemp(prefix='skill-test-')
|
||||
- CLAW_USER_ID / JIANGCHANG_USER_ID -> _test(与业务默认用户区分)
|
||||
|
||||
退出时:四个变量均恢复为进入前状态(若原先未设置则 pop),并删除临时目录。
|
||||
"""
|
||||
|
||||
def __init__(self, user_id: str = "_test") -> None:
|
||||
self._tmp: Optional[str] = None
|
||||
self._old: dict[str, Optional[str]] = {}
|
||||
self._user_id = user_id
|
||||
|
||||
def __enter__(self) -> str:
|
||||
self._tmp = tempfile.mkdtemp(prefix="skill-test-")
|
||||
for k in _ISOLATION_KEYS:
|
||||
self._old[k] = os.environ.get(k)
|
||||
os.environ["CLAW_DATA_ROOT"] = self._tmp
|
||||
os.environ["JIANGCHANG_DATA_ROOT"] = self._tmp
|
||||
os.environ["CLAW_USER_ID"] = self._user_id
|
||||
os.environ["JIANGCHANG_USER_ID"] = self._user_id
|
||||
return self._tmp
|
||||
|
||||
def __exit__(self, *exc_info: object) -> None:
|
||||
for k in _ISOLATION_KEYS:
|
||||
v = self._old.get(k)
|
||||
if v is None:
|
||||
os.environ.pop(k, None)
|
||||
else:
|
||||
os.environ[k] = v
|
||||
if self._tmp and os.path.isdir(self._tmp):
|
||||
shutil.rmtree(self._tmp, ignore_errors=True)
|
||||
self._tmp = None
|
||||
142
tests/adapter_test_utils.py
Normal file
142
tests/adapter_test_utils.py
Normal file
@@ -0,0 +1,142 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""
|
||||
企业技能对接外部系统(ERP / TMS / WMS / 船司 / 报关 / 邮件 / 浏览器 / LLM / RPA)时,
|
||||
测试侧共用的 **profile / 授权 / 禁止真实外呼** 工具。
|
||||
|
||||
- 仅标准库;不发起 HTTP;不启动浏览器。
|
||||
- 真实 API / RPA 契约测试请放在 ``tests/integration/*.sample``,并配合环境变量显式授权。
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
from typing import Any, Final
|
||||
|
||||
# 允许的 OPENCLAW_TEST_TARGET / OPENCLOW_TEST_TARGET 取值(后者为历史拼写兼容)
|
||||
ALLOWED_TEST_TARGETS: Final[frozenset[str]] = frozenset(
|
||||
{
|
||||
"unit",
|
||||
"mock",
|
||||
"simulator_api",
|
||||
"simulator_rpa",
|
||||
"real_api",
|
||||
"real_rpa",
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
def _raw_test_target_env() -> str:
|
||||
return (os.environ.get("OPENCLAW_TEST_TARGET") or os.environ.get("OPENCLOW_TEST_TARGET") or "").strip()
|
||||
|
||||
|
||||
def get_test_target() -> str:
|
||||
"""
|
||||
读取 ``OPENCLAW_TEST_TARGET`` 或 ``OPENCLOW_TEST_TARGET``(兼容旧拼写)。
|
||||
|
||||
未设置时返回 ``unit``。非法取值抛出 ``ValueError``,便于尽早发现拼写错误。
|
||||
"""
|
||||
raw = _raw_test_target_env()
|
||||
if not raw:
|
||||
return "unit"
|
||||
if raw not in ALLOWED_TEST_TARGETS:
|
||||
raise ValueError(
|
||||
f"Invalid test target {raw!r}. Allowed: {sorted(ALLOWED_TEST_TARGETS)}"
|
||||
)
|
||||
return raw
|
||||
|
||||
|
||||
def real_access_allowed(profile: str) -> bool:
|
||||
"""
|
||||
判断当前环境是否允许对 ``profile`` 做「真实通道」访问(仍不代替业务鉴权)。
|
||||
|
||||
- ``real_api``:需要 ``ALLOW_REAL_API == "1"``。
|
||||
- ``real_rpa``:需要 ``ALLOW_REAL_RPA == "1"``。
|
||||
- 写操作另需 ``ALLOW_WRITE_ACTIONS == "1"``(由 ``should_skip_profile(..., write=True)`` 组合检查)。
|
||||
"""
|
||||
if profile == "real_api":
|
||||
return os.environ.get("ALLOW_REAL_API", "") == "1"
|
||||
if profile == "real_rpa":
|
||||
return os.environ.get("ALLOW_REAL_RPA", "") == "1"
|
||||
if profile in ("mock", "simulator_api", "simulator_rpa"):
|
||||
return True
|
||||
raise ValueError(f"unknown profile for real_access_allowed: {profile!r}")
|
||||
|
||||
|
||||
def should_skip_profile(profile: str, write: bool = False) -> tuple[bool, str]:
|
||||
"""
|
||||
返回 ``(should_skip, reason)``。
|
||||
|
||||
策略摘要:
|
||||
- ``mock``:永不跳过(内存 / FakeAdapter,无外呼)。
|
||||
- ``simulator_api``:仅当 ``get_test_target() == "simulator_api"`` 时不跳过。
|
||||
- ``simulator_rpa``:仅当 target 为 ``simulator_rpa`` 时不跳过。
|
||||
- ``real_api``:target 须为 ``real_api`` 且 ``ALLOW_REAL_API=1``;``write=True`` 还须 ``ALLOW_WRITE_ACTIONS=1``。
|
||||
- ``real_rpa``:target 须为 ``real_rpa`` 且 ``ALLOW_REAL_RPA=1``;``write=True`` 还须 ``ALLOW_WRITE_ACTIONS=1``。
|
||||
"""
|
||||
target = get_test_target()
|
||||
|
||||
if profile == "mock":
|
||||
return (False, "")
|
||||
|
||||
if profile == "simulator_api":
|
||||
if target != "simulator_api":
|
||||
return (True, f"OPENCLAW_TEST_TARGET must be simulator_api (got {target!r})")
|
||||
return (False, "")
|
||||
|
||||
if profile == "simulator_rpa":
|
||||
if target != "simulator_rpa":
|
||||
return (True, f"OPENCLAW_TEST_TARGET must be simulator_rpa (got {target!r})")
|
||||
return (False, "")
|
||||
|
||||
if profile == "real_api":
|
||||
if target != "real_api":
|
||||
return (True, f"OPENCLAW_TEST_TARGET must be real_api (got {target!r})")
|
||||
if os.environ.get("ALLOW_REAL_API", "") != "1":
|
||||
return (True, "ALLOW_REAL_API is not 1")
|
||||
if write and os.environ.get("ALLOW_WRITE_ACTIONS", "") != "1":
|
||||
return (True, "ALLOW_WRITE_ACTIONS is not 1 (write requested)")
|
||||
return (False, "")
|
||||
|
||||
if profile == "real_rpa":
|
||||
if target != "real_rpa":
|
||||
return (True, f"OPENCLAW_TEST_TARGET must be real_rpa (got {target!r})")
|
||||
if os.environ.get("ALLOW_REAL_RPA", "") != "1":
|
||||
return (True, "ALLOW_REAL_RPA is not 1")
|
||||
if write and os.environ.get("ALLOW_WRITE_ACTIONS", "") != "1":
|
||||
return (True, "ALLOW_WRITE_ACTIONS is not 1 (write requested)")
|
||||
return (False, "")
|
||||
|
||||
raise ValueError(f"unknown profile: {profile!r}")
|
||||
|
||||
|
||||
class FakeAdapter:
|
||||
"""
|
||||
模拟外部系统 adapter,用于单元测试中的超时 / 鉴权 / 畸形响应等分支。
|
||||
|
||||
``mode``:``success`` | ``timeout`` | ``invalid_response`` | ``unauthorized``
|
||||
"""
|
||||
|
||||
def __init__(self, mode: str = "success") -> None:
|
||||
if mode not in ("success", "timeout", "invalid_response", "unauthorized"):
|
||||
raise ValueError(f"unsupported FakeAdapter mode: {mode!r}")
|
||||
self._mode = mode
|
||||
|
||||
def call(self, payload: Any) -> Any:
|
||||
if self._mode == "success":
|
||||
return {"success": True, "data": {"echo": payload}}
|
||||
if self._mode == "timeout":
|
||||
raise TimeoutError("simulated upstream timeout")
|
||||
if self._mode == "invalid_response":
|
||||
return object() # 非 dict,模拟无法解析的响应体
|
||||
if self._mode == "unauthorized":
|
||||
raise PermissionError("simulated 401/403")
|
||||
raise RuntimeError("unreachable")
|
||||
|
||||
|
||||
def assert_no_real_network_env() -> None:
|
||||
"""
|
||||
断言当前进程未显式打开「真实外呼」授权开关(默认安全 CI / 本地开发)。
|
||||
|
||||
若需在受控环境跑真实集成,请显式设置 ``ALLOW_REAL_*`` 并使用 ``tests/integration/*.sample``。
|
||||
"""
|
||||
assert os.environ.get("ALLOW_REAL_API", "") != "1", "ALLOW_REAL_API must not be 1 in default unit tests"
|
||||
assert os.environ.get("ALLOW_REAL_RPA", "") != "1", "ALLOW_REAL_RPA must not be 1 in default unit tests"
|
||||
34
tests/desktop/README.md
Normal file
34
tests/desktop/README.md
Normal file
@@ -0,0 +1,34 @@
|
||||
# 可选桌面 E2E(pytest)
|
||||
|
||||
本目录用于**匠厂桌面 + jiangchang_desktop_sdk** 的端到端冒烟,**不是** `python tests/run_tests.py` 默认路径的一部分。
|
||||
|
||||
## 依赖
|
||||
|
||||
- `pytest`
|
||||
- `jiangchang_desktop_sdk`(由 `conftest.py` 按多级路径尝试加入 `sys.path`,见该文件说明)
|
||||
- 匠厂桌面已安装或开发态可启动,且 `jiangchang://` 协议可用
|
||||
|
||||
## 为什么不默认跑
|
||||
|
||||
- 需要真实宿主环境,无法在纯 CI/无头环境稳定复现。
|
||||
- `unittest` 的 `tests/run_tests.py` **只发现** `tests/` 根目录下 `test_*.py`,**不递归**本目录。
|
||||
- 示例文件为 **`test_desktop_smoke.py.sample`**,避免被误收集。
|
||||
|
||||
## 如何启用
|
||||
|
||||
1. 将 `test_desktop_smoke.py.sample` **复制或重命名**为 `test_desktop_smoke.py`。
|
||||
2. 按你的技能名称、路由关键词、期望输出修改 `ask` 与断言。
|
||||
3. 运行:
|
||||
|
||||
```powershell
|
||||
pytest tests/desktop/test_desktop_smoke.py -v -s
|
||||
```
|
||||
|
||||
失败时,`conftest.py` 中的 `failure_snapshot_dir` fixture 会将现场目录指向 `tests/desktop/artifacts/`(需自行避免将大文件提交进仓库)。
|
||||
|
||||
## 与后端测试的关系
|
||||
|
||||
- **后端 / CLI / DB**:请使用仓库根目录下 `python tests/run_tests.py`(见 `tests/README.md`)。
|
||||
- **桌面 E2E**:仅在本目录用 **pytest** 维护,二者互补、互不替代。
|
||||
|
||||
若仅需验证 **service 业务逻辑**、契约与 golden 回归,**不要**写 desktop E2E;desktop 只用于验证**宿主 UI、路由、`jiangchang://` 协议、真实用户交互链路**等与桌面宿主强相关的行为。
|
||||
76
tests/desktop/conftest.py
Normal file
76
tests/desktop/conftest.py
Normal file
@@ -0,0 +1,76 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""
|
||||
可选桌面 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 给出可操作的排障说明。
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import importlib.util
|
||||
import os
|
||||
import sys
|
||||
|
||||
import pytest
|
||||
|
||||
|
||||
def _bootstrap_jiangchang_desktop_sdk() -> None:
|
||||
if importlib.util.find_spec("jiangchang_desktop_sdk") is not None:
|
||||
return
|
||||
|
||||
env_kit_root = (os.environ.get("JIANGCHANG_KIT_ROOT") or "").strip()
|
||||
|
||||
here = os.path.dirname(os.path.abspath(__file__))
|
||||
skill_root = os.path.abspath(os.path.join(here, "..", ".."))
|
||||
workspace_root = os.path.abspath(os.path.join(skill_root, ".."))
|
||||
open_claw_root = os.path.abspath(os.path.join(workspace_root, ".."))
|
||||
|
||||
candidates = []
|
||||
if env_kit_root:
|
||||
candidates.append(os.path.join(env_kit_root, "sdk", "jiangchang-desktop-sdk", "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")
|
||||
)
|
||||
|
||||
tried = []
|
||||
for cand in candidates:
|
||||
cand_abs = os.path.abspath(cand)
|
||||
tried.append(cand_abs)
|
||||
if os.path.isdir(cand_abs) and os.path.isdir(os.path.join(cand_abs, "jiangchang_desktop_sdk")):
|
||||
if cand_abs not in sys.path:
|
||||
sys.path.insert(0, cand_abs)
|
||||
return
|
||||
|
||||
pytest.exit(
|
||||
"无法定位 jiangchang_desktop_sdk。请任选一种方式解决:\n"
|
||||
" 1) pip install -e <path-to>/jiangchang-platform-kit/sdk/jiangchang-desktop-sdk\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"
|
||||
"已尝试的候选路径:\n - " + "\n - ".join(tried),
|
||||
returncode=4,
|
||||
)
|
||||
|
||||
|
||||
_bootstrap_jiangchang_desktop_sdk()
|
||||
|
||||
_ARTIFACT_DIR = os.path.join(os.path.dirname(__file__), "artifacts")
|
||||
|
||||
|
||||
@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)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def failure_snapshot_dir() -> str:
|
||||
os.makedirs(_ARTIFACT_DIR, exist_ok=True)
|
||||
return _ARTIFACT_DIR
|
||||
58
tests/desktop/test_desktop_smoke.py.sample
Normal file
58
tests/desktop/test_desktop_smoke.py.sample
Normal file
@@ -0,0 +1,58 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""
|
||||
可选桌面 E2E 示例(默认不执行:文件名后缀为 .sample)。
|
||||
|
||||
启用方式:
|
||||
复制本文件为 test_desktop_smoke.py,按目标技能修改提问文案与断言。
|
||||
运行:pytest tests/desktop/test_desktop_smoke.py -v -s
|
||||
|
||||
依赖:pytest、jiangchang_desktop_sdk、匠厂桌面宿主已就绪。
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
import os
|
||||
|
||||
import pytest
|
||||
|
||||
from jiangchang_desktop_sdk import AskOptions, JiangchangDesktopClient
|
||||
|
||||
_logger = logging.getLogger("skill-template-desktop-e2e-sample")
|
||||
if not _logger.handlers:
|
||||
_logger.setLevel(logging.INFO)
|
||||
_h = logging.StreamHandler()
|
||||
_h.setFormatter(logging.Formatter("[E2E-sample] %(message)s"))
|
||||
_logger.addHandler(_h)
|
||||
|
||||
_ARTIFACT_DIR = os.path.join(os.path.dirname(__file__), "artifacts")
|
||||
|
||||
|
||||
class TestDesktopSmokeSample:
|
||||
"""最小 smoke:拉起桌面、等待网关、发一条与技能路由相关的问句并做宽松断言。"""
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def connect_and_teardown(self, request):
|
||||
client = JiangchangDesktopClient()
|
||||
_logger.info("ensure_app_running")
|
||||
client.ensure_app_running(wait_timeout_s=30)
|
||||
client.bring_to_front()
|
||||
client.wait_gateway_ready(timeout_ms=30000)
|
||||
self.client = client
|
||||
yield
|
||||
try:
|
||||
rep = getattr(request.node, "rep_call", None)
|
||||
if rep is not None and rep.failed:
|
||||
os.makedirs(_ARTIFACT_DIR, exist_ok=True)
|
||||
paths = client.snapshot(_ARTIFACT_DIR, tag=request.node.name)
|
||||
_logger.error("snapshot: %s", paths)
|
||||
finally:
|
||||
client.disconnect()
|
||||
|
||||
def test_gateway_and_simple_ask(self) -> None:
|
||||
"""请将 SKILL 名与路由关键词改成你复制后的真实技能。"""
|
||||
answer = self.client.ask(
|
||||
"请使用当前已安装技能执行一次 health 检查并返回 OK 或等价结果。",
|
||||
AskOptions(timeout=120000, new_task=True),
|
||||
)
|
||||
assert answer
|
||||
assert "OK" in answer or "ok" in answer.lower() or "健康" in answer
|
||||
27
tests/fixtures/README.md
vendored
Normal file
27
tests/fixtures/README.md
vendored
Normal file
@@ -0,0 +1,27 @@
|
||||
# fixtures(脱敏样例)
|
||||
|
||||
本目录存放 **golden case / contract test** 用的输入、期望输出与 **adapter profile 样例**。
|
||||
|
||||
## 规则
|
||||
|
||||
- 只能放**虚构、脱敏、可公开**的数据(演示租户、演示 request_id 等)。
|
||||
- **禁止**:真实客户名、真实订单/提单号、真实报关单、真实 token/cookie、真实内网 URL、生产环境凭证。
|
||||
- `adapter_profiles.sample.yaml` 中的 URL 须使用 `localhost`、`example.invalid` 等明显占位;`credential_ref` 仅表达「从密钥库引用」的约定,**不得**内嵌密钥。
|
||||
|
||||
## 与本仓库测试的对应关系
|
||||
|
||||
| 文件 | 用途 |
|
||||
|------|------|
|
||||
| `sample_request.json` | 通用请求体样例,供 ``tests/samples/test_golden_cases.py.sample`` 演示读取。 |
|
||||
| `expected_response.json` | 与上配套的期望摘要字段(业务复制后按域替换)。 |
|
||||
| `adapter_profiles.sample.yaml` | 五类运行方式(mock / simulator_api / simulator_rpa / real_api / real_rpa)的配置表达样例。 |
|
||||
|
||||
---
|
||||
|
||||
## 给 AI 工具的替换规则
|
||||
|
||||
- 可以按业务**重命名字段**、增减键名,但必须保持**虚构、脱敏**;不得把真实客户标识原样搬进仓库。
|
||||
- 可以**新增**更多 fixture 文件,例如 `case_valid_001.json`、`case_missing_required_001.json`,并在测试中通过相对路径读取。
|
||||
- **禁止**将真实企业数据、真实单证截图中的文字、真实报关要素未经处理直接复制进 `fixtures/`。
|
||||
- 若用例来自真实客户样本,须先**脱敏、泛化、替换编号与地名**,再作为模板数据提交。
|
||||
|
||||
40
tests/fixtures/adapter_profiles.sample.yaml
vendored
Normal file
40
tests/fixtures/adapter_profiles.sample.yaml
vendored
Normal file
@@ -0,0 +1,40 @@
|
||||
# 样例:五类 adapter 运行方式(复制到业务仓后按环境改名;勿提交真实凭证)
|
||||
runtime:
|
||||
profile: mock
|
||||
|
||||
adapters:
|
||||
demo_external_system:
|
||||
interface: demo
|
||||
profile: mock
|
||||
|
||||
mock:
|
||||
target: mock
|
||||
channel: none
|
||||
|
||||
simulator_api:
|
||||
target: simulator
|
||||
channel: api
|
||||
base_url: http://localhost:5180/api/demo
|
||||
timeout_seconds: 10
|
||||
|
||||
simulator_rpa:
|
||||
target: simulator
|
||||
channel: rpa
|
||||
entry_url: http://localhost:5180/select
|
||||
timeout_seconds: 30
|
||||
|
||||
real_api:
|
||||
target: real
|
||||
channel: api
|
||||
base_url: https://example.invalid/api
|
||||
credential_ref: vault://tenant/demo/external-system/api
|
||||
write_enabled: false
|
||||
timeout_seconds: 20
|
||||
|
||||
real_rpa:
|
||||
target: real
|
||||
channel: rpa
|
||||
entry_url: https://example.invalid/portal
|
||||
credential_ref: vault://tenant/demo/external-system/rpa
|
||||
write_enabled: false
|
||||
timeout_seconds: 60
|
||||
6
tests/fixtures/expected_response.json
vendored
Normal file
6
tests/fixtures/expected_response.json
vendored
Normal file
@@ -0,0 +1,6 @@
|
||||
{
|
||||
"success": true,
|
||||
"data": {
|
||||
"status": "ok"
|
||||
}
|
||||
}
|
||||
11
tests/fixtures/sample_request.json
vendored
Normal file
11
tests/fixtures/sample_request.json
vendored
Normal file
@@ -0,0 +1,11 @@
|
||||
{
|
||||
"request_id": "req_test_001",
|
||||
"tenant_id": "tenant_demo",
|
||||
"actor": "tester",
|
||||
"parameters": {
|
||||
"input_text": "demo input",
|
||||
"options": {
|
||||
"dry_run": true
|
||||
}
|
||||
}
|
||||
}
|
||||
53
tests/integration/README.md
Normal file
53
tests/integration/README.md
Normal file
@@ -0,0 +1,53 @@
|
||||
# integration(可选集成 / 真实外联)
|
||||
|
||||
本目录**默认不执行**:
|
||||
|
||||
- ``python tests/run_tests.py`` 只收集 ``tests/`` 根目录下 ``test_*.py``,**不会**进入本目录。
|
||||
- 本目录下文件一律为 ``*.py.sample``,pytest / unittest **默认也不会**按普通 ``test_*.py`` 收集。
|
||||
|
||||
## 与 samples 的区别
|
||||
|
||||
| 维度 | `tests/samples/` | `tests/integration/` |
|
||||
|------|------------------|------------------------|
|
||||
| 目的 | 默认可**升级**为根目录 `test_*.py` 的 **业务单元 / service 契约 / golden** 范式 | **外联联调**范式:网络、浏览器、仿真或真实系统 |
|
||||
| 默认执行 | `.sample` 不执行;复制到根目录并改名后才进入默认套件 | **永远不**作为默认 unittest 的一部分;复制后仍需显式环境变量与手动/CI 任务触发 |
|
||||
| 典型内容 | `FakeAdapter`、fixture、隔离 DB | `localhost` 仿真、真实 API(授权)、真实 RPA(最高风险) |
|
||||
|
||||
**原则**:只要会访问**网络**、**浏览器**、**真实或仿真外部系统**,应优先放在 `integration`(并保持 `.sample` 直至团队同意启用),而不是写进默认 `tests/test_*.py`。
|
||||
|
||||
## 何时使用
|
||||
|
||||
| 场景 | 建议 ``OPENCLAW_TEST_TARGET`` | 其它条件 |
|
||||
|------|------------------------------|----------|
|
||||
| 本地 / CI 对接**仿真 HTTP** | ``simulator_api`` | 仿真服务如 ``http://localhost:5180`` |
|
||||
| **仿真页面** / 录播 RPA | ``simulator_rpa`` | 本地页面或沙箱浏览器 profile |
|
||||
| **真实 API**(只读优先) | ``real_api`` | ``ALLOW_REAL_API=1``;写操作另需 ``ALLOW_WRITE_ACTIONS=1`` |
|
||||
| **真实 RPA**(最高风险) | ``real_rpa`` | ``ALLOW_REAL_RPA=1``;**仅手动触发** |
|
||||
|
||||
## 环境变量(与 ``adapter_test_utils`` 一致)
|
||||
|
||||
- **目标档位**(二选一键名,后者为历史拼写):``OPENCLAW_TEST_TARGET`` 或 ``OPENCLOW_TEST_TARGET``
|
||||
取值:``unit`` | ``mock`` | ``simulator_api`` | ``simulator_rpa`` | ``real_api`` | ``real_rpa``
|
||||
默认未设置等价于 ``unit``。
|
||||
|
||||
- **授权开关**(显式 ``1`` 才启用):
|
||||
- ``ALLOW_REAL_API=1``
|
||||
- ``ALLOW_REAL_RPA=1``
|
||||
- ``ALLOW_WRITE_ACTIONS=1``(对写操作 / 提交表单类步骤)
|
||||
|
||||
## 安全约束
|
||||
|
||||
- **禁止**在样例或复制后的测试代码中硬编码真实账号、密码、token、cookie、生产 URL。
|
||||
- 凭证应来自**平台密钥库**、受控环境变量或本机安全配置(``.env`` 已列入 ``tests/.gitignore``,勿提交)。
|
||||
- 真实 RPA 可能对 ERP/TMS/WMS 等产生真实操作;**务必**沙箱账号、只读权限,并由人工触发 CI job。
|
||||
|
||||
## 如何运行某个样例
|
||||
|
||||
1. 将目标 ``*.sample`` 复制为 ``test_*.py``(去掉 ``.sample``)。
|
||||
2. 按技能修改其中的占位 URL / 路由 / 断言。
|
||||
3. 使用 **pytest** 指向该文件(或整目录),并导出上表所需环境变量。
|
||||
|
||||
```powershell
|
||||
$env:OPENCLAW_TEST_TARGET = "simulator_api"
|
||||
pytest tests/integration/test_simulator_api_contract.py -v
|
||||
```
|
||||
24
tests/integration/test_real_api_contract.py.sample
Normal file
24
tests/integration/test_real_api_contract.py.sample
Normal file
@@ -0,0 +1,24 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
# 样例:复制为 test_real_api_contract.py;需 OPENCLAW_TEST_TARGET=real_api 且 ALLOW_REAL_API=1。
|
||||
"""真实 HTTP API(只读优先;默认跳过)。"""
|
||||
from __future__ import annotations
|
||||
|
||||
import unittest
|
||||
|
||||
from adapter_test_utils import should_skip_profile
|
||||
|
||||
|
||||
class TestRealApiContractSample(unittest.TestCase):
|
||||
def test_skip_by_default(self) -> None:
|
||||
skip, reason = should_skip_profile("real_api", write=False)
|
||||
if skip:
|
||||
raise unittest.SkipTest(reason)
|
||||
# 写操作示例(复制后使用):
|
||||
# skip_w, _ = should_skip_profile("real_api", write=True)
|
||||
# 需要 ALLOW_WRITE_ACTIONS=1
|
||||
# 禁止在此文件写入真实 token;从 vault / 环境注入读取 endpoint。
|
||||
self.assertFalse(skip)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
26
tests/integration/test_real_rpa_contract.py.sample
Normal file
26
tests/integration/test_real_rpa_contract.py.sample
Normal file
@@ -0,0 +1,26 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
# 样例:复制为 test_real_rpa_contract.py;需 OPENCLAW_TEST_TARGET=real_rpa 且 ALLOW_REAL_RPA=1。
|
||||
"""
|
||||
真实 RPA / 桌面自动化(**最高风险**,仅手动触发)。
|
||||
|
||||
默认只读思路示例:打开沙箱门户、读取标题、**不提交表单**。
|
||||
任何写操作必须 ``should_skip_profile('real_rpa', write=True)`` 且 ``ALLOW_WRITE_ACTIONS=1``。
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import unittest
|
||||
|
||||
from adapter_test_utils import should_skip_profile
|
||||
|
||||
|
||||
class TestRealRpaContractSample(unittest.TestCase):
|
||||
def test_skip_by_default(self) -> None:
|
||||
skip, reason = should_skip_profile("real_rpa", write=False)
|
||||
if skip:
|
||||
raise unittest.SkipTest(reason)
|
||||
# 真实技能中:连接企业沙箱门户,执行只读断言;禁止默认操作生产系统。
|
||||
self.assertFalse(skip)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
24
tests/integration/test_simulator_api_contract.py.sample
Normal file
24
tests/integration/test_simulator_api_contract.py.sample
Normal file
@@ -0,0 +1,24 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
# 样例:复制为 test_simulator_api_contract.py 后,在 OPENCLAW_TEST_TARGET=simulator_api 下用 pytest 或 unittest 运行。
|
||||
"""仿真 HTTP API 契约(默认跳过;不引入 requests 运行时依赖)。"""
|
||||
from __future__ import annotations
|
||||
|
||||
import unittest
|
||||
|
||||
from adapter_test_utils import should_skip_profile
|
||||
|
||||
|
||||
class TestSimulatorApiContractSample(unittest.TestCase):
|
||||
"""演示:仅当 ``OPENCLAW_TEST_TARGET=simulator_api`` 时不跳过。"""
|
||||
|
||||
def test_skip_unless_simulator_api_target(self) -> None:
|
||||
skip, reason = should_skip_profile("simulator_api")
|
||||
if skip:
|
||||
raise unittest.SkipTest(reason)
|
||||
# 真实技能中:在此使用 urllib 访问 localhost 仿真,或通过 adapter factory 注入 stub。
|
||||
# 禁止默认连接生产 ERP/TMS。
|
||||
self.assertTrue(True)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
21
tests/integration/test_simulator_rpa_contract.py.sample
Normal file
21
tests/integration/test_simulator_rpa_contract.py.sample
Normal file
@@ -0,0 +1,21 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
# 样例:复制为 test_simulator_rpa_contract.py 后按需运行;默认跳过。
|
||||
"""仿真页面 / RPA 契约(不默认启动浏览器)。"""
|
||||
from __future__ import annotations
|
||||
|
||||
import unittest
|
||||
|
||||
from adapter_test_utils import should_skip_profile
|
||||
|
||||
|
||||
class TestSimulatorRpaContractSample(unittest.TestCase):
|
||||
def test_skip_unless_simulator_rpa_target(self) -> None:
|
||||
skip, reason = should_skip_profile("simulator_rpa")
|
||||
if skip:
|
||||
raise unittest.SkipTest(reason)
|
||||
# 真实技能中:可替换为 Playwright 连接 http://localhost:5180/select 等仿真入口。
|
||||
self.assertFalse(skip)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
62
tests/run_tests.py
Normal file
62
tests/run_tests.py
Normal file
@@ -0,0 +1,62 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""后端 / CLI / DB 单元测试入口(unittest,不强制 pytest)。
|
||||
|
||||
用法:
|
||||
python tests/run_tests.py # 发现 tests/ 根目录下全部 test_*.py
|
||||
python tests/run_tests.py -v # 详细输出
|
||||
python tests/run_tests.py cli_smoke # 仅匹配 test_cli_smoke.py
|
||||
python tests/run_tests.py test_cli_smoke
|
||||
|
||||
说明:
|
||||
只发现本目录(tests/)下一层的 test_*.py;不递归 ``samples/``、``integration/``、
|
||||
``desktop/``(后三者提供 ``*.sample`` 或 pytest 专用用例,见 ``tests/README.md``)。
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import sys
|
||||
import unittest
|
||||
|
||||
_TESTS_DIR = os.path.dirname(os.path.abspath(__file__))
|
||||
_SKILL_ROOT = os.path.abspath(os.path.join(_TESTS_DIR, ".."))
|
||||
_SCRIPTS_DIR = os.path.abspath(os.path.join(_SKILL_ROOT, "scripts"))
|
||||
|
||||
for _p in (_SCRIPTS_DIR, _TESTS_DIR):
|
||||
if _p not in sys.path:
|
||||
sys.path.insert(0, _p)
|
||||
|
||||
# Windows 下统一 stdout/stderr 为 UTF-8,避免 GBK 控制台中文/emoji 报错
|
||||
if sys.platform == "win32":
|
||||
import io
|
||||
|
||||
sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding="utf-8", errors="replace")
|
||||
sys.stderr = io.TextIOWrapper(sys.stderr.buffer, encoding="utf-8", errors="replace")
|
||||
|
||||
|
||||
def _pattern_from_argv(argv: list[str]) -> str:
|
||||
for a in argv[1:]:
|
||||
if a.startswith("-"):
|
||||
continue
|
||||
s = a if a.startswith("test_") else f"test_{a}"
|
||||
if not s.endswith(".py"):
|
||||
s = f"{s}.py"
|
||||
return s
|
||||
return "test_*.py"
|
||||
|
||||
|
||||
def _verbosity_from_argv(argv: list[str]) -> int:
|
||||
return 2 if ("-v" in argv or "--verbose" in argv) else 1
|
||||
|
||||
|
||||
def main() -> int:
|
||||
pattern = _pattern_from_argv(sys.argv)
|
||||
verbosity = _verbosity_from_argv(sys.argv)
|
||||
loader = unittest.TestLoader()
|
||||
suite = loader.discover(_TESTS_DIR, pattern=pattern)
|
||||
runner = unittest.TextTestRunner(verbosity=verbosity)
|
||||
result = runner.run(suite)
|
||||
return 0 if result.wasSuccessful() else 1
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
112
tests/samples/README.md
Normal file
112
tests/samples/README.md
Normal file
@@ -0,0 +1,112 @@
|
||||
# samples(业务测试样例,默认不执行)
|
||||
|
||||
本目录存放**可复制到根目录的业务测试范式**,面向「企业数字员工技能」中 **ERP、TMS、WMS、船司、报关、邮件、LLM、RPA** 等外联场景下的 **service / golden** 测法说明。
|
||||
|
||||
---
|
||||
|
||||
## 作用
|
||||
|
||||
- 这里的文件是 **`.py.sample`**:`python tests/run_tests.py` **不会**收集本目录(`run_tests.py` 只发现 `tests/` 根目录下的 `test_*.py`)。
|
||||
- 人类或 AI 需要启用范式时:**复制**到 `tests/test_<名称>.py`(去掉 `.sample`),再按业务改 import、断言与 fixture 路径。
|
||||
- **禁止**把真实联调、真实网络、真实 RPA 写进复制后的默认用例;此类内容应放在 `tests/integration/*.sample` 并配合环境变量开关。
|
||||
|
||||
---
|
||||
|
||||
## 两类样例文件
|
||||
|
||||
### 1. `test_service_contract.py.sample`
|
||||
|
||||
**适合:**
|
||||
|
||||
- 技能有清晰的 **service 层入口**,例如 `run(...)`、`handle(...)`、`execute(...)`。
|
||||
- 需要验证:**成功路径**、**参数/校验错误**、**adapter 异常被转换为业务可读错误**。
|
||||
- 需要用 **`FakeAdapter`**(见 `tests/adapter_test_utils.py`)或自建 **stub** 替代 ERP、TMS、WMS、船司、报关、LLM、RPA 等,**不**发真实 HTTP、**不**开真实浏览器。
|
||||
|
||||
### 2. `test_golden_cases.py.sample`
|
||||
|
||||
**适合:**
|
||||
|
||||
- **解析类**:邮件/单证/PDF/Excel 等结构化输出。
|
||||
- **计算类**:费用、税费、分摊、账单字段。
|
||||
- **校验类**:合规、字段一致性、状态机规则。
|
||||
|
||||
推荐配合根目录下的 **`tests/fixtures/sample_request.json`** 与 **`tests/fixtures/expected_response.json`**(复制技能后按域替换字段,保持脱敏)。
|
||||
|
||||
---
|
||||
|
||||
## 如何启用 service contract 样例
|
||||
|
||||
1. **复制文件**
|
||||
`tests/samples/test_service_contract.py.sample`
|
||||
→ `tests/test_service_contract.py`(或 `tests/test_<domain>_service_contract.py`,须符合根目录 `test_*.py` 命名以便 `run_tests.py` 收集)。
|
||||
|
||||
2. **替换示例 import**
|
||||
把注释中的示意:
|
||||
`# from service.your_service import run`
|
||||
改为真实模块,例如:
|
||||
`from service.quote_service import run_quote`
|
||||
(具体路径以技能代码为准。)
|
||||
|
||||
3. **注入假外部依赖**
|
||||
- **不要**在默认用例里直接 `requests` 访问 ERP/TMS/WMS 等。
|
||||
- **不要**打开真实网页或驱动真实 RPA。
|
||||
- **不要**调用真实 LLM;可 stub 返回固定 JSON。
|
||||
- 对 **timeout / unauthorized / invalid_response** 等分支,用 `FakeAdapter` 或自建 fake 模拟(与 `adapter_test_utils` 思路一致)。
|
||||
|
||||
4. **至少保留三类断言习惯**
|
||||
- **成功路径**:返回 `success: true` 或与技能约定的成功结构。
|
||||
- **参数错误**:缺必填字段时返回**结构化** error(字段/码),而非未捕获裸异常。
|
||||
- **外部异常**:adapter 超时、鉴权失败、脏响应等被转换为**可读、可重试标识明确**的错误形态(按产品约定)。
|
||||
|
||||
5. **给 AI 的提示**
|
||||
复制后全文搜索 `your_service`、`TODO`、`伪代码` 等占位符并替换;不要留下指向真实客户的示例数据。
|
||||
|
||||
---
|
||||
|
||||
## 如何启用 golden case 样例
|
||||
|
||||
1. **复制文件**
|
||||
`tests/samples/test_golden_cases.py.sample`
|
||||
→ `tests/test_golden_cases.py`(或 `tests/test_<domain>_golden.py`)。
|
||||
|
||||
2. **按业务修改 fixtures**(仍须脱敏)
|
||||
- `tests/fixtures/sample_request.json`:请求或输入快照。
|
||||
- `tests/fixtures/expected_response.json`:期望摘要或关键字段。
|
||||
可新增多组文件,如 `case_valid_001.json`,并在测试中显式读取路径。
|
||||
|
||||
3. **断言建议(避免脆弱用例)**
|
||||
- 比较**关键字段**,避免对整段 JSON 做机械全文 diff。
|
||||
- **忽略**时间戳、随机 id、`trace_id` 等非确定字段。
|
||||
- 对列表:断言**长度**、**关键元素**、**排序规则**(若业务保证顺序)。
|
||||
- 对金额/数量:断言 Decimal 字符串、或允许文档约定的小范围误差。
|
||||
|
||||
4. **与 service 的关系**
|
||||
golden 用例仍应通过 **service 层**(或纯函数)得到输出,再与 fixture 对比,而不是从 CLI 字符串 scrape。
|
||||
|
||||
---
|
||||
|
||||
## 不应该写在 samples 复制件里的内容
|
||||
|
||||
以下若出现,应移到 **`tests/integration/*.sample`** 或 **`tests/desktop/*.sample`**,并加环境变量授权说明:
|
||||
|
||||
- 访问**真实** API 或内网服务。
|
||||
- 打开**真实** RPA 页面或生产门户。
|
||||
- 读取**真实**客户文件、生产数据库。
|
||||
- 在测试中写入**真实**数据根或共享盘。
|
||||
- 任何真实 **token / cookie / 密码** 字面量。
|
||||
- 耗时极长、强依赖宿主 UI 抖动、无法在 CI 稳定复现的用例(不要塞进默认 `tests/test_*.py`)。
|
||||
|
||||
---
|
||||
|
||||
## 推荐命名(复制到根目录后)
|
||||
|
||||
| 用途 | 示例文件名 |
|
||||
|------|------------|
|
||||
| Service 契约 | `tests/test_service_contract.py` |
|
||||
| Golden / fixture 回归 | `tests/test_golden_cases.py` |
|
||||
| 领域规则 | `tests/test_<domain>_rules.py` |
|
||||
| 校验逻辑 | `tests/test_<domain>_validation.py` |
|
||||
|
||||
命名需满足:`tests/` 根目录下 `test_*.py`,以便 `python tests/run_tests.py` 默认发现。
|
||||
|
||||
更多分层说明见 **`tests/README.md`**(「我该把测试写在哪里?」与「新技能最小测试清单」)。
|
||||
37
tests/samples/test_golden_cases.py.sample
Normal file
37
tests/samples/test_golden_cases.py.sample
Normal file
@@ -0,0 +1,37 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
# 样例:复制为 ``tests/test_golden_cases.py`` 后运行;默认不收集。
|
||||
"""
|
||||
Golden fixture 回归(示意)。
|
||||
|
||||
- 从 ``tests/fixtures/`` 读取脱敏 JSON。
|
||||
- 调用 service 层(复制后替换为真实 import)。
|
||||
- 断言关键字段,而非全文 diff(外部系统常带时间戳/非确定性字段)。
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import os
|
||||
import unittest
|
||||
|
||||
|
||||
def _fixtures_dir() -> str:
|
||||
return os.path.join(os.path.dirname(__file__), "..", "fixtures")
|
||||
|
||||
|
||||
class TestGoldenCasesSample(unittest.TestCase):
|
||||
def test_sample_request_matches_expected_shape(self) -> None:
|
||||
req_path = os.path.join(_fixtures_dir(), "sample_request.json")
|
||||
exp_path = os.path.join(_fixtures_dir(), "expected_response.json")
|
||||
with open(req_path, encoding="utf-8") as f:
|
||||
req = json.load(f)
|
||||
with open(exp_path, encoding="utf-8") as f:
|
||||
exp = json.load(f)
|
||||
self.assertEqual(req["request_id"], "req_test_001")
|
||||
self.assertTrue(req["parameters"]["options"]["dry_run"])
|
||||
# 示意:out = run(req); 然后 assert out["success"] == exp["success"] 等关键字段
|
||||
self.assertTrue(exp["success"])
|
||||
self.assertEqual(exp["data"]["status"], "ok")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
46
tests/samples/test_service_contract.py.sample
Normal file
46
tests/samples/test_service_contract.py.sample
Normal file
@@ -0,0 +1,46 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
# 本文件为样例:复制到 ``tests/test_service_contract.py``(或业务仓 tests 根)后按实际 service 改名运行。
|
||||
# 默认 **不会** 被 ``python tests/run_tests.py`` 收集(扩展名为 .sample)。
|
||||
"""
|
||||
Service 层契约测试范式(示意)。
|
||||
|
||||
复制为真实技能后:
|
||||
1. 将 ``from service.your_service import run`` 替换为实际模块与入口函数。
|
||||
2. 用 ``FakeAdapter`` / 内存 stub 替代真实 ERP/TMS/WMS 等外呼。
|
||||
3. 需要真实外网或真实 RPA 时,勿在本文件直接写死凭证;改用 ``tests/integration/*.sample`` + 环境开关。
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import unittest
|
||||
|
||||
from adapter_test_utils import FakeAdapter
|
||||
|
||||
|
||||
# from service.your_service import run # noqa: ERA001 — 复制后取消注释并实现
|
||||
|
||||
|
||||
class TestServiceContractSample(unittest.TestCase):
|
||||
"""三类常见契约:成功路径、结构化错误、上游超时转可重试错误。"""
|
||||
|
||||
def test_success_path_returns_success_true(self) -> None:
|
||||
# 示意:run(payload, adapter=FakeAdapter("success")) -> {"success": True, ...}
|
||||
ad = FakeAdapter("success")
|
||||
out = ad.call({"op": "ping"})
|
||||
self.assertTrue(out.get("success"))
|
||||
|
||||
def test_missing_required_field_returns_structured_error_not_bare_exception(self) -> None:
|
||||
# 示意:业务层应对必填字段做校验,返回 {"success": False, "error": {"code": "...", "message": "..."}}
|
||||
err = {"success": False, "error": {"code": "VALIDATION", "message": "missing shipment_id"}}
|
||||
self.assertFalse(err["success"])
|
||||
self.assertIn("code", err["error"])
|
||||
|
||||
def test_adapter_timeout_maps_to_retriable_error(self) -> None:
|
||||
# 示意:捕获 TimeoutError 后映射为业务可重试错误码,而不是让裸异常穿透 CLI
|
||||
ad = FakeAdapter("timeout")
|
||||
with self.assertRaises(TimeoutError):
|
||||
ad.call({})
|
||||
# 真实实现中:except TimeoutError: return {"success": False, "error": {"code": "UPSTREAM_TIMEOUT", "retriable": True}}
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
159
tests/test_adapter_profile_policy.py
Normal file
159
tests/test_adapter_profile_policy.py
Normal file
@@ -0,0 +1,159 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""adapter profile / 环境开关策略:默认不触碰真实 API 与真实 RPA。"""
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import unittest
|
||||
|
||||
from adapter_test_utils import (
|
||||
ALLOWED_TEST_TARGETS,
|
||||
FakeAdapter,
|
||||
assert_no_real_network_env,
|
||||
get_test_target,
|
||||
should_skip_profile,
|
||||
)
|
||||
|
||||
|
||||
def _save_env(keys: tuple[str, ...]) -> dict[str, str | None]:
|
||||
return {k: os.environ.get(k) for k in keys}
|
||||
|
||||
|
||||
def _restore_env(saved: dict[str, str | None]) -> None:
|
||||
for k, v in saved.items():
|
||||
if v is None:
|
||||
os.environ.pop(k, None)
|
||||
else:
|
||||
os.environ[k] = v
|
||||
|
||||
|
||||
_ENV_KEYS = (
|
||||
"OPENCLAW_TEST_TARGET",
|
||||
"OPENCLOW_TEST_TARGET",
|
||||
"ALLOW_REAL_API",
|
||||
"ALLOW_REAL_RPA",
|
||||
"ALLOW_WRITE_ACTIONS",
|
||||
)
|
||||
|
||||
|
||||
class TestAdapterProfilePolicy(unittest.TestCase):
|
||||
def test_get_test_target_reads_opencylow_typo_alias(self) -> None:
|
||||
"""兼容环境变量历史拼写 OPENCLOW_TEST_TARGET。"""
|
||||
saved = _save_env(_ENV_KEYS)
|
||||
try:
|
||||
os.environ.pop("OPENCLAW_TEST_TARGET", None)
|
||||
os.environ.pop("OPENCLOW_TEST_TARGET", None)
|
||||
os.environ["OPENCLOW_TEST_TARGET"] = "mock"
|
||||
self.assertEqual(get_test_target(), "mock")
|
||||
finally:
|
||||
_restore_env(saved)
|
||||
|
||||
def test_get_test_target_defaults_to_unit(self) -> None:
|
||||
saved = _save_env(_ENV_KEYS)
|
||||
try:
|
||||
for k in _ENV_KEYS:
|
||||
os.environ.pop(k, None)
|
||||
self.assertEqual(get_test_target(), "unit")
|
||||
finally:
|
||||
_restore_env(saved)
|
||||
|
||||
def test_get_test_target_invalid_raises(self) -> None:
|
||||
saved = _save_env(_ENV_KEYS)
|
||||
try:
|
||||
for k in _ENV_KEYS:
|
||||
os.environ.pop(k, None)
|
||||
os.environ["OPENCLAW_TEST_TARGET"] = "not_a_valid_target"
|
||||
with self.assertRaises(ValueError):
|
||||
get_test_target()
|
||||
finally:
|
||||
_restore_env(saved)
|
||||
|
||||
def test_allowed_targets_contains_expected(self) -> None:
|
||||
self.assertTrue({"unit", "mock", "simulator_api", "real_api"}.issubset(ALLOWED_TEST_TARGETS))
|
||||
|
||||
def test_simulator_api_runs_when_target_matches(self) -> None:
|
||||
saved = _save_env(_ENV_KEYS)
|
||||
try:
|
||||
os.environ["OPENCLAW_TEST_TARGET"] = "simulator_api"
|
||||
skip, _ = should_skip_profile("simulator_api")
|
||||
self.assertFalse(skip)
|
||||
finally:
|
||||
_restore_env(saved)
|
||||
|
||||
def test_unit_target_skips_simulator_and_real_profiles(self) -> None:
|
||||
saved = _save_env(_ENV_KEYS)
|
||||
try:
|
||||
for k in _ENV_KEYS:
|
||||
os.environ.pop(k, None)
|
||||
self.assertEqual(get_test_target(), "unit")
|
||||
for prof in ("simulator_api", "simulator_rpa", "real_api", "real_rpa"):
|
||||
skip, reason = should_skip_profile(prof)
|
||||
self.assertTrue(skip, msg=f"{prof}: {reason}")
|
||||
skip_mock, _ = should_skip_profile("mock")
|
||||
self.assertFalse(skip_mock)
|
||||
finally:
|
||||
_restore_env(saved)
|
||||
|
||||
def test_real_api_requires_allow_flag(self) -> None:
|
||||
saved = _save_env(_ENV_KEYS)
|
||||
try:
|
||||
os.environ["OPENCLAW_TEST_TARGET"] = "real_api"
|
||||
os.environ.pop("ALLOW_REAL_API", None)
|
||||
skip, reason = should_skip_profile("real_api")
|
||||
self.assertTrue(skip)
|
||||
self.assertIn("ALLOW_REAL_API", reason)
|
||||
os.environ["ALLOW_REAL_API"] = "1"
|
||||
skip2, _ = should_skip_profile("real_api", write=False)
|
||||
self.assertFalse(skip2)
|
||||
finally:
|
||||
_restore_env(saved)
|
||||
|
||||
def test_real_api_write_requires_allow_write_actions(self) -> None:
|
||||
saved = _save_env(_ENV_KEYS)
|
||||
try:
|
||||
os.environ["OPENCLAW_TEST_TARGET"] = "real_api"
|
||||
os.environ["ALLOW_REAL_API"] = "1"
|
||||
os.environ.pop("ALLOW_WRITE_ACTIONS", None)
|
||||
skip, reason = should_skip_profile("real_api", write=True)
|
||||
self.assertTrue(skip)
|
||||
self.assertIn("ALLOW_WRITE_ACTIONS", reason)
|
||||
os.environ["ALLOW_WRITE_ACTIONS"] = "1"
|
||||
skip2, _ = should_skip_profile("real_api", write=True)
|
||||
self.assertFalse(skip2)
|
||||
finally:
|
||||
_restore_env(saved)
|
||||
|
||||
def test_real_rpa_requires_allow_rpa_flag(self) -> None:
|
||||
saved = _save_env(_ENV_KEYS)
|
||||
try:
|
||||
os.environ["OPENCLAW_TEST_TARGET"] = "real_rpa"
|
||||
os.environ.pop("ALLOW_REAL_RPA", None)
|
||||
skip, reason = should_skip_profile("real_rpa")
|
||||
self.assertTrue(skip)
|
||||
self.assertIn("ALLOW_REAL_RPA", reason)
|
||||
os.environ["ALLOW_REAL_RPA"] = "1"
|
||||
skip2, _ = should_skip_profile("real_rpa", write=False)
|
||||
self.assertFalse(skip2)
|
||||
finally:
|
||||
_restore_env(saved)
|
||||
|
||||
def test_fake_adapter_modes(self) -> None:
|
||||
self.assertEqual(FakeAdapter("success").call({"a": 1})["success"], True)
|
||||
with self.assertRaises(TimeoutError):
|
||||
FakeAdapter("timeout").call({})
|
||||
bad = FakeAdapter("invalid_response").call({})
|
||||
self.assertNotIsInstance(bad, dict)
|
||||
with self.assertRaises(PermissionError):
|
||||
FakeAdapter("unauthorized").call({})
|
||||
|
||||
def test_assert_no_real_network_env(self) -> None:
|
||||
saved = _save_env(("ALLOW_REAL_API", "ALLOW_REAL_RPA"))
|
||||
try:
|
||||
os.environ.pop("ALLOW_REAL_API", None)
|
||||
os.environ.pop("ALLOW_REAL_RPA", None)
|
||||
assert_no_real_network_env()
|
||||
finally:
|
||||
_restore_env(saved)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
75
tests/test_cli_smoke.py
Normal file
75
tests/test_cli_smoke.py
Normal file
@@ -0,0 +1,75 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""通用 CLI 冒烟:用法、health、version、logs、log-get(不触网、不深测 publish 占位)。"""
|
||||
from __future__ import annotations
|
||||
|
||||
import io
|
||||
import json
|
||||
import os
|
||||
import unittest
|
||||
from contextlib import redirect_stderr, redirect_stdout
|
||||
|
||||
from _support import IsolatedDataRoot
|
||||
|
||||
# scripts/ 已由 _support 注入 sys.path
|
||||
from cli.app import main
|
||||
from util.constants import SKILL_SLUG
|
||||
|
||||
|
||||
class TestCliSmoke(unittest.TestCase):
|
||||
def test_main_empty_argv_shows_usage_and_nonzero(self) -> None:
|
||||
buf = io.StringIO()
|
||||
err = io.StringIO()
|
||||
with redirect_stdout(buf), redirect_stderr(err):
|
||||
rc = main([])
|
||||
self.assertEqual(rc, 1)
|
||||
out = buf.getvalue()
|
||||
self.assertIn("模板技能", out)
|
||||
self.assertIn("health", out)
|
||||
|
||||
def test_health_zero(self) -> None:
|
||||
buf = io.StringIO()
|
||||
with redirect_stdout(buf), redirect_stderr(io.StringIO()):
|
||||
rc = main(["health"])
|
||||
self.assertEqual(rc, 0)
|
||||
|
||||
def test_version_json_and_matches_constants_slug(self) -> None:
|
||||
buf = io.StringIO()
|
||||
with redirect_stdout(buf), redirect_stderr(io.StringIO()):
|
||||
rc = main(["version"])
|
||||
self.assertEqual(rc, 0)
|
||||
payload = json.loads(buf.getvalue().strip())
|
||||
self.assertIn("version", payload)
|
||||
self.assertIn("skill", payload)
|
||||
self.assertEqual(payload["skill"], SKILL_SLUG)
|
||||
|
||||
def test_logs_empty_returns_zero(self) -> None:
|
||||
with IsolatedDataRoot():
|
||||
buf = io.StringIO()
|
||||
with redirect_stdout(buf), redirect_stderr(io.StringIO()):
|
||||
rc = main(["logs"])
|
||||
self.assertEqual(rc, 0)
|
||||
self.assertIn("暂无", buf.getvalue())
|
||||
|
||||
def test_log_get_non_numeric_returns_nonzero(self) -> None:
|
||||
buf = io.StringIO()
|
||||
with redirect_stdout(buf), redirect_stderr(io.StringIO()):
|
||||
rc = main(["log-get", "not-a-number"])
|
||||
self.assertNotEqual(rc, 0)
|
||||
self.assertIn("数字", buf.getvalue())
|
||||
|
||||
def test_publish_placeholder_returns_nonzero_without_network(self) -> None:
|
||||
"""占位 publish:不验证业务成功,仅确认模板提示且退出码非 0(无 AUTH_BASE 时不发 HTTP)。"""
|
||||
old_auth = os.environ.pop("JIANGCHANG_AUTH_BASE_URL", None)
|
||||
try:
|
||||
buf = io.StringIO()
|
||||
with redirect_stdout(buf), redirect_stderr(io.StringIO()):
|
||||
rc = main(["publish"])
|
||||
self.assertNotEqual(rc, 0)
|
||||
self.assertIn("模板", buf.getvalue())
|
||||
finally:
|
||||
if old_auth is not None:
|
||||
os.environ["JIANGCHANG_AUTH_BASE_URL"] = old_auth
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
41
tests/test_db_smoke.py
Normal file
41
tests/test_db_smoke.py
Normal file
@@ -0,0 +1,41 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""publish_logs 表:init 幂等、写入与查询(全程隔离数据目录)。"""
|
||||
from __future__ import annotations
|
||||
|
||||
import unittest
|
||||
|
||||
from _support import IsolatedDataRoot
|
||||
|
||||
|
||||
class TestDbSmoke(unittest.TestCase):
|
||||
def test_init_db_idempotent_and_crud_smoke(self) -> None:
|
||||
with IsolatedDataRoot():
|
||||
from db.connection import init_db
|
||||
from db import publish_logs_repository as plr
|
||||
|
||||
init_db()
|
||||
init_db()
|
||||
|
||||
new_id = plr.save_publish_log(
|
||||
account_id="acc-1",
|
||||
article_id=42,
|
||||
article_title="t",
|
||||
status="ok",
|
||||
error_msg=None,
|
||||
)
|
||||
self.assertIsInstance(new_id, int)
|
||||
self.assertGreater(new_id, 0)
|
||||
|
||||
rows = plr.list_publish_logs(limit=10)
|
||||
self.assertTrue(any(r[0] == new_id for r in rows))
|
||||
|
||||
row = plr.get_publish_log_by_id(new_id)
|
||||
self.assertIsNotNone(row)
|
||||
self.assertEqual(int(row[0]), new_id)
|
||||
|
||||
missing = plr.get_publish_log_by_id(999_999)
|
||||
self.assertIsNone(missing)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
51
tests/test_entrypoint_subprocess.py
Normal file
51
tests/test_entrypoint_subprocess.py
Normal file
@@ -0,0 +1,51 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""通过 subprocess 调用 ``python scripts/main.py``,验证真实入口(非仅 import main)。"""
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import os
|
||||
import subprocess
|
||||
import sys
|
||||
import unittest
|
||||
|
||||
from _support import IsolatedDataRoot, get_skill_root
|
||||
|
||||
|
||||
def _run_main(args: list[str], timeout: int = 15) -> subprocess.CompletedProcess[str]:
|
||||
root = get_skill_root()
|
||||
main_py = os.path.join(root, "scripts", "main.py")
|
||||
env = os.environ.copy()
|
||||
env.setdefault("PYTHONIOENCODING", "utf-8")
|
||||
env["PYTHONDONTWRITEBYTECODE"] = "1"
|
||||
return subprocess.run(
|
||||
[sys.executable, main_py, *args],
|
||||
cwd=root,
|
||||
env=env,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
encoding="utf-8",
|
||||
errors="replace",
|
||||
timeout=timeout,
|
||||
)
|
||||
|
||||
|
||||
class TestEntrypointSubprocess(unittest.TestCase):
|
||||
def test_health_exit_zero(self) -> None:
|
||||
with IsolatedDataRoot():
|
||||
cp = _run_main(["health"])
|
||||
self.assertEqual(cp.returncode, 0, msg=cp.stderr + cp.stdout)
|
||||
|
||||
def test_version_json_and_skill_slug(self) -> None:
|
||||
with IsolatedDataRoot():
|
||||
cp = _run_main(["version"])
|
||||
self.assertEqual(cp.returncode, 0, msg=cp.stderr + cp.stdout)
|
||||
payload = json.loads(cp.stdout.strip())
|
||||
self.assertIn("version", payload)
|
||||
self.assertIn("skill", payload)
|
||||
from util.constants import SKILL_SLUG
|
||||
|
||||
self.assertEqual(payload["skill"], SKILL_SLUG)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
65
tests/test_runtime_paths.py
Normal file
65
tests/test_runtime_paths.py
Normal file
@@ -0,0 +1,65 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""util.runtime_paths:技能根、DB 路径、数据目录均落在隔离数据根下。"""
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import unittest
|
||||
|
||||
from _support import IsolatedDataRoot, get_skill_root
|
||||
|
||||
_FAKE_CLAW_ROOT = r"D:\jiangchang-data"
|
||||
|
||||
|
||||
class TestRuntimePaths(unittest.TestCase):
|
||||
def test_paths_under_isolated_root(self) -> None:
|
||||
win_default = r"D:\jiangchang-data"
|
||||
with IsolatedDataRoot() as tmp:
|
||||
import importlib
|
||||
|
||||
import util.runtime_paths as rp
|
||||
|
||||
importlib.reload(rp)
|
||||
|
||||
skill_root = rp.get_skill_root()
|
||||
self.assertTrue(os.path.isdir(skill_root))
|
||||
self.assertTrue(os.path.isfile(os.path.join(skill_root, "SKILL.md")))
|
||||
|
||||
db_path = rp.get_db_path()
|
||||
self.assertTrue(os.path.normpath(db_path).startswith(os.path.normpath(tmp)))
|
||||
self.assertNotIn(win_default, db_path)
|
||||
|
||||
data_dir = rp.get_skill_data_dir()
|
||||
self.assertTrue(os.path.normpath(data_dir).startswith(os.path.normpath(tmp)))
|
||||
self.assertIn("_test", data_dir.replace("\\", "/"))
|
||||
self.assertNotIn(win_default, data_dir)
|
||||
|
||||
self.assertEqual(get_skill_root(), skill_root)
|
||||
|
||||
def test_isolation_overrides_prefixed_claw_data_root(self) -> None:
|
||||
"""先设假的 CLAW_DATA_ROOT,再进入隔离;get_db_path 必须落在 tempdir(防漏设 CLAW_*)。"""
|
||||
saved = {k: os.environ.get(k) for k in ("CLAW_DATA_ROOT", "JIANGCHANG_DATA_ROOT", "CLAW_USER_ID", "JIANGCHANG_USER_ID")}
|
||||
try:
|
||||
os.environ["CLAW_DATA_ROOT"] = _FAKE_CLAW_ROOT
|
||||
os.environ["JIANGCHANG_DATA_ROOT"] = _FAKE_CLAW_ROOT
|
||||
with IsolatedDataRoot() as tmp:
|
||||
import importlib
|
||||
|
||||
import util.runtime_paths as rp
|
||||
|
||||
importlib.reload(rp)
|
||||
db_path = rp.get_db_path()
|
||||
norm_db = os.path.normcase(os.path.normpath(db_path))
|
||||
norm_tmp = os.path.normcase(os.path.normpath(tmp))
|
||||
self.assertTrue(norm_db.startswith(norm_tmp), msg=f"db_path={db_path!r} tmp={tmp!r}")
|
||||
norm_fake = os.path.normcase(os.path.normpath(_FAKE_CLAW_ROOT))
|
||||
self.assertFalse(norm_db.startswith(norm_fake), msg=f"db_path must not be under fake CLAW root: {db_path!r}")
|
||||
finally:
|
||||
for k, v in saved.items():
|
||||
if v is None:
|
||||
os.environ.pop(k, None)
|
||||
else:
|
||||
os.environ[k] = v
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
55
tests/test_skill_metadata.py
Normal file
55
tests/test_skill_metadata.py
Normal file
@@ -0,0 +1,55 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""SKILL.md 与 util.constants.SKILL_SLUG 一致性(防复制后漏改 constants)。"""
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import re
|
||||
import unittest
|
||||
|
||||
from _support import get_skill_root
|
||||
|
||||
|
||||
def _parse_openclaw_slug(skill_md_text: str) -> str:
|
||||
"""从 YAML frontmatter 中解析 metadata.openclaw.slug(无 PyYAML 依赖)。"""
|
||||
parts = skill_md_text.split("---", 2)
|
||||
if len(parts) < 2:
|
||||
raise ValueError("missing frontmatter")
|
||||
fm = parts[1]
|
||||
lines = fm.splitlines()
|
||||
for i, line in enumerate(lines):
|
||||
if re.match(r"^\s+openclaw:\s*$", line):
|
||||
for j in range(i + 1, len(lines)):
|
||||
inner = lines[j]
|
||||
if inner.strip() and inner[0] not in (" ", "\t"):
|
||||
break
|
||||
m = re.match(r"^(\s+)slug:\s*(.+)\s*$", inner)
|
||||
if m:
|
||||
return m.group(2).strip().strip("\"'")
|
||||
break
|
||||
raise ValueError("slug not found under openclaw")
|
||||
|
||||
|
||||
class TestSkillMetadata(unittest.TestCase):
|
||||
def test_skill_slug_alignment(self) -> None:
|
||||
root = get_skill_root()
|
||||
skill_dir = os.path.basename(root)
|
||||
md_path = os.path.join(root, "SKILL.md")
|
||||
with open(md_path, encoding="utf-8") as f:
|
||||
md = f.read()
|
||||
slug_md = _parse_openclaw_slug(md)
|
||||
|
||||
from util.constants import SKILL_SLUG
|
||||
|
||||
self.assertEqual(slug_md, SKILL_SLUG, "SKILL.md slug 必须与 constants.SKILL_SLUG 一致")
|
||||
|
||||
if skill_dir == "skill-template":
|
||||
# 模板仓库本体:允许占位 slug
|
||||
self.assertEqual(SKILL_SLUG, "your-skill-slug")
|
||||
else:
|
||||
self.assertNotEqual(SKILL_SLUG, "your-skill-slug", "复制为真实技能后不得保留占位 slug")
|
||||
self.assertEqual(SKILL_SLUG, skill_dir)
|
||||
self.assertEqual(slug_md, skill_dir)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user