tests/test_xbrain_recall_runtime.py
Rama main · réplica de lectura del repositorio original.
from __future__ import annotations
import json
import os
import subprocess
import sys
import tempfile
import unittest
from unittest import mock
from pathlib import Path
ROOT = Path(__file__).resolve().parents[1]
SCRIPT = ROOT / "scripts" / "xbrain_recall.py"
sys.path.insert(0, str(Path(__file__).resolve().parent))
from _isolation import isolated_env, needs_posix_exec
class XbrainRecallRuntimeTests(unittest.TestCase):
def test_backend_diagnostics_distinguish_empty_failure_and_timeout(self):
sys.path.insert(0, str(ROOT / "scripts"))
import xbrain_recall as xb
outcomes = [(subprocess.CompletedProcess([], 0, "[]", ""), "empty"),
(subprocess.CompletedProcess([], 7, "", "secret"), "error"),
(subprocess.CompletedProcess([], 0, "unexpected output", ""), "invalid_response"),
(subprocess.TimeoutExpired("fixture", 1), "timeout")]
with mock.patch.object(xb, "_resolve_gbrain_dir", return_value=Path("fixture")):
for outcome, expected in outcomes:
state = {}
kwargs = {"side_effect": outcome} if isinstance(outcome, Exception) else {"return_value": outcome}
with mock.patch.object(xb.subprocess, "run", **kwargs), mock.patch.object(xb.xkb_failures, "note"):
self.assertEqual(xb.xbrain_query("fixture", diagnostics=state), [])
self.assertEqual(state["status"], expected)
self.assertTrue(state["attempted"])
self.assertNotIn("secret", str(state))
def make_fixture(self, root: Path) -> tuple[Path, Path]:
gbrain = root / "gbrain"
(gbrain / "src").mkdir(parents=True)
(gbrain / "src" / "cli.ts").write_text("// fixture\n", encoding="utf-8")
fake_bin = root / "bin"
fake_bin.mkdir()
bun = fake_bin / "bun"
bun.write_text(
"#!/usr/bin/env python3\n"
"import json, os, sys\n"
"if os.environ.get('FIXTURE_BUN_FAILURE'):\n"
" print(os.environ.get('GEMINI_API_KEY', ''), file=sys.stderr)\n"
" raise SystemExit(7)\n"
"key = os.environ.get('GEMINI_API_KEY', '')\n"
"source = {'process-placeholder': 'process', 'explicit-placeholder': 'explicit', 'inherited-placeholder': 'inherited'}.get(key, 'none')\n"
"print(json.dumps([{'slug': source, 'chunk_text': 'fixture', 'score': 1.0}]))\n",
encoding="utf-8",
)
bun.chmod(0o755)
return gbrain, fake_bin
def run_cli(
self,
root: Path,
*,
env_file: Path | None = None,
explicit_env_file: Path | None = None,
process_key: str | None = None,
extra_env: dict[str, str] | None = None,
) -> subprocess.CompletedProcess[str]:
gbrain, fake_bin = self.make_fixture(root)
isolated_home = root / "isolated-home"
isolated_home.mkdir(exist_ok=True)
env = isolated_env(
home=isolated_home,
PATH=str(fake_bin) + os.pathsep + os.environ.get("PATH", ""),
GBRAIN_DIR=str(gbrain),
XKB_DATA_DIR=str(root / "runtime-data"),
PYTHONPATH="",
)
if env_file is not None:
env["XKB_ENV_FILE"] = str(env_file)
if process_key is not None:
env["GEMINI_API_KEY"] = process_key
if extra_env:
env.update(extra_env)
command = [sys.executable, str(SCRIPT), "fixture query", "--json"]
if explicit_env_file is not None:
command.extend(["--env-file", str(explicit_env_file)])
return subprocess.run(command, cwd=ROOT, env=env, capture_output=True, text=True, encoding="utf-8", errors="replace", check=False)
@needs_posix_exec
def test_process_environment_takes_precedence_over_xkb_env_file(self) -> None:
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
env_file = root / "runtime.env"
env_file.write_text("GEMINI_API_KEY=file-placeholder\n", encoding="utf-8")
result = self.run_cli(root, env_file=env_file, process_key="process-placeholder")
self.assertEqual(result.returncode, 0, result.stderr)
self.assertEqual(json.loads(result.stdout)[0]["slug"], "process")
@needs_posix_exec
def test_explicit_env_file_is_used_instead_of_xkb_env_file(self) -> None:
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
inherited = root / "inherited.env"
explicit = root / "explicit.env"
inherited.write_text("GEMINI_API_KEY=inherited-placeholder\n", encoding="utf-8")
explicit.write_text("GEMINI_API_KEY=explicit-placeholder\n", encoding="utf-8")
result = self.run_cli(root, env_file=inherited, explicit_env_file=explicit)
self.assertEqual(result.returncode, 0, result.stderr)
self.assertEqual(json.loads(result.stdout)[0]["slug"], "explicit")
@needs_posix_exec
def test_explicit_env_file_still_yields_to_process_environment(self) -> None:
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
explicit = root / "explicit.env"
explicit.write_text("GEMINI_API_KEY=explicit-placeholder\n", encoding="utf-8")
result = self.run_cli(root, explicit_env_file=explicit, process_key="process-placeholder")
self.assertEqual(result.returncode, 0, result.stderr)
self.assertEqual(json.loads(result.stdout)[0]["slug"], "process")
def test_missing_credential_is_actionable_without_host_fallback_or_secret_leak(self) -> None:
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
secret = "host-secret-placeholder"
result = self.run_cli(
root,
extra_env={"OPENCLAW_JSON": secret},
)
combined = result.stdout + result.stderr
self.assertEqual(result.returncode, 1)
self.assertIn("GEMINI_API_KEY", result.stderr)
self.assertIn("process environment", result.stderr)
self.assertFalse(secret in combined)
self.assertFalse((root / "isolated-home" / ".openclaw").exists())
def test_active_recall_source_has_no_private_config_fallback(self) -> None:
source = SCRIPT.read_text(encoding="utf-8")
for forbidden in ("OPENCLAW_JSON", "openclaw.json", "$HOME/.openclaw", "/root/.openclaw"):
self.assertNotIn(forbidden, source)
def test_failed_backend_does_not_echo_credential(self) -> None:
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
secret = "fixture-secret-that-must-not-leak"
result = self.run_cli(
root,
process_key=secret,
extra_env={"FIXTURE_BUN_FAILURE": "1"},
)
combined = result.stdout + result.stderr
self.assertEqual(result.returncode, 0, result.stderr)
self.assertEqual(json.loads(result.stdout), [])
self.assertNotIn(secret, combined)
if __name__ == "__main__":
unittest.main()