From 65d939dc5b7a5f424b0189a7b2d8dad60224d892 Mon Sep 17 00:00:00 2001 From: chendelian <116870791@qq.com> Date: Mon, 11 May 2026 11:25:53 +0800 Subject: [PATCH] test(cli): add stage C CLI tests Co-authored-by: Cursor --- tests/test_cli_stage_c.py | 511 ++++++++++++++++++++++++++++++++++++++ 1 file changed, 511 insertions(+) create mode 100644 tests/test_cli_stage_c.py diff --git a/tests/test_cli_stage_c.py b/tests/test_cli_stage_c.py new file mode 100644 index 0000000..a9ab4e6 --- /dev/null +++ b/tests/test_cli_stage_c.py @@ -0,0 +1,511 @@ +# -*- coding: utf-8 -*- +"""Stage C: CLI — add-web v2, local_encrypted add-secret, pick-web fields, get-credential.""" +from __future__ import annotations + +import json +import os +import shutil +import sqlite3 +import subprocess +import sys +import tempfile +from dataclasses import dataclass + +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") + +from util.constants import SKILL_SLUG # noqa: E402 + + +def _cli_env(roots: dict, **extra: str) -> dict: + e = {**os.environ, **extra} + 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 _db_path(roots: dict) -> str: + return os.path.join(roots["data_root"], roots["user_id"], SKILL_SLUG, "account-manager.db") + + +@dataclass +class _ProcOut: + returncode: int + stdout: str + stderr: str + + +def _run_cli(env: dict, args: list[str], stdin: str | None = None) -> _ProcOut: + inp = stdin.encode("utf-8") if stdin is not None else None + r = subprocess.run( + [sys.executable, MAIN_PY, *args], + input=inp, + capture_output=True, + env=env, + cwd=os.path.dirname(_scripts), + ) + stdout_txt = r.stdout.decode("utf-8") if r.stdout else "" + stderr_txt = r.stderr.decode("utf-8", errors="replace") if r.stderr else "" + return _ProcOut(returncode=r.returncode, stdout=stdout_txt, stderr=stderr_txt) + + +def _parse_ok_data(line: str) -> dict: + j = json.loads(line.strip().splitlines()[0]) + assert j.get("success") is True + return j["data"] + + +@pytest.fixture() +def iso_cli_env(): + """Function-scoped data root for subprocess CLI tests.""" + tmp = tempfile.mkdtemp(prefix="cli_stage_c_") + data_root = os.path.join(tmp, "claw-data") + os.makedirs(data_root) + uid = "77701" + yield {"data_root": data_root, "user_id": uid, "tmp_dir": tmp} + shutil.rmtree(tmp, ignore_errors=True) + + +@pytest.fixture() +def with_master_key(monkeypatch): + from cryptography.fernet import Fernet + + key = Fernet.generate_key().decode("ascii") + monkeypatch.setenv("ACCOUNT_MANAGER_MASTER_KEY", key) + yield key + + +class TestAddWebV2: + def test_add_web_explicit_auth_strategy(self, iso_cli_env): + env = _cli_env(iso_cli_env) + proc = _run_cli( + env, + [ + "account", + "add-web", + "--platform", + "maersk", + "--label", + "demo", + "--auth-strategy", + "qr_code_manual", + "--session-persistent", + "1", + ], + ) + assert proc.returncode == 0 + data = _parse_ok_data(proc.stdout) + assert data["auth_strategy"] == "qr_code_manual" + assert data["session_persistent"] == 1 + conn = sqlite3.connect(_db_path(iso_cli_env)) + try: + row = conn.execute( + "SELECT auth_strategy, session_persistent FROM accounts WHERE id = ?", + (data["id"],), + ).fetchone() + assert row == ("qr_code_manual", 1) + finally: + conn.close() + + def test_add_web_platform_default_maersk(self, iso_cli_env): + env = _cli_env(iso_cli_env) + proc = _run_cli( + env, + ["account", "add-web", "--platform", "maersk", "--label", "demo2"], + ) + assert proc.returncode == 0 + data = _parse_ok_data(proc.stdout) + assert data["auth_strategy"] == "qr_code_manual" + + def test_add_web_bad_auth_strategy(self, iso_cli_env): + env = _cli_env(iso_cli_env) + proc = _run_cli( + env, + [ + "account", + "add-web", + "--platform", + "maersk", + "--label", + "x", + "--auth-strategy", + "foo", + ], + ) + assert proc.returncode == 0 + j = json.loads(proc.stdout.strip().splitlines()[0]) + assert j["success"] is False + assert j["error"]["code"] == "ERR_AUTH_STRATEGY_NOT_SUPPORTED" + conn = sqlite3.connect(_db_path(iso_cli_env)) + try: + n = conn.execute("SELECT COUNT(*) FROM accounts").fetchone()[0] + assert n == 0 + finally: + conn.close() + + def test_add_web_device_fingerprint_wrong_strategy(self, iso_cli_env): + env = _cli_env(iso_cli_env) + proc = _run_cli( + env, + [ + "account", + "add-web", + "--platform", + "maersk", + "--label", + "x", + "--auth-strategy", + "password_auto", + "--device-fingerprint", + "fp1", + ], + ) + assert proc.returncode == 0 + j = json.loads(proc.stdout.strip().splitlines()[0]) + assert j["success"] is False + assert j["error"]["code"] == "ERR_DEVICE_FINGERPRINT_NOT_APPLICABLE" + + +class TestAddSecretLocalEncrypted: + def test_add_secret_local_encrypted_success(self, iso_cli_env, with_master_key): + env = _cli_env(iso_cli_env) + proc = _run_cli( + env, + [ + "account", + "add-secret", + "--platform", + "maersk", + "--credential-type", + "password", + "--secret-storage", + "local_encrypted", + "--secret-stdin", + "--label", + "pw1", + ], + stdin="my-pw\n", + ) + assert proc.returncode == 0 + data = _parse_ok_data(proc.stdout) + assert data["secret_storage"] == "local_encrypted" + conn = sqlite3.connect(_db_path(iso_cli_env)) + try: + row = conn.execute( + "SELECT secret_storage, secret_mask, length(secret_ciphertext) FROM credentials WHERE id = ?", + (data["credential_id"],), + ).fetchone() + assert row is not None + assert row[0] == "local_encrypted" + assert row[1] # mask non-empty + assert row[2] and row[2] > 0 + finally: + conn.close() + + def test_add_secret_local_encrypted_requires_stdin(self, iso_cli_env, with_master_key): + env = _cli_env(iso_cli_env) + proc = _run_cli( + env, + [ + "account", + "add-secret", + "--platform", + "maersk", + "--credential-type", + "password", + "--secret-storage", + "local_encrypted", + "--label", + "x", + ], + ) + assert proc.returncode == 0 + j = json.loads(proc.stdout.strip().splitlines()[0]) + assert j["success"] is False + assert j["error"]["code"] == "ERR_LOCAL_ENCRYPTED_REQUIRES_STDIN" + + def test_add_secret_local_encrypted_type_not_allowed(self, iso_cli_env, with_master_key): + env = _cli_env(iso_cli_env) + proc = _run_cli( + env, + [ + "account", + "add-secret", + "--platform", + "maersk", + "--credential-type", + "browser_profile", + "--secret-storage", + "local_encrypted", + "--secret-stdin", + ], + stdin="x\n", + ) + assert proc.returncode == 0 + j = json.loads(proc.stdout.strip().splitlines()[0]) + assert j["success"] is False + assert j["error"]["code"] == "ERR_LOCAL_ENCRYPTED_TYPE_NOT_ALLOWED" + + def test_add_secret_local_encrypted_master_missing(self, iso_cli_env, monkeypatch): + monkeypatch.delenv("ACCOUNT_MANAGER_MASTER_KEY", raising=False) + env = _cli_env(iso_cli_env) + env.pop("ACCOUNT_MANAGER_MASTER_KEY", None) + proc = _run_cli( + env, + [ + "account", + "add-secret", + "--platform", + "maersk", + "--credential-type", + "password", + "--secret-storage", + "local_encrypted", + "--secret-stdin", + ], + stdin="pw\n", + ) + assert proc.returncode == 0 + j = json.loads(proc.stdout.strip().splitlines()[0]) + assert j["success"] is False + assert j["error"]["code"] == "ERR_MASTER_KEY_MISSING" + + +class TestPickWebReturns: + def test_pick_web_has_v2_fields(self, iso_cli_env, with_master_key): + env = _cli_env(iso_cli_env) + aw = _run_cli( + env, + ["account", "add-web", "--platform", "maersk", "--label", "pw"], + ) + assert aw.returncode == 0 + aid = _parse_ok_data(aw.stdout)["id"] + sec = _run_cli( + env, + [ + "account", + "add-secret", + "--platform", + "maersk", + "--credential-type", + "password", + "--secret-storage", + "local_encrypted", + "--secret-stdin", + "--account-id", + str(aid), + ], + stdin="secret-line\n", + ) + assert sec.returncode == 0 + + pick = _run_cli(env, ["account", "pick-web", "--platform", "maersk"]) + assert pick.returncode == 0 + payload = json.loads(pick.stdout.strip().splitlines()[0]) + assert "auth_strategy" in payload + assert "session_persistent" in payload + assert "device_fingerprint" in payload + + def test_pick_web_legacy_keys_present(self, iso_cli_env): + env = _cli_env(iso_cli_env) + aw = _run_cli( + env, + ["account", "add-web", "--platform", "maersk", "--label", "legacy"], + ) + assert aw.returncode == 0 + pick = _run_cli(env, ["account", "pick-web", "--platform", "maersk"]) + assert pick.returncode == 0 + payload = json.loads(pick.stdout.strip().splitlines()[0]) + for k in ("id", "platform_key", "login_id", "profile_dir", "url"): + assert k in payload + + +class TestGetCredential: + def test_get_credential_no_reveal_no_plaintext(self, iso_cli_env, with_master_key): + env = _cli_env(iso_cli_env) + aw = _run_cli( + env, + ["account", "add-web", "--platform", "maersk", "--label", "g1"], + ) + aid = _parse_ok_data(aw.stdout)["id"] + sec = _run_cli( + env, + [ + "account", + "add-secret", + "--platform", + "maersk", + "--credential-type", + "password", + "--secret-storage", + "local_encrypted", + "--secret-stdin", + "--account-id", + str(aid), + ], + stdin="hidden-pw\n", + ) + assert sec.returncode == 0 + + g = _run_cli(env, ["account", "get-credential", str(aid)]) + assert g.returncode == 0 + j = json.loads(g.stdout.strip().splitlines()[0]) + assert j["success"] is True + assert "plaintext" not in j + + def test_get_credential_reveal_missing_lease_token(self, iso_cli_env, with_master_key): + env = _cli_env(iso_cli_env) + aw = _run_cli(env, ["account", "add-web", "--platform", "maersk", "--label", "g2"]) + aid = _parse_ok_data(aw.stdout)["id"] + sec = _run_cli( + env, + [ + "account", + "add-secret", + "--platform", + "maersk", + "--credential-type", + "password", + "--secret-storage", + "local_encrypted", + "--secret-stdin", + "--account-id", + str(aid), + ], + stdin="p\n", + ) + assert sec.returncode == 0 + + g = _run_cli(env, ["account", "get-credential", str(aid), "--reveal"]) + j = json.loads(g.stdout.strip().splitlines()[0]) + assert j["success"] is False + assert j["error"]["code"] == "ERR_LEASE_INVALID" + + def test_get_credential_wrong_lease_account(self, iso_cli_env, with_master_key): + env = _cli_env(iso_cli_env) + a1 = _parse_ok_data( + _run_cli( + env, + ["account", "add-web", "--platform", "maersk", "--label", "acc1"], + ).stdout + )["id"] + sec1 = _run_cli( + env, + [ + "account", + "add-secret", + "--platform", + "maersk", + "--credential-type", + "password", + "--secret-storage", + "local_encrypted", + "--secret-stdin", + "--account-id", + str(a1), + ], + stdin="s1\n", + ) + assert sec1.returncode == 0 + + lease_out = _run_cli( + env, + ["account", "pick-web", "--platform", "maersk", "--lease", "--holder", "t"], + ) + tok = json.loads(lease_out.stdout.strip().splitlines()[0])["lease_token"] + + a2 = _parse_ok_data( + _run_cli( + env, + ["account", "add-web", "--platform", "maersk", "--label", "acc2"], + ).stdout + )["id"] + sec2 = _run_cli( + env, + [ + "account", + "add-secret", + "--platform", + "maersk", + "--credential-type", + "password", + "--secret-storage", + "local_encrypted", + "--secret-stdin", + "--account-id", + str(a2), + ], + stdin="s2\n", + ) + assert sec2.returncode == 0 + + bad = _run_cli( + env, + [ + "account", + "get-credential", + str(a2), + "--reveal", + "--lease-token", + tok, + ], + ) + j = json.loads(bad.stdout.strip().splitlines()[0]) + assert j["success"] is False + assert j["error"]["code"] == "ERR_LEASE_INVALID" + + def test_get_credential_reveal_success(self, iso_cli_env, with_master_key): + env = _cli_env(iso_cli_env) + a1 = _parse_ok_data( + _run_cli( + env, + ["account", "add-web", "--platform", "maersk", "--label", "okc"], + ).stdout + )["id"] + secret_text = "correct-horse-battery" + sec = _run_cli( + env, + [ + "account", + "add-secret", + "--platform", + "maersk", + "--credential-type", + "password", + "--secret-storage", + "local_encrypted", + "--secret-stdin", + "--account-id", + str(a1), + ], + stdin=secret_text + "\n", + ) + assert sec.returncode == 0 + + lease_out = _run_cli( + env, + ["account", "pick-web", "--platform", "maersk", "--lease", "--holder", "h"], + ) + tok = json.loads(lease_out.stdout.strip().splitlines()[0])["lease_token"] + + g = _run_cli( + env, + [ + "account", + "get-credential", + str(a1), + "--reveal", + "--lease-token", + tok, + ], + ) + j = json.loads(g.stdout.strip().splitlines()[0]) + assert j["success"] is True + assert j["plaintext"] == secret_text