Files
phase-memory/tests/test_audit_operations.py

73 lines
2.6 KiB
Python

from datetime import datetime, timezone
from phase_memory.lifecycle import plan_compaction
from phase_memory.models import MemoryNode
from phase_memory.runtime import PhaseMemoryRuntime
def test_audit_export_traces_policy_denial_package_compile_and_review_apply() -> None:
runtime = PhaseMemoryRuntime()
graph = {
"schema_version": "markitect.memory.graph.v1",
"id": "audit.graph",
"nodes": [
{
"id": "node.public",
"kind": "knowledge",
"text": "Public node.",
"policy": {"labels": ["public"], "trust_zone": "local"},
},
{
"id": "node.secret",
"kind": "knowledge",
"text": "Secret node.",
"policy": {"labels": ["restricted"], "secret": True, "trust_zone": "local"},
},
],
"edges": [],
"events": [],
}
activation = runtime.plan_activation(
graph,
max_items=3,
max_tokens=30,
policy_context={"denied_labels": ["restricted"], "secrets_allowed": False, "trust_zone": "local"},
)
runtime.compile_package(activation["data"]["package_request"]["selection"])
node = runtime.graph_store.save_node(MemoryNode("audit.review", "episode", "Review text"))
compact = plan_compaction([node])
runtime.apply_lifecycle_actions([compact])
runtime.apply_lifecycle_actions([compact], approval_marker="review:audit")
export = runtime.export_audit_events()
operations = [event["operation"] for event in export["batch"]["events"]]
assert export["valid"] is True
assert "graph.activation.plan" in operations
assert "package.compile" in operations
assert operations.count("lifecycle.apply") == 2
assert activation["data"]["policy_denials"][0]["id"] == "node.secret"
def test_audit_retention_plan_identifies_eligible_records() -> None:
runtime = PhaseMemoryRuntime()
runtime.audit_sink.record(
{
"schema_version": "phase_memory.audit.event.v1",
"operation_id": "op:old",
"operation": "manual",
"timestamp": "2026-01-01T00:00:00+00:00",
"subject": {"kind": "audit_events", "id": "old"},
"source": {"ref": "test"},
"dry_run": True,
"allowed": True,
}
)
plan = runtime.audit_retention_plan(retention_days=30, now=datetime(2026, 5, 19, tzinfo=timezone.utc))
assert plan["valid"] is True
assert plan["plan"]["eligible_operation_ids"] == ["op:old"]
assert plan["plan"]["eligible_count"] == 1