File size: 14,314 Bytes
5729d24 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 343 344 345 346 347 348 349 350 351 352 353 354 355 356 357 358 359 360 361 | """
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"
)
|