test(crypto+db+cli): add master key, crypto, credentials_repo, and CLI tests

Co-authored-by: Cursor <cursoragent@cursor.com>

tests/test_cli_master_key.py

@@ -0,0 +1,150 @@
+# -*- 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()