From 4fe1351e1e50d0ca157f32be6c66e78ca40f7980 Mon Sep 17 00:00:00 2001 From: chendelian <116870791@qq.com> Date: Mon, 11 May 2026 18:51:56 +0800 Subject: [PATCH] test(compat): add v1 compatibility suite; docs: add V1_COMPATIBILITY.md Co-authored-by: Cursor --- references/CLI.md | 7 +- references/V1_COMPATIBILITY.md | 43 +++++ tests/test_usage.py | 33 ++++ tests/test_v1_compatibility.py | 332 +++++++++++++++++++++++++++++++++ 4 files changed, 414 insertions(+), 1 deletion(-) create mode 100644 references/V1_COMPATIBILITY.md create mode 100644 tests/test_usage.py create mode 100644 tests/test_v1_compatibility.py diff --git a/references/CLI.md b/references/CLI.md index a4a92b6..8d2a2c9 100644 --- a/references/CLI.md +++ b/references/CLI.md @@ -2,7 +2,8 @@ 将 `{baseDir}` 替换为技能根目录。所有命令通过 `python {baseDir}/scripts/main.py` 调用（工作目录可为技能根或 `scripts/`，与命令中的路径一致即可）。下文示例简写为 `python main.py`，表示在 `{baseDir}/scripts/` 下执行。 -> **说明**：`scripts/cli/app.py` 内置的帮助字符串 `_USAGE` 主要覆盖 platform / account / credential / lease / open；实际路由还支持 **`account init-master-key` / `account export-credentials` / `account import-credentials`**（本文档以代码为准）。 +> **说明**：内置 `_USAGE`（无参数启动 CLI）已列出加密三类子命令；更完整的 JSON 约定仍以本文与源码为准。 + ## stdout / JSON 约定（集成必读） @@ -201,3 +202,7 @@ ACCOUNT_MANAGER_MASTER_KEY="" python main.py account init-maste - **数据与路径**：库文件、profile、`master.key` 布局见 [RUNTIME.md](RUNTIME.md)。 - **机器可读集成**：见 [INTEGRATION.md](INTEGRATION.md)。 - **错误码**：见 [ERRORS.md](ERRORS.md)。 + +## 内置 `_USAGE`（阶段 F） + +无参数运行 CLI（例如 `python scripts/main.py` / `python -m cli.app`，cwd=`scripts/`）打印的内置 `_USAGE` 已与 `main()` 路由对齐，包含 **`account init-master-key` / `account export-credentials` / `account import-credentials`**。面向外部仓库的契约摘要见 [V1_COMPATIBILITY.md](V1_COMPATIBILITY.md)。 diff --git a/references/V1_COMPATIBILITY.md b/references/V1_COMPATIBILITY.md new file mode 100644 index 0000000..f789ea8 --- /dev/null +++ b/references/V1_COMPATIBILITY.md @@ -0,0 +1,43 @@ +# v1 调用方兼容承诺（account-manager v2） + +account-manager v2 **不破坏**下列面向脚本 / 其它仓库（通过 CLI subprocess 集成）的契约。 + +## CLI 命令 + +下列命令的**子命令名与既有必填参数**保持不变（v2 仅 **扩展** 可选参数，不移除）： + +- `account pick-web` / `lease release` / `account mark-used` / `account mark-session` / `account mark-error` +- `account get` / `account list` / `account add-web` / `account add-secret` +- 兼容入口：`get`、`list-json`、`pick-web`、`add`、`delete`、`open` + +加密相关：`account init-master-key` / `account export-credentials` / `account import-credentials`（**仅新增**，不影响旧调用方）。 + +内置 `--help` / 无参数打印的 `_USAGE` 文案见仓库 `scripts/cli/app.py`（阶段 F 已与路由对齐）。 + +## JSON 字段 + +- **`account pick-web`**：成功时为 **单层 JSON 对象**（**无** `success`/`data` 外层）。保留字段包括但不限于：`id`、`platform_key`、`environment`、`role`、`profile_dir`、`lease_token`、`lease_expires_at`（带 `--lease` 时）。v2 **允许追加**字段（如 `auth_strategy`、`session_persistent`、`device_fingerprint`），**不得改名或删除**上述 v1 字段。 +- **`account get`**：仍为单行账号 JSON（字段集合见 [SCHEMA.md](SCHEMA.md)）。 +- **`lease release` / `account mark-*`**：成功时使用 `_ok_json`（`{"success": true, "data": {...}}`）；失败为统一错误 JSON。 + +## 错误码 + +下列 **`error.code` 字面值**保持不变： + +- `ERROR:NO_ACCOUNT` +- `ERROR:LEASE_CONFLICT` +- `ERROR:ACCOUNT_NOT_FOUND` +- `ERROR:INVALID_PLATFORM` +- `ERROR:ACCOUNT_DISABLED` +- `ERROR:DB_ERROR` + +v2 新增的 `ERR_*` 代码（如 `ERR_LEASE_INVALID`、`ERR_AUTH_STRATEGY_NOT_SUPPORTED`）仅出现在 **v2 新命令**路径，不应干扰上述 monitor 命令的失败解析。 + +## stderr + +v2 可能在 stderr 打印迁移摘要、`[export] WARNING` 等。**调用方应只解析 stdout**，并与 monitor 一样优先取 **最后一行非空行** 做 `json.loads`（见集成测试 `tests/test_v1_compatibility.py`）。 + +## 集成验证 + +- 仓库内 **`tests/test_v1_compatibility.py`** 以 subprocess 覆盖典型调用序列与 stdout 形态。 +- 已知调用方：**`monitor-competitor-rates`**（物流仓库）—— 仅 CLI，无 Python import。 diff --git a/tests/test_usage.py b/tests/test_usage.py new file mode 100644 index 0000000..4503ebe --- /dev/null +++ b/tests/test_usage.py @@ -0,0 +1,33 @@ +# -*- coding: utf-8 -*- +"""Smoke test: built-in CLI usage lists routed account subcommands.""" +from __future__ import annotations + +import os +import subprocess +import sys + +import pytest + +_scripts = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "scripts") + + +def test_usage_includes_all_account_subcommands(): + env = {**os.environ, "PYTHONPATH": _scripts} + r = subprocess.run( + [sys.executable, "-m", "cli.app"], + capture_output=True, + text=True, + env=env, + cwd=_scripts, + ) + assert r.returncode == 1 + usage = r.stdout + for cmd in ( + "init-master-key", + "export-credentials", + "import-credentials", + "add-web", + "get-credential", + "pick-web", + ): + assert cmd in usage, f"USAGE 缺少 {cmd}" diff --git a/tests/test_v1_compatibility.py b/tests/test_v1_compatibility.py new file mode 100644 index 0000000..4a714c4 --- /dev/null +++ b/tests/test_v1_compatibility.py @@ -0,0 +1,332 @@ +# -*- coding: utf-8 -*- +"""v1 CLI compatibility — subprocess-only, mirrors monitor-competitor-rates calling patterns.""" +from __future__ import annotations + +import json +import os +import shutil +import subprocess +import sys +import tempfile +from concurrent.futures import ThreadPoolExecutor, wait + +import pytest + +_scripts = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "scripts") +if _scripts not in sys.path: + sys.path.insert(0, _scripts) + +MAIN_PY = os.path.join(_scripts, "main.py") + + +def _cli_env(roots: dict) -> dict: + e = {**os.environ} + e["JIANGCHANG_DATA_ROOT"] = roots["data_root"] + e["JIANGCHANG_USER_ID"] = roots["user_id"] + e["CLAW_DATA_ROOT"] = roots["data_root"] + e["CLAW_USER_ID"] = roots["user_id"] + return e + + +def _run_cli(env: dict, args: list[str], stdin: str | None = None) -> subprocess.CompletedProcess: + inp = stdin.encode("utf-8") if stdin is not None else None + return subprocess.run( + [sys.executable, MAIN_PY, *args], + input=inp, + capture_output=True, + env=env, + cwd=os.path.dirname(_scripts), + ) + + +def _parse_last_json(stdout: str) -> dict: + lines = [ln for ln in (stdout or "").strip().split("\n") if ln.strip()] + assert lines, "stdout had no non-empty lines" + return json.loads(lines[-1]) + + +def run_cli_json_last(env: dict, args: list[str], *, stdin: str | None = None) -> dict: + """Parse **last** stdout line as JSON (monitor `_parse_stdout_json` semantics).""" + r = _run_cli(env, args, stdin=stdin) + assert r.returncode == 0, (r.stdout, r.stderr) + return _parse_last_json(r.stdout.decode("utf-8")) + + +@pytest.fixture() +def iso_cli_env(): + tmp = tempfile.mkdtemp(prefix="v1_compat_") + data_root = os.path.join(tmp, "claw-data") + os.makedirs(data_root) + uid = "compat_user" + yield {"data_root": data_root, "user_id": uid, "tmp_dir": tmp} + shutil.rmtree(tmp, ignore_errors=True) + + +def _add_monitor_account(env: dict) -> dict: + out = run_cli_json_last( + env, + [ + "account", + "add-web", + "--platform", + "maersk", + "--label", + "monitor-test", + "--environment", + "simulator", + "--role", + "booking", + "--tenant-id", + "tenant-test", + ], + ) + assert out.get("success") is True + return out["data"] + + +class TestV1PickWeb: + def test_pick_web_returns_v1_shape(self, iso_cli_env): + env = _cli_env(iso_cli_env) + add_data = _add_monitor_account(env) + aid = add_data["id"] + assert isinstance(aid, int) + + pick = run_cli_json_last( + env, + [ + "account", + "pick-web", + "--platform", + "maersk", + "--environment", + "simulator", + "--role", + "booking", + "--lease", + "--holder", + "monitor-competitor-rates", + "--purpose", + "maersk_sim_rpa", + "--ttl-sec", + "1800", + ], + ) + + for key in ("id", "platform_key", "environment", "role", "profile_dir", "lease_token"): + assert key in pick, f"pick-web 缺少 v1 字段 {key}" + + assert isinstance(pick["id"], int) + assert isinstance(pick["lease_token"], str) and len(pick["lease_token"]) > 0 + assert pick["platform_key"] == "maersk" + assert pick["environment"] == "simulator" + assert pick["role"] == "booking" + + for v2_key in ("auth_strategy", "session_persistent", "device_fingerprint"): + assert v2_key in pick, f"v2 字段 {v2_key} 应在 pick-web 返回中" + + assert len(pick) >= 10 + + def test_pick_web_no_account_returns_legacy_error_code(self, iso_cli_env): + env = _cli_env(iso_cli_env) + proc = _run_cli( + env, + [ + "account", + "pick-web", + "--platform", + "maersk", + "--environment", + "simulator", + "--role", + "booking", + "--lease", + "--holder", + "monitor-competitor-rates", + ], + ) + assert proc.returncode == 0 + payload = _parse_last_json(proc.stdout.decode("utf-8")) + assert payload.get("success") is False + assert payload["error"]["code"] == "ERROR:NO_ACCOUNT" + + def test_pick_web_lease_race_returns_legacy_error_shape(self, iso_cli_env): + """Parallel picks share one account — second outcome may be NO_ACCOUNT or LEASE_CONFLICT.""" + env = _cli_env(iso_cli_env) + _add_monitor_account(env) + + args = [ + "account", + "pick-web", + "--platform", + "maersk", + "--environment", + "simulator", + "--role", + "booking", + "--lease", + "--holder", + "monitor-competitor-rates", + "--ttl-sec", + "3600", + ] + + def _pick(): + return _run_cli(env, args) + + with ThreadPoolExecutor(max_workers=2) as ex: + f1 = ex.submit(_pick) + f2 = ex.submit(_pick) + wait([f1, f2]) + proc_a = f1.result() + proc_b = f2.result() + + payloads = [] + for proc in (proc_a, proc_b): + assert proc.returncode == 0 + payloads.append(_parse_last_json(proc.stdout.decode("utf-8"))) + + codes = [] + successes = 0 + for p in payloads: + if p.get("success") is False: + codes.append(p["error"]["code"]) + elif "lease_token" in p: + successes += 1 + + assert successes >= 1 + assert "ERROR:LEASE_CONFLICT" in codes or "ERROR:NO_ACCOUNT" in codes + + +class TestV1LeaseAndMarks: + def test_lease_release_returns_success_json(self, iso_cli_env): + env = _cli_env(iso_cli_env) + _add_monitor_account(env) + pick = run_cli_json_last( + env, + [ + "account", + "pick-web", + "--platform", + "maersk", + "--environment", + "simulator", + "--role", + "booking", + "--lease", + "--holder", + "monitor-competitor-rates", + "--ttl-sec", + "900", + ], + ) + tok = pick["lease_token"] + rel = run_cli_json_last(env, ["lease", "release", tok]) + assert rel.get("success") is True + assert rel["data"]["status"] == "released" + + def test_mark_used_returns_success(self, iso_cli_env): + env = _cli_env(iso_cli_env) + aid = _add_monitor_account(env)["id"] + out = run_cli_json_last(env, ["account", "mark-used", str(aid)]) + assert out.get("success") is True + assert int(out["data"]["id"]) == aid + + def test_mark_session_accepts_v1_status_values(self, iso_cli_env): + env = _cli_env(iso_cli_env) + aid = _add_monitor_account(env)["id"] + for status in ("likely_valid", "needs_login", "unknown"): + out = run_cli_json_last( + env, + ["account", "mark-session", str(aid), "--session-status", status], + ) + assert out.get("success") is True + assert out["data"]["session_status"] == status + + def test_mark_error_accepts_long_message(self, iso_cli_env): + env = _cli_env(iso_cli_env) + aid = _add_monitor_account(env)["id"] + long_msg = "x" * 600 + out = run_cli_json_last( + env, + [ + "account", + "mark-error", + str(aid), + "--code", + "ERR_TEST", + "--message", + long_msg, + ], + ) + assert out.get("success") is True + assert out["data"]["error_code"] == "ERR_TEST" + + +class TestV1EndToEnd: + def test_full_monitor_sequence_end_to_end(self, iso_cli_env): + env = _cli_env(iso_cli_env) + aid = _add_monitor_account(env)["id"] + + pick1 = run_cli_json_last( + env, + [ + "account", + "pick-web", + "--platform", + "maersk", + "--environment", + "simulator", + "--role", + "booking", + "--lease", + "--holder", + "monitor-competitor-rates", + "--purpose", + "maersk_sim_rpa", + "--ttl-sec", + "900", + ], + ) + tok = pick1["lease_token"] + assert pick1["id"] == aid + + run_cli_json_last(env, ["account", "mark-used", str(aid)]) + run_cli_json_last( + env, + ["account", "mark-session", str(aid), "--session-status", "likely_valid"], + ) + run_cli_json_last(env, ["lease", "release", tok]) + + pick2 = run_cli_json_last( + env, + [ + "account", + "pick-web", + "--platform", + "maersk", + "--environment", + "simulator", + "--role", + "booking", + "--lease", + "--holder", + "monitor-competitor-rates", + "--ttl-sec", + "900", + ], + ) + assert "lease_token" in pick2 + + acc = run_cli_json_last(env, ["account", "get", str(aid)]) + assert acc.get("session_status") == "likely_valid" + + +class TestStdoutIsolation: + def test_stderr_does_not_pollute_stdout_last_line(self, iso_cli_env): + env = _cli_env(iso_cli_env) + proc = _run_cli(env, ["account", "export-credentials", "--plaintext"]) + assert proc.returncode == 0 + assert b"[export] WARNING" in proc.stderr + lines = [ln for ln in proc.stdout.decode("utf-8").strip().split("\n") if ln.strip()] + payload = json.loads(lines[-1]) + assert isinstance(payload, list)