From dd6236866eb8d3350aaa5f51118341429357189e Mon Sep 17 00:00:00 2001 From: chendelian <116870791@qq.com> Date: Sun, 3 May 2026 17:52:52 +0800 Subject: [PATCH] =?UTF-8?q?=E6=96=B0=E5=A2=9E=E5=AE=8C=E5=96=84=E6=B5=8B?= =?UTF-8?q?=E8=AF=95=E6=A8=A1=E6=9D=BF?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- tests/.gitignore | 10 ++ tests/README.md | 121 ++++++++++++- tests/_support.py | 77 +++++++++ tests/adapter_test_utils.py | 142 ++++++++++++++++ tests/desktop/README.md | 32 ++++ tests/desktop/conftest.py | 76 +++++++++ tests/desktop/test_desktop_smoke.py.sample | 58 +++++++ tests/fixtures/README.md | 17 ++ tests/fixtures/adapter_profiles.sample.yaml | 40 +++++ tests/fixtures/expected_response.json | 6 + tests/fixtures/sample_request.json | 11 ++ tests/integration/README.md | 43 +++++ .../test_real_api_contract.py.sample | 24 +++ .../test_real_rpa_contract.py.sample | 26 +++ .../test_simulator_api_contract.py.sample | 24 +++ .../test_simulator_rpa_contract.py.sample | 21 +++ tests/run_tests.py | 62 +++++++ tests/samples/README.md | 5 + tests/samples/test_golden_cases.py.sample | 37 ++++ tests/samples/test_service_contract.py.sample | 46 +++++ tests/test_adapter_profile_policy.py | 159 ++++++++++++++++++ tests/test_cli_smoke.py | 75 +++++++++ tests/test_db_smoke.py | 41 +++++ tests/test_entrypoint_subprocess.py | 51 ++++++ tests/test_runtime_paths.py | 65 +++++++ tests/test_skill_metadata.py | 55 ++++++ 26 files changed, 1319 insertions(+), 5 deletions(-) create mode 100644 tests/.gitignore create mode 100644 tests/_support.py create mode 100644 tests/adapter_test_utils.py create mode 100644 tests/desktop/README.md create mode 100644 tests/desktop/conftest.py create mode 100644 tests/desktop/test_desktop_smoke.py.sample create mode 100644 tests/fixtures/README.md create mode 100644 tests/fixtures/adapter_profiles.sample.yaml create mode 100644 tests/fixtures/expected_response.json create mode 100644 tests/fixtures/sample_request.json create mode 100644 tests/integration/README.md create mode 100644 tests/integration/test_real_api_contract.py.sample create mode 100644 tests/integration/test_real_rpa_contract.py.sample create mode 100644 tests/integration/test_simulator_api_contract.py.sample create mode 100644 tests/integration/test_simulator_rpa_contract.py.sample create mode 100644 tests/run_tests.py create mode 100644 tests/samples/README.md create mode 100644 tests/samples/test_golden_cases.py.sample create mode 100644 tests/samples/test_service_contract.py.sample create mode 100644 tests/test_adapter_profile_policy.py create mode 100644 tests/test_cli_smoke.py create mode 100644 tests/test_db_smoke.py create mode 100644 tests/test_entrypoint_subprocess.py create mode 100644 tests/test_runtime_paths.py create mode 100644 tests/test_skill_metadata.py diff --git a/tests/.gitignore b/tests/.gitignore new file mode 100644 index 0000000..e6fce5d --- /dev/null +++ b/tests/.gitignore @@ -0,0 +1,10 @@ +# 测试运行产物与本地密钥，勿提交 +__pycache__/ +*.py[cod] +.pytest_cache/ +desktop/artifacts/ +integration/artifacts/ +*.log +.env +.env.* +!.env.sample diff --git a/tests/README.md b/tests/README.md index c45d9ab..c95dfe6 100644 --- a/tests/README.md +++ b/tests/README.md @@ -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 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_*``（用于必跑用例自检）。 + +--- + +## 6. 与 content-manager 的关系 + +- **未**复制 article / media / prompt / NotebookLM 等业务测试。 +- 本骨架面向**通用外联技能**；业务契约请放在各自技能的 ``tests/samples`` / ``tests/integration`` 复制件中实现。 diff --git a/tests/_support.py b/tests/_support.py new file mode 100644 index 0000000..62a551b --- /dev/null +++ b/tests/_support.py @@ -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 diff --git a/tests/adapter_test_utils.py b/tests/adapter_test_utils.py new file mode 100644 index 0000000..f99be16 --- /dev/null +++ b/tests/adapter_test_utils.py @@ -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" diff --git a/tests/desktop/README.md b/tests/desktop/README.md new file mode 100644 index 0000000..338187d --- /dev/null +++ b/tests/desktop/README.md @@ -0,0 +1,32 @@ +# 可选桌面 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** 维护，二者互补、互不替代。 diff --git a/tests/desktop/conftest.py b/tests/desktop/conftest.py new file mode 100644 index 0000000..2ae9e48 --- /dev/null +++ b/tests/desktop/conftest.py @@ -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 /jiangchang-platform-kit/sdk/jiangchang-desktop-sdk\n" + " 2) 设置环境变量 JIANGCHANG_KIT_ROOT=/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 diff --git a/tests/desktop/test_desktop_smoke.py.sample b/tests/desktop/test_desktop_smoke.py.sample new file mode 100644 index 0000000..095a0df --- /dev/null +++ b/tests/desktop/test_desktop_smoke.py.sample @@ -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 diff --git a/tests/fixtures/README.md b/tests/fixtures/README.md new file mode 100644 index 0000000..c78193e --- /dev/null +++ b/tests/fixtures/README.md @@ -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）的配置表达样例。 | diff --git a/tests/fixtures/adapter_profiles.sample.yaml b/tests/fixtures/adapter_profiles.sample.yaml new file mode 100644 index 0000000..9a30db2 --- /dev/null +++ b/tests/fixtures/adapter_profiles.sample.yaml @@ -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 diff --git a/tests/fixtures/expected_response.json b/tests/fixtures/expected_response.json new file mode 100644 index 0000000..d031ce4 --- /dev/null +++ b/tests/fixtures/expected_response.json @@ -0,0 +1,6 @@ +{ + "success": true, + "data": { + "status": "ok" + } +} diff --git a/tests/fixtures/sample_request.json b/tests/fixtures/sample_request.json new file mode 100644 index 0000000..96ad266 --- /dev/null +++ b/tests/fixtures/sample_request.json @@ -0,0 +1,11 @@ +{ + "request_id": "req_test_001", + "tenant_id": "tenant_demo", + "actor": "tester", + "parameters": { + "input_text": "demo input", + "options": { + "dry_run": true + } + } +} diff --git a/tests/integration/README.md b/tests/integration/README.md new file mode 100644 index 0000000..359693f --- /dev/null +++ b/tests/integration/README.md @@ -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 +``` diff --git a/tests/integration/test_real_api_contract.py.sample b/tests/integration/test_real_api_contract.py.sample new file mode 100644 index 0000000..68c1bef --- /dev/null +++ b/tests/integration/test_real_api_contract.py.sample @@ -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() diff --git a/tests/integration/test_real_rpa_contract.py.sample b/tests/integration/test_real_rpa_contract.py.sample new file mode 100644 index 0000000..f1e6d25 --- /dev/null +++ b/tests/integration/test_real_rpa_contract.py.sample @@ -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() diff --git a/tests/integration/test_simulator_api_contract.py.sample b/tests/integration/test_simulator_api_contract.py.sample new file mode 100644 index 0000000..f512e36 --- /dev/null +++ b/tests/integration/test_simulator_api_contract.py.sample @@ -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() diff --git a/tests/integration/test_simulator_rpa_contract.py.sample b/tests/integration/test_simulator_rpa_contract.py.sample new file mode 100644 index 0000000..1815483 --- /dev/null +++ b/tests/integration/test_simulator_rpa_contract.py.sample @@ -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() diff --git a/tests/run_tests.py b/tests/run_tests.py new file mode 100644 index 0000000..0bfb501 --- /dev/null +++ b/tests/run_tests.py @@ -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()) diff --git a/tests/samples/README.md b/tests/samples/README.md new file mode 100644 index 0000000..225da81 --- /dev/null +++ b/tests/samples/README.md @@ -0,0 +1,5 @@ +# samples（样例，默认不执行） + +本目录下的 ``*.py.sample`` **不会被** ``python tests/run_tests.py`` 收集。 + +复制为 ``tests/test_*.py``（去掉 ``.sample``）后，再按业务实现 service / golden 断言。 diff --git a/tests/samples/test_golden_cases.py.sample b/tests/samples/test_golden_cases.py.sample new file mode 100644 index 0000000..6a35750 --- /dev/null +++ b/tests/samples/test_golden_cases.py.sample @@ -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() diff --git a/tests/samples/test_service_contract.py.sample b/tests/samples/test_service_contract.py.sample new file mode 100644 index 0000000..258b8cf --- /dev/null +++ b/tests/samples/test_service_contract.py.sample @@ -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() diff --git a/tests/test_adapter_profile_policy.py b/tests/test_adapter_profile_policy.py new file mode 100644 index 0000000..5c92e39 --- /dev/null +++ b/tests/test_adapter_profile_policy.py @@ -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() diff --git a/tests/test_cli_smoke.py b/tests/test_cli_smoke.py new file mode 100644 index 0000000..688cfac --- /dev/null +++ b/tests/test_cli_smoke.py @@ -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() diff --git a/tests/test_db_smoke.py b/tests/test_db_smoke.py new file mode 100644 index 0000000..bcceea4 --- /dev/null +++ b/tests/test_db_smoke.py @@ -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() diff --git a/tests/test_entrypoint_subprocess.py b/tests/test_entrypoint_subprocess.py new file mode 100644 index 0000000..45c1184 --- /dev/null +++ b/tests/test_entrypoint_subprocess.py @@ -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() diff --git a/tests/test_runtime_paths.py b/tests/test_runtime_paths.py new file mode 100644 index 0000000..c2642fe --- /dev/null +++ b/tests/test_runtime_paths.py @@ -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() diff --git a/tests/test_skill_metadata.py b/tests/test_skill_metadata.py new file mode 100644 index 0000000..fa2d2ce --- /dev/null +++ b/tests/test_skill_metadata.py @@ -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()