92 lines
3.4 KiB
Python
92 lines
3.4 KiB
Python
# ============================================================
|
||
# 审计 append-only 守护黄金测试（plan.md §3.6 / 矩阵 116 行剩余项）
|
||
# 覆盖：跨进程锁并发写不丢行、清理策略（保留窗口归档不删除证据）。
|
||
# ============================================================
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import threading
|
||
from pathlib import Path
|
||
|
||
from server.agent_core.audit_filelock import locked_append
|
||
from server.agent_core.audit_ledger import AnchorLedger
|
||
from server.agent_core.audit_mirror import AuditMirror
|
||
|
||
|
||
def _event(i: int) -> dict:
|
||
return {"id": i, "at": "2026-08-01T00:00:00+08:00", "actor": "tester",
|
||
"category": "TEST", "action": f"test.{i}", "target": {"type": "TEST"},
|
||
"power": "P0", "rationale": {}, "evidenceRefs": [], "beforeSnapshot": None,
|
||
"result": "SUCCESS", "prevHash": "GENESIS", "hash": "0" * 64}
|
||
|
||
|
||
def test_locked_append_concurrent_no_lost_lines(tmp_path: Path):
|
||
"""跨进程锁：4 线程×50 次并发 append 同一 JSONL，200 行全保留、无交错损坏。"""
|
||
target = tmp_path / "shared.jsonl"
|
||
|
||
def worker(n: int) -> None:
|
||
for i in range(50):
|
||
with locked_append(target) as fh:
|
||
fh.write(json.dumps({"n": n, "i": i}, sort_keys=True) + "\n")
|
||
|
||
threads = [threading.Thread(target=worker, args=(n,)) for n in range(4)]
|
||
for t in threads:
|
||
t.start()
|
||
for t in threads:
|
||
t.join()
|
||
lines = target.read_text(encoding="utf-8").splitlines()
|
||
assert len(lines) == 200, f"expected 200 lines, got {len(lines)}"
|
||
for ln in lines:
|
||
assert json.loads(ln), "行应为合法 JSON（无交错损坏）"
|
||
|
||
|
||
def test_ledger_archive_old_keeps_window_and_evidence(tmp_path: Path):
|
||
"""清理策略：超过保留窗口的旧锚定记录移入归档目录，主文件保留窗口；证据不丢失。"""
|
||
ledger = AnchorLedger("tenant-a", "world-1", ledger_dir=str(tmp_path))
|
||
evs = []
|
||
for i in range(5):
|
||
evs.append(_event(i + 1))
|
||
ledger.append(evs, reason="test")
|
||
assert ledger.status(evs)["records"] == 5
|
||
|
||
r = ledger.archive_old(keep_last=2)
|
||
assert r["archived"] == 3
|
||
assert r["kept"] == 2
|
||
assert len(ledger.read()) == 2
|
||
|
||
archive = Path(r["archive"])
|
||
assert archive.exists()
|
||
archived_lines = archive.read_text(encoding="utf-8").splitlines()
|
||
assert len(archived_lines) == 3
|
||
# 归档记录仍是合法 JSON（分级存储，非删除）
|
||
for ln in archived_lines:
|
||
rec = json.loads(ln)
|
||
assert rec.get("root") and rec.get("count")
|
||
|
||
|
||
def test_ledger_archive_under_window_is_noop(tmp_path: Path):
|
||
"""保留窗口内不归档（no-op），主文件不变。"""
|
||
ledger = AnchorLedger("tenant-a", "world-1", ledger_dir=str(tmp_path))
|
||
evs = []
|
||
for i in range(3):
|
||
evs.append(_event(i + 1))
|
||
ledger.append(evs, reason="test")
|
||
r = ledger.archive_old(keep_last=10)
|
||
assert r["archived"] == 0
|
||
assert r["kept"] == 3
|
||
assert len(ledger.read()) == 3
|
||
|
||
|
||
def test_mirror_append_uses_cross_process_lock(tmp_path: Path):
|
||
"""镜像 append 经 locked_append 落盘：并发下仍完整（回归保护）。"""
|
||
m = AuditMirror("tenant-a", "world-1", mirror_dir=str(tmp_path))
|
||
threads = [
|
||
threading.Thread(target=m.append_event, args=(_event(i + 1),))
|
||
for i in range(5)
|
||
]
|
||
for t in threads:
|
||
t.start()
|
||
for t in threads:
|
||
t.join()
|
||
assert m.count() == 5
|