新增完善测试模板

This commit is contained in:
2026-05-03 17:52:52 +08:00
parent 07fa0b0038
commit dd6236866e
26 changed files with 1319 additions and 5 deletions

10
tests/.gitignore vendored Normal file
View File

@@ -0,0 +1,10 @@
# 测试运行产物与本地密钥,勿提交
__pycache__/
*.py[cod]
.pytest_cache/
desktop/artifacts/
integration/artifacts/
*.log
.env
.env.*
!.env.sample

View File

@@ -1,8 +1,119 @@
# tests # skill-template 测试说明(企业数字员工技能)
这里放模板复制后的单元测试或最小回归测试 本目录提供面向 **ERP / TMS / WMS / 船司 / 报关 / 邮件 / 浏览器 / LLM / RPA** 等外联场景的**分层测试骨架**:默认仅内存与本地 SQLite**真实 API / 真实 RPA** 必须通过环境变量显式授权,且仅以 ``*.sample`` 或复制后的文件提供范式
模板仓库本身不强制附带业务测试,但建议新 skill 至少补: 复制为新技能后,请保留隔离数据根与 profile 策略,再按域扩展用例。
- `health` / `version` 最小冒烟测试 ---
- 关键 `service` 层函数的单元测试
## 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 E2Epytest
见 **`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_*``(用于必跑用例自检)。
---
## 6. 与 content-manager 的关系
- **未**复制 article / media / prompt / NotebookLM 等业务测试。
- 本骨架面向**通用外联技能**;业务契约请放在各自技能的 ``tests/samples`` / ``tests/integration`` 复制件中实现。

77
tests/_support.py Normal file
View 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
View 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"

32
tests/desktop/README.md Normal file
View File

@@ -0,0 +1,32 @@
# 可选桌面 E2Epytest
本目录用于**匠厂桌面 + 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** 维护,二者互补、互不替代。

76
tests/desktop/conftest.py Normal file
View File

@@ -0,0 +1,76 @@
# -*- coding: utf-8 -*-
"""
可选桌面 E2Epytest 收集时自动加载,负责把 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

View 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

17
tests/fixtures/README.md vendored Normal file
View File

@@ -0,0 +1,17 @@
# 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的配置表达样例。 |

View 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
View File

@@ -0,0 +1,6 @@
{
"success": true,
"data": {
"status": "ok"
}
}

11
tests/fixtures/sample_request.json vendored Normal file
View 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
}
}
}

View File

@@ -0,0 +1,43 @@
# integration可选集成 / 真实外联)
本目录**默认不执行**
- ``python tests/run_tests.py`` 只收集 ``tests/`` 根目录下 ``test_*.py``**不会**进入本目录。
- 本目录下文件一律为 ``*.py.sample``pytest / unittest **默认也不会**按普通 ``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
```

View 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()

View 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()

View 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()

View 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
View 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())

5
tests/samples/README.md Normal file
View File

@@ -0,0 +1,5 @@
# samples样例默认不执行
本目录下的 ``*.py.sample`` **不会被** ``python tests/run_tests.py`` 收集。
复制为 ``tests/test_*.py``(去掉 ``.sample``)后,再按业务实现 service / golden 断言。

View 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()

View 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()

View 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
View 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
View 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()

View 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()

View 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()

View 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()