cybersec-audit-trail-stack / runtime_tests_test_audit_log.py
HirModel's picture
Upload 21 files
5729d24 verified
Raw
History Blame Contribute Delete
14.3 kB
"""
Tests for src/primordial_os/audit_log.py — Phase 6 Audit Log
"""
import dataclasses
import inspect
import pytest
from primordial_os.audit_log import (
AuditEvent,
append_to_audit_chain,
compute_event_hash,
create_audit_event,
create_runtime_audit_event,
verify_audit_chain,
)
from primordial_os.hir_kernel import HIRInput
from primordial_os.oam_detector import OAMSignals
from primordial_os.runtime import run_evaluation
# ---------------------------------------------------------------------------
# Fixtures
# ---------------------------------------------------------------------------
def _green_decision():
return run_evaluation(
HIRInput(0.9, 0.9, 0.9, 0.9, pressure=0.0),
OAMSignals(
certainty_level=0.5,
decision_made_for_user=False,
context_preservation=0.9,
dignity_preserved=0.9,
identity_assigned=False,
session_load_ratio=0.2,
uncertainty_disclosed=True,
autonomy_scope=0.2,
consent_present=True,
clinical_language_present=False,
memory_accessed_without_consent=False,
end_of_life_decision_attempted=False,
human_override_available=True,
),
)
def _red_decision():
return run_evaluation(
HIRInput(0.9, 0.9, 0.9, 0.9, pressure=0.0),
OAMSignals(
certainty_level=0.5,
decision_made_for_user=True, # agency_outsourcing → RED
context_preservation=0.9,
dignity_preserved=0.9,
identity_assigned=False,
session_load_ratio=0.2,
uncertainty_disclosed=True,
autonomy_scope=0.2,
consent_present=True,
clinical_language_present=False,
memory_accessed_without_consent=False,
end_of_life_decision_attempted=False,
human_override_available=True,
),
)
def _fixed_hash_args(**overrides) -> dict:
defaults = dict(
timestamp="2026-05-02T12:00:00+00:00",
event_type="runtime_evaluation",
final_gate_state="GREEN",
hard_stop=False,
resonance=0.9,
pressure_adjusted_stability=0.9,
max_oam_severity=0.0,
summary="Gate: GREEN. Resonance=0.900. PAS=0.900. OAM severity=0.000.",
previous_hash=None,
)
defaults.update(overrides)
return defaults
# ---------------------------------------------------------------------------
# compute_event_hash
# ---------------------------------------------------------------------------
class TestComputeEventHash:
def test_returns_64_char_hex_string(self):
h = compute_event_hash(**_fixed_hash_args())
assert len(h) == 64
assert all(c in "0123456789abcdef" for c in h)
def test_deterministic_same_inputs(self):
args = _fixed_hash_args()
assert compute_event_hash(**args) == compute_event_hash(**args)
def test_different_summary_produces_different_hash(self):
h1 = compute_event_hash(**_fixed_hash_args(summary="Gate: GREEN."))
h2 = compute_event_hash(**_fixed_hash_args(summary="Gate: RED."))
assert h1 != h2
def test_different_gate_state_produces_different_hash(self):
h1 = compute_event_hash(**_fixed_hash_args(final_gate_state="GREEN"))
h2 = compute_event_hash(**_fixed_hash_args(final_gate_state="RED"))
assert h1 != h2
def test_different_resonance_produces_different_hash(self):
h1 = compute_event_hash(**_fixed_hash_args(resonance=0.9))
h2 = compute_event_hash(**_fixed_hash_args(resonance=0.5))
assert h1 != h2
def test_previous_hash_affects_current_hash(self):
h_no_prev = compute_event_hash(**_fixed_hash_args(previous_hash=None))
h_with_prev = compute_event_hash(**_fixed_hash_args(previous_hash="abc123"))
assert h_no_prev != h_with_prev
def test_different_previous_hashes_produce_different_hashes(self):
h1 = compute_event_hash(**_fixed_hash_args(previous_hash="aaa"))
h2 = compute_event_hash(**_fixed_hash_args(previous_hash="bbb"))
assert h1 != h2
def test_hard_stop_bool_affects_hash(self):
h_false = compute_event_hash(**_fixed_hash_args(hard_stop=False))
h_true = compute_event_hash(**_fixed_hash_args(hard_stop=True))
assert h_false != h_true
def test_timestamp_affects_hash(self):
h1 = compute_event_hash(**_fixed_hash_args(timestamp="2026-05-02T12:00:00+00:00"))
h2 = compute_event_hash(**_fixed_hash_args(timestamp="2026-05-02T13:00:00+00:00"))
assert h1 != h2
# ---------------------------------------------------------------------------
# AuditEvent structure
# ---------------------------------------------------------------------------
class TestAuditEventStructure:
def test_audit_event_is_frozen(self):
event = create_audit_event(_green_decision())
with pytest.raises((AttributeError, TypeError)):
event.summary = "tampered" # type: ignore[misc]
def test_create_returns_audit_event_instance(self):
event = create_audit_event(_green_decision())
assert isinstance(event, AuditEvent)
def test_current_hash_is_64_chars(self):
event = create_audit_event(_green_decision())
assert len(event.current_hash) == 64
def test_previous_hash_none_for_first_event(self):
event = create_audit_event(_green_decision())
assert event.previous_hash is None
def test_previous_hash_propagated_when_supplied(self):
sentinel = "a" * 64
event = create_audit_event(_green_decision(), previous_hash=sentinel)
assert event.previous_hash == sentinel
def test_gate_state_stored_as_string(self):
event = create_audit_event(_green_decision())
assert event.final_gate_state == "GREEN"
def test_hard_stop_matches_decision(self):
green = create_audit_event(_green_decision())
red = create_audit_event(_red_decision())
assert green.hard_stop is False
assert red.hard_stop is True
def test_resonance_matches_decision(self):
decision = _green_decision()
event = create_audit_event(decision)
assert event.resonance == pytest.approx(decision.hir_result.resonance)
def test_event_type_default(self):
event = create_audit_event(_green_decision())
assert event.event_type == "runtime_evaluation"
def test_event_type_custom(self):
event = create_audit_event(_green_decision(), event_type="manual_review")
assert event.event_type == "manual_review"
def test_timestamp_is_non_empty_string(self):
event = create_audit_event(_green_decision())
assert isinstance(event.timestamp, str)
assert len(event.timestamp) > 0
# ---------------------------------------------------------------------------
# verify_audit_chain
# ---------------------------------------------------------------------------
class TestVerifyAuditChain:
def test_empty_chain_is_valid(self):
assert verify_audit_chain([]) is True
def test_single_event_chain_is_valid(self):
event = create_audit_event(_green_decision())
assert verify_audit_chain([event]) is True
def test_two_event_chain_is_valid(self):
e1 = create_audit_event(_green_decision())
e2 = create_audit_event(_red_decision(), previous_hash=e1.current_hash)
assert verify_audit_chain([e1, e2]) is True
def test_three_event_chain_is_valid(self):
e1 = create_audit_event(_green_decision())
e2 = create_audit_event(_red_decision(), previous_hash=e1.current_hash)
e3 = create_audit_event(_green_decision(), previous_hash=e2.current_hash)
assert verify_audit_chain([e1, e2, e3]) is True
def test_tampered_summary_breaks_chain(self):
event = create_audit_event(_green_decision())
tampered = dataclasses.replace(event, summary="TAMPERED")
assert verify_audit_chain([tampered]) is False
def test_tampered_gate_state_breaks_chain(self):
event = create_audit_event(_green_decision())
tampered = dataclasses.replace(event, final_gate_state="RED")
assert verify_audit_chain([tampered]) is False
def test_tampered_resonance_breaks_chain(self):
event = create_audit_event(_green_decision())
tampered = dataclasses.replace(event, resonance=0.0)
assert verify_audit_chain([tampered]) is False
def test_tampered_hard_stop_breaks_chain(self):
event = create_audit_event(_green_decision())
tampered = dataclasses.replace(event, hard_stop=True)
assert verify_audit_chain([tampered]) is False
def test_tampered_previous_hash_breaks_chain(self):
e1 = create_audit_event(_green_decision())
e2 = create_audit_event(_red_decision(), previous_hash=e1.current_hash)
# Detach e2 from e1 by altering its previous_hash field
tampered_e2 = dataclasses.replace(e2, previous_hash="wrong_hash")
assert verify_audit_chain([e1, tampered_e2]) is False
def test_broken_chain_link_breaks_verification(self):
e1 = create_audit_event(_green_decision())
# e2 has wrong previous_hash — not linked to e1
e2 = create_audit_event(_red_decision(), previous_hash=None)
assert verify_audit_chain([e1, e2]) is False
def test_middle_event_tamper_breaks_chain(self):
e1 = create_audit_event(_green_decision())
e2 = create_audit_event(_red_decision(), previous_hash=e1.current_hash)
e3 = create_audit_event(_green_decision(), previous_hash=e2.current_hash)
tampered_e2 = dataclasses.replace(e2, summary="TAMPERED")
assert verify_audit_chain([e1, tampered_e2, e3]) is False
def test_reordered_chain_breaks_verification(self):
e1 = create_audit_event(_green_decision())
e2 = create_audit_event(_red_decision(), previous_hash=e1.current_hash)
# Swap order — e2 comes before e1
assert verify_audit_chain([e2, e1]) is False
# ---------------------------------------------------------------------------
# create_runtime_audit_event
# ---------------------------------------------------------------------------
class TestCreateRuntimeAuditEvent:
def test_returns_audit_event_instance(self):
event = create_runtime_audit_event(_green_decision())
assert isinstance(event, AuditEvent)
def test_previous_hash_is_none(self):
event = create_runtime_audit_event(_green_decision())
assert event.previous_hash is None
def test_default_event_type_is_runtime_evaluation(self):
event = create_runtime_audit_event(_green_decision())
assert event.event_type == "runtime_evaluation"
def test_custom_event_type_propagated(self):
event = create_runtime_audit_event(_green_decision(), event_type="test_event")
assert event.event_type == "test_event"
def test_hash_is_64_chars(self):
event = create_runtime_audit_event(_green_decision())
assert len(event.current_hash) == 64
def test_hash_verifies_as_single_chain(self):
event = create_runtime_audit_event(_green_decision())
assert verify_audit_chain([event]) is True
# ---------------------------------------------------------------------------
# append_to_audit_chain
# ---------------------------------------------------------------------------
class TestAppendToAuditChain:
def test_append_to_empty_tuple_sets_previous_hash_none(self):
chain = append_to_audit_chain((), _green_decision())
assert chain[0].previous_hash is None
def test_append_to_empty_list_sets_previous_hash_none(self):
chain = append_to_audit_chain([], _green_decision())
assert chain[0].previous_hash is None
def test_append_to_nonempty_chain_links_previous_hash(self):
chain = append_to_audit_chain((), _green_decision())
chain2 = append_to_audit_chain(chain, _red_decision())
assert chain2[1].previous_hash == chain[0].current_hash
def test_append_does_not_mutate_original_list(self):
original: list = []
append_to_audit_chain(original, _green_decision())
assert len(original) == 0
def test_append_does_not_mutate_original_tuple(self):
first = append_to_audit_chain((), _green_decision())
_ = append_to_audit_chain(first, _red_decision())
assert len(first) == 1
def test_result_is_tuple(self):
chain = append_to_audit_chain((), _green_decision())
assert isinstance(chain, tuple)
def test_chain_length_grows_by_one_per_append(self):
chain = append_to_audit_chain((), _green_decision())
chain2 = append_to_audit_chain(chain, _red_decision())
assert len(chain) == 1
assert len(chain2) == 2
def test_single_append_chain_verifies_true(self):
chain = append_to_audit_chain((), _green_decision())
assert verify_audit_chain(chain) is True
def test_double_append_chain_verifies_true(self):
chain = append_to_audit_chain((), _green_decision())
chain = append_to_audit_chain(chain, _red_decision())
assert verify_audit_chain(chain) is True
def test_triple_append_chain_verifies_true(self):
chain = append_to_audit_chain((), _green_decision())
chain = append_to_audit_chain(chain, _red_decision())
chain = append_to_audit_chain(chain, _green_decision())
assert verify_audit_chain(chain) is True
def test_tampered_appended_event_breaks_verification(self):
chain = append_to_audit_chain((), _green_decision())
tampered = dataclasses.replace(chain[0], summary="TAMPERED")
assert verify_audit_chain((tampered,)) is False
# ---------------------------------------------------------------------------
# Source hygiene
# ---------------------------------------------------------------------------
class TestSourceHygiene:
def test_no_coherence_terminology_in_source(self):
import primordial_os.audit_log as mod
source = inspect.getsource(mod)
assert "coherence" not in source.lower(), (
"Banned term 'coherence' found in audit_log source"
)