host
29fa0c8813
- 每 VM 一个 SQLite 状态库（cloud/chv/app_state.py）： state_dir/<vm>/app-state.sqlite（0600，WAL），表 tasks/goals/ sessions/audit（只追加）/meta；写操作过无凭证守卫（键名 + 值模式）；备份用 sqlite3 在线备份 + manifest（sha256），恢复 先验 manifest、先留 pre-restore 副本；DDL 用 SQLite 与 Postgres 共通受限子集，换 Postgres = 换连接工厂。 - 记忆文件接线（cloud/chv/memory_files.py）：宿主镜像读/列/ 写/追加（路径穿越守卫、写审计）；MemorySystemClient 调真实 记忆系统 CLI（search/get/explain/forget-plan/forget-confirm/ backup/restore/export），不重造引擎；forget = plan → confirm → 本地清行 → search 验证查不到；pull_memory_home 经 SSH + tar 拉取客机记忆家目录。 - 客内脚本 chv-guest/chv-memory.py（纯标准库，list/read/ append/forget/export），家目录 /home/chv/memory（workspace 之外，不同步到 Gitea）；build.sh 拷入 + buildinfo 登记。 - 新测试 cloud/tests/test_chv_muse_p7.py 46 项全过（含真 CLI write/search/forget/backup/restore 实测、假 ssh 拉取、否定 测试）；全量 cloud 套件 1913 项 OK（skipped=8）。 - 存储选型：SQLite 不是 Postgres（见差距表 §4；轻量 VM 跑 Postgres server 太重，宿主共享 Postgres 破坏每用户 VM 隔离）；忘记是尽力删除 + 验证，不是密码学擦除。 - 真机待验证（未做，等测试机空闲）：见 docs/chv-muse-p7-gap.md §6 清单
503 lines
22 KiB
Python
503 lines
22 KiB
Python
"""对齐 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()
|