151 lines
4.7 KiB
Python
151 lines
4.7 KiB
Python
# -*- coding: utf-8 -*-
|
|
import json
|
|
import os
|
|
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")
|
|
|
|
|
|
def _iso_cli_env(data_root: str, uid: str = "cli_u") -> dict:
|
|
"""Align skill roots so subprocess + in-process tests hit the same DB."""
|
|
return {
|
|
"JIANGCHANG_DATA_ROOT": data_root,
|
|
"JIANGCHANG_USER_ID": uid,
|
|
"CLAW_DATA_ROOT": data_root,
|
|
"CLAW_USER_ID": uid,
|
|
}
|
|
|
|
|
|
@dataclass
|
|
class _ProcOut:
|
|
returncode: int
|
|
stdout: str
|
|
stderr: str
|
|
|
|
|
|
def _run_cli(env: dict, args: list[str], stdin: str | None = None) -> _ProcOut:
|
|
merged = {**os.environ, **env}
|
|
merged.pop("ACCOUNT_MANAGER_MASTER_KEY", None)
|
|
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=merged,
|
|
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 test_init_master_key_writes_file(monkeypatch):
|
|
tmp = tempfile.mkdtemp(prefix="cli_mk_")
|
|
data_root = os.path.join(tmp, "dr")
|
|
os.makedirs(data_root)
|
|
env = _iso_cli_env(data_root)
|
|
proc = _run_cli(env, ["account", "init-master-key"])
|
|
assert proc.returncode == 0
|
|
payload = json.loads(proc.stdout.strip().splitlines()[0])
|
|
assert payload["success"] is True
|
|
assert os.path.isfile(payload["path"])
|
|
|
|
|
|
def test_init_master_key_rejects_duplicate(monkeypatch):
|
|
tmp = tempfile.mkdtemp(prefix="cli_mk_d_")
|
|
data_root = os.path.join(tmp, "dr")
|
|
os.makedirs(data_root)
|
|
env = _iso_cli_env(data_root)
|
|
p1 = _run_cli(env, ["account", "init-master-key"])
|
|
assert p1.returncode == 0
|
|
p2 = _run_cli(env, ["account", "init-master-key"])
|
|
assert p2.returncode == 0
|
|
payload = json.loads(p2.stdout.strip().splitlines()[0])
|
|
assert payload["success"] is False
|
|
assert payload["error"]["code"] == "ERR_MASTER_KEY_EXISTS"
|
|
|
|
|
|
def test_export_import_roundtrip(monkeypatch):
|
|
tmp = tempfile.mkdtemp(prefix="cli_ex_")
|
|
data_root = os.path.join(tmp, "dr")
|
|
os.makedirs(data_root)
|
|
env = _iso_cli_env(data_root)
|
|
|
|
init = _run_cli(env, ["account", "init-master-key"])
|
|
assert init.returncode == 0
|
|
|
|
# Insert encrypted credential via repo in-process (same paths as CLI)
|
|
monkeypatch.setenv("JIANGCHANG_DATA_ROOT", data_root)
|
|
monkeypatch.setenv("JIANGCHANG_USER_ID", "cli_u")
|
|
monkeypatch.setenv("CLAW_DATA_ROOT", data_root)
|
|
monkeypatch.setenv("CLAW_USER_ID", "cli_u")
|
|
monkeypatch.delenv("ACCOUNT_MANAGER_MASTER_KEY", raising=False)
|
|
|
|
from util.constants import SKILL_SLUG
|
|
from util.logging_config import setup_skill_logging
|
|
|
|
setup_skill_logging(SKILL_SLUG, "openclaw.skill.account_manager_cli_inline")
|
|
|
|
from db.connection import get_conn, init_db
|
|
from db.credentials_repo import insert_credential_encrypted, read_credential_plaintext
|
|
|
|
init_db()
|
|
conn = get_conn()
|
|
try:
|
|
cid = insert_credential_encrypted(
|
|
conn,
|
|
account_id=None,
|
|
platform_key="icbc_sim",
|
|
credential_label="ICBC password label",
|
|
credential_type="password",
|
|
plaintext="roundtrip-secret",
|
|
expires_at=None,
|
|
extra_json="{}",
|
|
now=1_750_000_000,
|
|
)
|
|
conn.commit()
|
|
finally:
|
|
conn.close()
|
|
|
|
conn = get_conn()
|
|
try:
|
|
cur = conn.cursor()
|
|
cur.execute("SELECT COUNT(*) FROM credentials")
|
|
assert int(cur.fetchone()[0]) == 1
|
|
finally:
|
|
conn.close()
|
|
|
|
ex = _run_cli(env, ["account", "export-credentials", "--plaintext"])
|
|
assert ex.returncode == 0
|
|
arr = json.loads(ex.stdout.strip())
|
|
assert len(arr) == 1
|
|
assert arr[0]["plaintext"] == "roundtrip-secret"
|
|
|
|
imp = _run_cli(
|
|
env,
|
|
["account", "import-credentials", "--replace-all"],
|
|
stdin=json.dumps(arr, ensure_ascii=False),
|
|
)
|
|
assert imp.returncode == 0
|
|
|
|
conn = get_conn()
|
|
try:
|
|
cur = conn.cursor()
|
|
cur.execute("SELECT COUNT(*) FROM credentials WHERE secret_storage='local_encrypted'")
|
|
assert int(cur.fetchone()[0]) == 1
|
|
cur.execute("SELECT id FROM credentials LIMIT 1")
|
|
new_id = int(cur.fetchone()[0])
|
|
pt = read_credential_plaintext(conn, new_id)
|
|
assert pt == "roundtrip-secret"
|
|
finally:
|
|
conn.close()
|