34 lines
1.2 KiB
Python
34 lines
1.2 KiB
Python
from edgemind.sidecar import AuditLogStore, merge_plan_deltas
|
|
|
|
|
|
def test_audit_log_append_and_iterate():
|
|
s = AuditLogStore(":memory:")
|
|
id1 = s.append("agentA", "start", {"detail": 1})
|
|
id2 = s.append("agentB", "update", {"detail": 2})
|
|
entries = s.iter_entries()
|
|
assert len(entries) == 2
|
|
assert entries[0]["id"] == id1
|
|
assert entries[1]["id"] == id2
|
|
assert s.verify_chain() is True
|
|
|
|
|
|
def test_audit_log_chain_tamper_detection():
|
|
s = AuditLogStore(":memory:")
|
|
s.append("a", "one", {})
|
|
s.append("b", "two", {})
|
|
# Tamper directly in DB
|
|
cur = s.conn.cursor()
|
|
cur.execute("UPDATE audit_log SET entry_json = ? WHERE id = 1", ('{"actor":"x"}',))
|
|
s.conn.commit()
|
|
assert s.verify_chain() is False
|
|
|
|
|
|
def test_merge_plan_deltas_simple():
|
|
d1 = {"id": "d1", "patches": [{"op_id": "a", "op": "set", "k": "x", "v": 1}], "timestamp": "2021-01-01T00:00:00"}
|
|
d2 = {"id": "d2", "patches": [{"op_id": "b", "op": "set", "k": "y", "v": 2}], "timestamp": "2021-01-01T00:00:01"}
|
|
merged = merge_plan_deltas([d2, d1])
|
|
# Two patches, deterministic order by timestamp
|
|
assert len(merged["patches"]) == 2
|
|
assert merged["patches"][0]["op_id"] == "a"
|
|
assert merged["patches"][1]["op_id"] == "b"
|