client-commons/account-manager
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

test(cli): add stage C CLI tests

Browse Source 
Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
chendelian
2026-05-11 11:25:53 +08:00
parent 04cafae3d7
commit 65d939dc5b
1 changed files with 511 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

511
tests/test_cli_stage_c.py Normal file
View File

@@ -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