generated from coulomb/repo-seed
128 lines
4.7 KiB
Python
128 lines
4.7 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
|
|
|
|
|
|
def test_audit_retention_apply_prunes_eligible_records_and_records_apply() -> 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,
|
|
}
|
|
)
|
|
runtime.audit_sink.record(
|
|
{
|
|
"schema_version": "phase_memory.audit.event.v1",
|
|
"operation_id": "op:new",
|
|
"operation": "manual",
|
|
"timestamp": "2026-05-18T00:00:00+00:00",
|
|
"subject": {"kind": "audit_events", "id": "new"},
|
|
"source": {"ref": "test"},
|
|
"dry_run": True,
|
|
"allowed": True,
|
|
}
|
|
)
|
|
|
|
plan = runtime.audit_retention_plan(retention_days=30, now=datetime(2026, 5, 19, tzinfo=timezone.utc))
|
|
applied = runtime.apply_audit_retention(plan["plan"])
|
|
remaining_ids = [event["operation_id"] for event in runtime.audit_sink.query()]
|
|
|
|
assert applied["valid"] is True
|
|
assert applied["result"]["pruned_operation_ids"] == ["op:old"]
|
|
assert "op:old" not in remaining_ids
|
|
assert "op:new" in remaining_ids
|
|
assert any(event["operation"] == "audit.retention.apply" for event in runtime.audit_sink.query())
|
|
|
|
|
|
def test_audit_retention_apply_noop_and_unsupported_paths() -> None:
|
|
runtime = PhaseMemoryRuntime()
|
|
noop = runtime.apply_audit_retention(retention_days=30, now=datetime(2026, 5, 19, tzinfo=timezone.utc))
|
|
assert noop["valid"] is True
|
|
assert noop["result"]["changed"] is False
|
|
|
|
class UnsupportedAuditSink:
|
|
def record(self, event):
|
|
return {"recorded": True, "event": event}
|
|
|
|
unsupported = PhaseMemoryRuntime(audit_sink=UnsupportedAuditSink())
|
|
result = unsupported.apply_audit_retention(retention_days=30)
|
|
|
|
assert result["valid"] is False
|
|
assert result["diagnostics"][0]["code"] == "audit_retention_apply_unsupported"
|