merge(chv-muse-round12): 合入 P7 状态与记忆分离 (29fa0c8)
冲突按并集解：build.sh 保留 P2+P3+P4 检查/拷入 + P7 的 chv-memory 检查/拷入（P7 的重复 runcmd 行删掉——同一文件 写两次会盖掉 authd 启动行）；buildinfo.py 加 chv-memory 一项； production-readiness.md P7 一节顺延为 §18。P7 测试 46 项全过。
This commit is contained in:
host committed
commit
320cd36921
8 files changed
+1667
-1
No files matched your search
@@ -0,0 +1,502 @@
|
||||
"""对齐 Muse P7 测试：状态与记忆分离。
|
||||
|
||||
只测本轨交付物（cloud/chv/app_state.py、memory_files.py、
|
||||
chv-guest/chv-memory.py、build.sh 干跑、buildinfo.py 内容项）。
|
||||
真实记忆系统 CLI 只在本地临时 home 上跑（init/write/search/
|
||||
forget/backup/restore 全是真调用）；其余用假 ssh/假 CLI。
|
||||
绝不连真实网络、不起真实 VM、不改生产行为。
|
||||
"""
|
||||
import importlib.util
|
||||
import io
|
||||
import json
|
||||
import os
|
||||
import shutil
|
||||
import sqlite3
|
||||
import subprocess
|
||||
import sys
|
||||
import tarfile
|
||||
import tempfile
|
||||
import unittest
|
||||
from contextlib import redirect_stdout
|
||||
from pathlib import Path
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[2]
|
||||
WORKSPACE = Path(__file__).resolve().parents[3]
|
||||
MEMSYS_CLI = WORKSPACE / "memory-system" / "dist" / "cli.js"
|
||||
|
||||
sys.path.insert(0, str(ROOT))
|
||||
|
||||
from cloud.chv import app_state
|
||||
from cloud.chv import memory_files
|
||||
from cloud.chv.app_state import (AppStateStore, SecretRejectedError,
|
||||
restore_app_state, state_db_path,
|
||||
verify_backup)
|
||||
from cloud.chv.memory_files import (GUEST_MEMORY_HOME, MemoryFiles,
|
||||
MemorySystemClient,
|
||||
build_pull_command, forget,
|
||||
memory_home_path, pull_memory_home)
|
||||
|
||||
|
||||
def _load_guest_script(name):
|
||||
"""把 chv-guest 下的独立脚本当模块加载（无包结构）。"""
|
||||
path = ROOT / "chv-guest" / name
|
||||
spec = importlib.util.spec_from_file_location(
|
||||
"chv_guest_p7_" + name[:-3], str(path))
|
||||
module = importlib.util.module_from_spec(spec)
|
||||
spec.loader.exec_module(module)
|
||||
return module
|
||||
|
||||
|
||||
CHV_MEMORY = _load_guest_script("chv-memory.py")
|
||||
BUILDINFO = _load_guest_script("buildinfo.py")
|
||||
|
||||
|
||||
def _has_cli():
|
||||
"""真实记忆系统 CLI 是否在位。"""
|
||||
return os.path.isfile(str(MEMSYS_CLI)) and shutil.which("node")
|
||||
|
||||
|
||||
class SecretGuardTest(unittest.TestCase):
|
||||
def test_key_name_token_rejected(self):
|
||||
with self.assertRaises(SecretRejectedError) as ctx:
|
||||
app_state.assert_no_secrets({"GITEA_TOKEN": "abc"})
|
||||
self.assertNotIn("abc", str(ctx.exception))
|
||||
|
||||
def test_key_name_password_rejected(self):
|
||||
with self.assertRaises(SecretRejectedError):
|
||||
app_state.assert_no_secrets({"nested": {"db_password": "x"}})
|
||||
|
||||
def test_value_pattern_ghp_rejected(self):
|
||||
with self.assertRaises(SecretRejectedError) as ctx:
|
||||
app_state.assert_no_secrets(
|
||||
{"note": "see ghp_" + "a" * 24 + " here"})
|
||||
self.assertNotIn("ghp_", str(ctx.exception))
|
||||
|
||||
def test_plain_text_passes(self):
|
||||
app_state.assert_no_secrets(
|
||||
{"title": "买菜", "note": "记得买冰美式豆子"})
|
||||
|
||||
|
||||
class AppStateStoreTest(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.tmp = tempfile.TemporaryDirectory()
|
||||
self.path = os.path.join(self.tmp.name, "vm1", "app-state.sqlite")
|
||||
self.store = AppStateStore(self.path)
|
||||
self.store.open()
|
||||
|
||||
def tearDown(self):
|
||||
self.store.close()
|
||||
self.tmp.cleanup()
|
||||
|
||||
def test_open_sets_schema_and_modes(self):
|
||||
self.assertEqual(self.store.schema_version(), "1")
|
||||
self.assertEqual(oct(os.stat(self.path).st_mode & 0o777), "0o600")
|
||||
parent = os.path.dirname(self.path)
|
||||
self.assertEqual(oct(os.stat(parent).st_mode & 0o777), "0o700")
|
||||
|
||||
def test_state_db_path_helper(self):
|
||||
self.assertEqual(state_db_path("/s", "vm1"),
|
||||
"/s/vm1/app-state.sqlite")
|
||||
|
||||
def test_task_roundtrip(self):
|
||||
self.store.put_task("u1", "t1", "code", "running",
|
||||
{"repo": "admin/x", "step": "build"})
|
||||
got = self.store.get_task("t1")
|
||||
self.assertEqual(got["status"], "running")
|
||||
self.assertEqual(got["payload"]["repo"], "admin/x")
|
||||
self.assertIsNone(self.store.get_task("nope"))
|
||||
listed = self.store.list_tasks("u1")
|
||||
self.assertEqual([t["id"] for t in listed], ["t1"])
|
||||
self.assertEqual(self.store.list_tasks("u1", "done"), [])
|
||||
|
||||
def test_update_task_status(self):
|
||||
self.store.put_task("u1", "t1", "code", "running", {})
|
||||
self.assertTrue(self.store.update_task_status("t1", "done"))
|
||||
self.assertEqual(self.store.get_task("t1")["status"], "done")
|
||||
self.assertFalse(self.store.update_task_status("nope", "done"))
|
||||
|
||||
def test_goal_and_session_roundtrip(self):
|
||||
self.store.put_goal("u1", "g1", "学会做饭",
|
||||
{"steps": ["买菜", "开火"]}, "active")
|
||||
goal = self.store.get_goal("g1")
|
||||
self.assertEqual(goal["title"], "学会做饭")
|
||||
self.assertEqual(goal["plan"]["steps"][0], "买菜")
|
||||
self.assertEqual(len(self.store.list_goals("u1")), 1)
|
||||
self.store.put_session("u1", "s1", "chat-9", "聊了晚饭")
|
||||
session = self.store.get_session("s1")
|
||||
self.assertEqual(session["chat_id"], "chat-9")
|
||||
self.assertIsNone(self.store.get_session("nope"))
|
||||
|
||||
def test_task_with_secret_key_refused_and_not_stored(self):
|
||||
with self.assertRaises(SecretRejectedError):
|
||||
self.store.put_task("u1", "t2", "code", "running",
|
||||
{"GITEA_TOKEN": "abc"})
|
||||
self.assertIsNone(self.store.get_task("t2"))
|
||||
|
||||
def test_task_with_secret_value_refused(self):
|
||||
with self.assertRaises(SecretRejectedError):
|
||||
self.store.put_task("u1", "t2", "code", "running",
|
||||
{"note": "key sk-" + "b" * 24})
|
||||
|
||||
def test_audit_append_only(self):
|
||||
rid = self.store.append_audit("host", "task-start", "t1",
|
||||
{"kind": "code"})
|
||||
self.assertTrue(rid)
|
||||
rows = self.store.list_audit(target="t1")
|
||||
self.assertEqual(len(rows), 1)
|
||||
self.assertEqual(rows[0]["action"], "task-start")
|
||||
source = Path(app_state.__file__).read_text()
|
||||
self.assertNotIn("DELETE FROM audit", source)
|
||||
self.assertNotIn("UPDATE audit", source)
|
||||
|
||||
def test_audit_detail_secret_refused(self):
|
||||
with self.assertRaises(SecretRejectedError):
|
||||
self.store.append_audit("host", "x", "t1",
|
||||
{"token": "ghp_" + "c" * 24})
|
||||
|
||||
def test_counts(self):
|
||||
self.store.put_task("u1", "t1", "code", "running", {})
|
||||
self.store.put_goal("u1", "g1", "标题", {}, "active")
|
||||
self.assertEqual(self.store.counts()["tasks"], 1)
|
||||
self.assertEqual(self.store.counts()["goals"], 1)
|
||||
self.assertEqual(self.store.counts()["audit"], 0)
|
||||
|
||||
def test_ddl_has_no_sqlite_only_syntax(self):
|
||||
self.assertNotIn("AUTOINCREMENT", app_state.DDL)
|
||||
self.assertNotIn("PRAGMA", app_state.DDL)
|
||||
for keyword in ("CREATE TABLE IF NOT EXISTS meta",
|
||||
"CREATE TABLE IF NOT EXISTS tasks",
|
||||
"CREATE TABLE IF NOT EXISTS goals",
|
||||
"CREATE TABLE IF NOT EXISTS sessions",
|
||||
"CREATE TABLE IF NOT EXISTS audit"):
|
||||
self.assertIn(keyword, app_state.DDL)
|
||||
|
||||
def test_backup_restore_roundtrip(self):
|
||||
self.store.put_task("u1", "t1", "code", "done", {"n": "1"})
|
||||
self.store.append_audit("host", "done", "t1", {})
|
||||
snap = os.path.join(self.tmp.name, "snap")
|
||||
manifest = self.store.backup(snap)
|
||||
self.assertEqual(manifest["tasks"], "1")
|
||||
self.assertEqual(verify_backup(snap)["sha256"],
|
||||
manifest["sha256"])
|
||||
self.store.close()
|
||||
os.remove(self.path)
|
||||
restore_app_state(snap, self.path)
|
||||
self.store.open()
|
||||
self.assertEqual(self.store.get_task("t1")["status"], "done")
|
||||
self.assertEqual(self.store.counts()["audit"], 1)
|
||||
|
||||
def test_restore_keeps_pre_restore_copy(self):
|
||||
self.store.put_task("u1", "t1", "code", "running", {})
|
||||
snap = os.path.join(self.tmp.name, "snap")
|
||||
self.store.backup(snap)
|
||||
self.store.close()
|
||||
restore_app_state(snap, self.path)
|
||||
preserved = [p for p in os.listdir(os.path.dirname(self.path))
|
||||
if p.startswith("app-state.sqlite.pre-restore-")]
|
||||
self.assertEqual(len(preserved), 1)
|
||||
|
||||
def test_restore_corrupt_backup_refused(self):
|
||||
self.store.put_task("u1", "t1", "code", "running", {})
|
||||
snap = os.path.join(self.tmp.name, "snap")
|
||||
self.store.backup(snap)
|
||||
self.store.close()
|
||||
with open(os.path.join(snap, "app-state.sqlite"),
|
||||
"ab") as handle:
|
||||
handle.write(b"corrupt")
|
||||
with self.assertRaises(app_state.AppStateError):
|
||||
restore_app_state(snap, self.path)
|
||||
self.store.open()
|
||||
self.assertEqual(self.store.get_task("t1")["status"], "running")
|
||||
|
||||
def test_backup_has_no_token_bytes(self):
|
||||
sentinel = "ghp_" + "z" * 24
|
||||
with self.assertRaises(SecretRejectedError):
|
||||
self.store.put_task("u1", "t9", "code", "running",
|
||||
{"note": sentinel})
|
||||
snap = os.path.join(self.tmp.name, "snap")
|
||||
self.store.backup(snap)
|
||||
blob = Path(os.path.join(snap, "app-state.sqlite")).read_bytes()
|
||||
self.assertNotIn(b"ghp_", blob)
|
||||
manifest = Path(os.path.join(snap, "manifest.json")).read_bytes()
|
||||
self.assertNotIn(b"ghp_", manifest)
|
||||
|
||||
def test_store_not_open_raises(self):
|
||||
closed = AppStateStore(os.path.join(self.tmp.name, "x.db"))
|
||||
with self.assertRaises(app_state.AppStateError):
|
||||
closed.get_task("t1")
|
||||
|
||||
|
||||
class SeparationTest(unittest.TestCase):
|
||||
def test_store_path_outside_guest_disk(self):
|
||||
db = state_db_path("/state", "vm1")
|
||||
disk = "/state/vm1/disk.raw"
|
||||
self.assertNotEqual(db, disk)
|
||||
self.assertFalse(db.startswith(disk + "/"))
|
||||
|
||||
def test_memory_home_path_outside_guest_disk(self):
|
||||
home = memory_home_path("/state", "vm1")
|
||||
self.assertEqual(home, "/state/vm1/memory-home")
|
||||
self.assertNotEqual(home, "/state/vm1/disk.raw")
|
||||
|
||||
def test_guest_script_has_no_state_db_reference(self):
|
||||
source = (ROOT / "chv-guest" / "chv-memory.py").read_text()
|
||||
self.assertNotIn("app-state", source)
|
||||
self.assertNotIn("app_state", source)
|
||||
|
||||
def test_app_state_has_no_guest_paths(self):
|
||||
source = Path(app_state.__file__).read_text()
|
||||
self.assertNotIn("/home/chv", source)
|
||||
self.assertNotIn("GUEST_MEMORY_HOME", source)
|
||||
|
||||
def test_guest_list_cannot_see_store_file(self):
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
home = os.path.join(tmp, "memory")
|
||||
store = os.path.join(tmp, "app-state.sqlite")
|
||||
Path(store).write_text("fake-db")
|
||||
CHV_MEMORY._ensure(home)
|
||||
listed = CHV_MEMORY.cmd_list(home)["files"]
|
||||
self.assertNotIn("app-state.sqlite", listed)
|
||||
self.assertNotIn("../app-state.sqlite", listed)
|
||||
|
||||
class MemoryFilesTest(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.tmp = tempfile.TemporaryDirectory()
|
||||
self.home = os.path.join(self.tmp.name, "memory-home")
|
||||
self.files = MemoryFiles(self.home)
|
||||
self.files.ensure()
|
||||
|
||||
def tearDown(self):
|
||||
self.tmp.cleanup()
|
||||
|
||||
def test_ensure_layout(self):
|
||||
self.assertTrue(os.path.isfile(
|
||||
os.path.join(self.home, "MEMORY.md")))
|
||||
self.assertTrue(os.path.isdir(
|
||||
os.path.join(self.home, "memory")))
|
||||
self.assertEqual(oct(os.stat(self.home).st_mode & 0o777), "0o700")
|
||||
|
||||
def test_memory_home_path_helper(self):
|
||||
self.assertEqual(memory_home_path("/s", "vm1"),
|
||||
"/s/vm1/memory-home")
|
||||
|
||||
def test_write_read_list_append(self):
|
||||
self.files.write_file("MEMORY.md", "# 记忆\n", actor="host")
|
||||
self.files.append_lines("memory/2026-11-07.md",
|
||||
["- 喜欢冰美式", "- 怕辣"], actor="host")
|
||||
self.assertIn("MEMORY.md", self.files.list_files())
|
||||
self.assertIn("memory/2026-11-07.md", self.files.list_files())
|
||||
text = self.files.read_file("memory/2026-11-07.md")
|
||||
self.assertIn("冰美式", text)
|
||||
audit = Path(os.path.join(
|
||||
self.home, "memory", "audit.jsonl")).read_text()
|
||||
self.assertIn('"action": "write"', audit)
|
||||
self.assertIn('"action": "append"', audit)
|
||||
|
||||
def test_read_missing_raises(self):
|
||||
with self.assertRaises(memory_files.MemoryError):
|
||||
self.files.read_file("memory/nope.md")
|
||||
|
||||
def test_path_traversal_rejected(self):
|
||||
for bad in ("../escape.md", "../../etc/passwd",
|
||||
"/etc/passwd", "memory/../../escape.md"):
|
||||
with self.assertRaises(memory_files.MemoryError, msg=bad):
|
||||
self.files.read_file(bad)
|
||||
with self.assertRaises(memory_files.MemoryError, msg=bad):
|
||||
self.files.write_file(bad, "x")
|
||||
|
||||
def test_write_secret_refused(self):
|
||||
with self.assertRaises(SecretRejectedError):
|
||||
self.files.write_file("MEMORY.md", "token ghp_" + "d" * 24)
|
||||
|
||||
def test_remove_lines(self):
|
||||
self.files.write_file("MEMORY.md",
|
||||
"# 记忆\n- 喜欢冰美式\n- 怕辣\n")
|
||||
removed = self.files.remove_lines("MEMORY.md", "冰美式")
|
||||
self.assertEqual(removed, 1)
|
||||
self.assertNotIn("冰美式", self.files.read_file("MEMORY.md"))
|
||||
self.assertIn("怕辣", self.files.read_file("MEMORY.md"))
|
||||
|
||||
|
||||
class ForgetLocalOnlyTest(unittest.TestCase):
|
||||
def test_forget_local_only_marks_pending(self):
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
files = MemoryFiles(os.path.join(tmp, "home"))
|
||||
files.ensure()
|
||||
files.write_file("MEMORY.md", "# 记忆\n- 喜欢冰美式\n")
|
||||
result = forget(files, None, "冰美式", actor="host")
|
||||
self.assertEqual(result["mode"], "local-only")
|
||||
self.assertEqual(result["lines_removed"], 1)
|
||||
self.assertFalse(result["verified"])
|
||||
self.assertNotIn("冰美式", files.read_file("MEMORY.md"))
|
||||
audit = Path(os.path.join(
|
||||
tmp, "home", "memory", "audit.jsonl")).read_text()
|
||||
self.assertIn("forget-pending", audit)
|
||||
|
||||
|
||||
@unittest.skipUnless(_has_cli(), "memory-system CLI not available")
|
||||
class RealCliForgetTest(unittest.TestCase):
|
||||
def test_real_cli_write_search_forget_verify(self):
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
home = os.path.join(tmp, "home")
|
||||
client = MemorySystemClient(str(MEMSYS_CLI))
|
||||
self.assertTrue(client.available())
|
||||
client._run(["init", home])
|
||||
written = client._run([
|
||||
"write", home, "--kind", "fact",
|
||||
"--text", "用户喜欢喝冰美式",
|
||||
"--chat", "c1", "--message", "m1",
|
||||
"--speaker", "user", "--quote", "我喜欢喝冰美式"])
|
||||
claim_id = written.get("id")
|
||||
self.assertTrue(str(claim_id).startswith("claim:"))
|
||||
found = client.search(home, "冰美式")
|
||||
hits = found.get("hits")
|
||||
self.assertTrue(isinstance(hits, list) and len(hits) == 1)
|
||||
files = MemoryFiles(home)
|
||||
files.write_file("MEMORY.md",
|
||||
"# 记忆\n- 用户喜欢喝冰美式\n")
|
||||
result = forget(files, client, "冰美式", actor="host")
|
||||
self.assertEqual(result["mode"], "cli")
|
||||
self.assertEqual(result["claims"], 1)
|
||||
self.assertTrue(result["verified"])
|
||||
after = client.search(home, "冰美式")
|
||||
self.assertEqual(after.get("hits"), [])
|
||||
self.assertNotIn("冰美式", files.read_file("MEMORY.md"))
|
||||
|
||||
def test_real_cli_backup_restore(self):
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
home = os.path.join(tmp, "home")
|
||||
client = MemorySystemClient(str(MEMSYS_CLI))
|
||||
client._run(["init", home])
|
||||
client._run(["write", home, "--kind", "fact",
|
||||
"--text", "用户住在上海", "--chat", "c1",
|
||||
"--message", "m1", "--speaker", "user",
|
||||
"--quote", "我住在上海"])
|
||||
snap = client.backup(home)
|
||||
snapshot = snap.get("snapshot") or snap.get("path") or ""
|
||||
self.assertTrue(snapshot)
|
||||
victim = os.path.join(home, "MEMORY.md")
|
||||
with open(victim, "w", encoding="utf-8") as handle:
|
||||
handle.write("# 已破坏\n")
|
||||
client.restore(home, str(snapshot))
|
||||
# 恢复后 MEMORY.md 回到快照内容（不再是破坏后的文本）；
|
||||
# claim 本身活在 bank + 每日笔记里，search 能查到才是
|
||||
# 恢复成功的证据。
|
||||
self.assertNotIn("已破坏", Path(victim).read_text())
|
||||
found = client.search(home, "上海")
|
||||
self.assertTrue(len(found.get("hits", [])) >= 1)
|
||||
|
||||
def test_client_failure_raises(self):
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
fake = os.path.join(tmp, "fake-cli.js")
|
||||
Path(fake).write_text("#!/usr/bin/env node\n"
|
||||
"process.exit(1);\n")
|
||||
client = MemorySystemClient(fake)
|
||||
with self.assertRaises(memory_files.MemoryError):
|
||||
client.search(os.path.join(tmp, "home"), "x")
|
||||
|
||||
|
||||
class PullMemoryHomeTest(unittest.TestCase):
|
||||
def test_build_pull_command(self):
|
||||
cmd = build_pull_command(
|
||||
["ssh", "-p", "2222", "chv@127.0.0.1"], "/home/chv/memory")
|
||||
self.assertEqual(cmd, ["ssh", "-p", "2222", "chv@127.0.0.1",
|
||||
"tar -czf - -C /home/chv memory"])
|
||||
|
||||
def test_pull_with_fake_ssh(self):
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
remote_parent = os.path.join(tmp, "remote")
|
||||
remote_home = os.path.join(remote_parent, "memory")
|
||||
os.makedirs(os.path.join(remote_home, "memory"))
|
||||
Path(os.path.join(remote_home, "MEMORY.md")).write_text(
|
||||
"# 记忆\n")
|
||||
Path(os.path.join(remote_home, "memory", "2026-11-07.md")
|
||||
).write_text("- 喜欢冰美式\n")
|
||||
fake_ssh = os.path.join(tmp, "fake-ssh")
|
||||
Path(fake_ssh).write_text(
|
||||
"#!/bin/sh\n"
|
||||
"tar -czf - -C %s memory\n" % remote_parent)
|
||||
os.chmod(fake_ssh, 0o755)
|
||||
local = os.path.join(tmp, "local")
|
||||
result = pull_memory_home([fake_ssh], GUEST_MEMORY_HOME,
|
||||
local)
|
||||
self.assertEqual(result["files"], 2)
|
||||
self.assertIn("冰美式", Path(os.path.join(
|
||||
local, "memory", "memory", "2026-11-07.md")).read_text())
|
||||
|
||||
def test_pull_ssh_failure_raises(self):
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
fake_ssh = os.path.join(tmp, "fake-ssh")
|
||||
Path(fake_ssh).write_text("#!/bin/sh\nexit 3\n")
|
||||
os.chmod(fake_ssh, 0o755)
|
||||
with self.assertRaises(memory_files.MemoryError):
|
||||
pull_memory_home([fake_ssh], GUEST_MEMORY_HOME,
|
||||
os.path.join(tmp, "local"))
|
||||
|
||||
|
||||
class GuestScriptTest(unittest.TestCase):
|
||||
def test_list_read_append_forget_export(self):
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
home = os.path.join(tmp, "memory")
|
||||
self.assertEqual(
|
||||
CHV_MEMORY.cmd_list(home)["files"], ["MEMORY.md"])
|
||||
CHV_MEMORY.cmd_append(home, "MEMORY.md",
|
||||
["- 喜欢冰美式", "- 怕辣"])
|
||||
read = CHV_MEMORY.cmd_read(home, "MEMORY.md")
|
||||
self.assertIn("冰美式", read["text"])
|
||||
forgot = CHV_MEMORY.cmd_forget(home, "冰美式")
|
||||
self.assertEqual(forgot["removed"], 1)
|
||||
self.assertNotIn("冰美式",
|
||||
CHV_MEMORY.cmd_read(home, "MEMORY.md")["text"])
|
||||
dest = os.path.join(tmp, "export")
|
||||
exported = CHV_MEMORY.cmd_export(home, dest)
|
||||
self.assertEqual(exported["files"], 1)
|
||||
self.assertTrue(os.path.isfile(
|
||||
os.path.join(dest, "MEMORY.md")))
|
||||
|
||||
def test_main_list_prints_json(self):
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
home = os.path.join(tmp, "memory")
|
||||
buf = io.StringIO()
|
||||
with redirect_stdout(buf):
|
||||
rc = CHV_MEMORY.main(["--home", home, "list"])
|
||||
self.assertEqual(rc, 0)
|
||||
self.assertEqual(json.loads(buf.getvalue())["files"],
|
||||
["MEMORY.md"])
|
||||
|
||||
def test_read_traversal_rejected(self):
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
home = os.path.join(tmp, "memory")
|
||||
with self.assertRaises(SystemExit):
|
||||
CHV_MEMORY.cmd_read(home, "../escape.md")
|
||||
|
||||
def test_append_secret_refused(self):
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
home = os.path.join(tmp, "memory")
|
||||
with self.assertRaises(SystemExit):
|
||||
CHV_MEMORY.cmd_append(home, "MEMORY.md",
|
||||
["token ghp_" + "e" * 24])
|
||||
|
||||
def test_export_excludes_audit_log(self):
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
home = os.path.join(tmp, "memory")
|
||||
CHV_MEMORY._ensure(home)
|
||||
Path(os.path.join(home, "audit.jsonl")).write_text("{}\n")
|
||||
dest = os.path.join(tmp, "export")
|
||||
CHV_MEMORY.cmd_export(home, dest)
|
||||
self.assertFalse(os.path.exists(
|
||||
os.path.join(dest, "audit.jsonl")))
|
||||
|
||||
|
||||
class BuildIntegrationTest(unittest.TestCase):
|
||||
def test_build_sh_copies_chv_memory(self):
|
||||
text = (ROOT / "chv-guest" / "build.sh").read_text()
|
||||
self.assertIn("chv-memory.py", text)
|
||||
self.assertIn("/usr/local/bin/chv-memory", text)
|
||||
|
||||
def test_buildinfo_lists_chv_memory(self):
|
||||
self.assertIn("chv-memory", BUILDINFO.BASE_CONTENTS)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in new issue
Block a user