""" Tests for src/primordial_os/audit_store.py — Phase 10 Audit Store """ import inspect import json from pathlib import Path import pytest from primordial_os.audit_log import ( AuditEvent, append_to_audit_chain, create_audit_event, ) from primordial_os.audit_store import ( append_audit_event_jsonl, audit_event_from_dict, audit_event_to_dict, read_audit_events_jsonl, verify_audit_jsonl, ) from primordial_os.hir_kernel import HIRInput from primordial_os.oam_detector import OAMSignals from primordial_os.runtime import run_evaluation # --------------------------------------------------------------------------- # Shared 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, 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, ), ) # --------------------------------------------------------------------------- # audit_event_to_dict # --------------------------------------------------------------------------- class TestAuditEventToDict: def test_returns_dict(self): event = create_audit_event(_green_decision()) assert isinstance(audit_event_to_dict(event), dict) def test_all_expected_keys_present(self): event = create_audit_event(_green_decision()) d = audit_event_to_dict(event) expected = { "timestamp", "event_type", "final_gate_state", "hard_stop", "resonance", "pressure_adjusted_stability", "max_oam_severity", "summary", "previous_hash", "current_hash", } assert set(d.keys()) == expected def test_no_extra_keys(self): event = create_audit_event(_green_decision()) d = audit_event_to_dict(event) assert len(d) == 10 def test_values_match_event_fields(self): event = create_audit_event(_green_decision()) d = audit_event_to_dict(event) assert d["timestamp"] == event.timestamp assert d["event_type"] == event.event_type assert d["final_gate_state"] == event.final_gate_state assert d["hard_stop"] == event.hard_stop assert d["resonance"] == event.resonance assert d["pressure_adjusted_stability"] == event.pressure_adjusted_stability assert d["max_oam_severity"] == event.max_oam_severity assert d["summary"] == event.summary assert d["current_hash"] == event.current_hash def test_previous_hash_none_preserved(self): event = create_audit_event(_green_decision()) d = audit_event_to_dict(event) assert d["previous_hash"] is None def test_previous_hash_string_preserved(self): sentinel = "b" * 64 event = create_audit_event(_green_decision(), previous_hash=sentinel) d = audit_event_to_dict(event) assert d["previous_hash"] == sentinel def test_no_raw_user_data_keys(self): event = create_audit_event(_green_decision()) d = audit_event_to_dict(event) forbidden = {"prompt", "user_input", "phi", "pii", "clinical_data", "raw_input"} assert not (forbidden & set(d.keys())) def test_is_json_serializable(self): event = create_audit_event(_green_decision()) d = audit_event_to_dict(event) serialized = json.dumps(d) assert isinstance(serialized, str) # --------------------------------------------------------------------------- # audit_event_from_dict # --------------------------------------------------------------------------- class TestAuditEventFromDict: def test_returns_audit_event(self): event = create_audit_event(_green_decision()) result = audit_event_from_dict(audit_event_to_dict(event)) assert isinstance(result, AuditEvent) def test_round_trip_equality(self): event = create_audit_event(_green_decision()) result = audit_event_from_dict(audit_event_to_dict(event)) assert result == event def test_round_trip_with_previous_hash(self): e1 = create_audit_event(_green_decision()) e2 = create_audit_event(_red_decision(), previous_hash=e1.current_hash) result = audit_event_from_dict(audit_event_to_dict(e2)) assert result == e2 def test_result_is_frozen(self): event = create_audit_event(_green_decision()) result = audit_event_from_dict(audit_event_to_dict(event)) with pytest.raises((AttributeError, TypeError)): result.summary = "tampered" # type: ignore[misc] def test_round_trip_hash_still_verifies(self): from primordial_os.audit_log import verify_audit_chain e1 = create_audit_event(_green_decision()) e2 = create_audit_event(_red_decision(), previous_hash=e1.current_hash) r1 = audit_event_from_dict(audit_event_to_dict(e1)) r2 = audit_event_from_dict(audit_event_to_dict(e2)) assert verify_audit_chain([r1, r2]) is True # --------------------------------------------------------------------------- # append_audit_event_jsonl # --------------------------------------------------------------------------- class TestAppendAuditEventJsonl: def test_creates_file_when_absent(self, tmp_path): p = tmp_path / "audit.jsonl" event = create_audit_event(_green_decision()) append_audit_event_jsonl(p, event) assert p.exists() def test_single_append_produces_one_line(self, tmp_path): p = tmp_path / "audit.jsonl" event = create_audit_event(_green_decision()) append_audit_event_jsonl(p, event) non_empty = [ln for ln in p.read_text("utf-8").splitlines() if ln.strip()] assert len(non_empty) == 1 def test_two_appends_produce_two_lines(self, tmp_path): p = tmp_path / "audit.jsonl" e1 = create_audit_event(_green_decision()) e2 = create_audit_event(_red_decision(), previous_hash=e1.current_hash) append_audit_event_jsonl(p, e1) append_audit_event_jsonl(p, e2) non_empty = [ln for ln in p.read_text("utf-8").splitlines() if ln.strip()] assert len(non_empty) == 2 def test_five_appends_produce_five_lines(self, tmp_path): p = tmp_path / "audit.jsonl" prev = None for _ in range(5): e = create_audit_event(_green_decision(), previous_hash=prev) append_audit_event_jsonl(p, e) prev = e.current_hash non_empty = [ln for ln in p.read_text("utf-8").splitlines() if ln.strip()] assert len(non_empty) == 5 def test_each_line_is_valid_json(self, tmp_path): p = tmp_path / "audit.jsonl" e1 = create_audit_event(_green_decision()) e2 = create_audit_event(_red_decision(), previous_hash=e1.current_hash) append_audit_event_jsonl(p, e1) append_audit_event_jsonl(p, e2) for raw_line in p.read_text("utf-8").splitlines(): if raw_line.strip(): assert isinstance(json.loads(raw_line), dict) def test_file_is_valid_utf8(self, tmp_path): p = tmp_path / "audit.jsonl" event = create_audit_event(_green_decision()) append_audit_event_jsonl(p, event) p.read_bytes().decode("utf-8") def test_existing_content_preserved_on_second_append(self, tmp_path): p = tmp_path / "audit.jsonl" e1 = create_audit_event(_green_decision()) append_audit_event_jsonl(p, e1) first_content = p.read_text("utf-8") e2 = create_audit_event(_red_decision(), previous_hash=e1.current_hash) append_audit_event_jsonl(p, e2) second_content = p.read_text("utf-8") assert second_content.startswith(first_content) def test_hidden_file_path_rejected(self, tmp_path): p = tmp_path / ".hidden.jsonl" event = create_audit_event(_green_decision()) with pytest.raises(ValueError): append_audit_event_jsonl(p, event) def test_missing_parent_directory_rejected(self, tmp_path): p = tmp_path / "no_such_dir" / "audit.jsonl" event = create_audit_event(_green_decision()) with pytest.raises(FileNotFoundError): append_audit_event_jsonl(p, event) def test_no_forbidden_field_names_written(self, tmp_path): p = tmp_path / "audit.jsonl" event = create_audit_event(_green_decision()) append_audit_event_jsonl(p, event) content = p.read_text("utf-8") for term in ("prompt", "user_input", "phi", "pii", "clinical_data"): assert term not in content.lower() # --------------------------------------------------------------------------- # read_audit_events_jsonl # --------------------------------------------------------------------------- class TestReadAuditEventsJsonl: def test_returns_tuple(self, tmp_path): p = tmp_path / "audit.jsonl" event = create_audit_event(_green_decision()) append_audit_event_jsonl(p, event) result = read_audit_events_jsonl(p) assert isinstance(result, tuple) def test_single_event_round_trip(self, tmp_path): p = tmp_path / "audit.jsonl" event = create_audit_event(_green_decision()) append_audit_event_jsonl(p, event) result = read_audit_events_jsonl(p) assert len(result) == 1 assert result[0] == event def test_multiple_events_round_trip(self, tmp_path): p = tmp_path / "audit.jsonl" e1 = create_audit_event(_green_decision()) e2 = create_audit_event(_red_decision(), previous_hash=e1.current_hash) append_audit_event_jsonl(p, e1) append_audit_event_jsonl(p, e2) result = read_audit_events_jsonl(p) assert len(result) == 2 assert result[0] == e1 assert result[1] == e2 def test_order_preserved(self, tmp_path): p = tmp_path / "audit.jsonl" written: list[AuditEvent] = [] prev = None for _ in range(4): e = create_audit_event(_green_decision(), previous_hash=prev) append_audit_event_jsonl(p, e) written.append(e) prev = e.current_hash result = read_audit_events_jsonl(p) assert list(result) == written def test_all_elements_are_audit_event_instances(self, tmp_path): p = tmp_path / "audit.jsonl" e1 = create_audit_event(_green_decision()) e2 = create_audit_event(_red_decision(), previous_hash=e1.current_hash) append_audit_event_jsonl(p, e1) append_audit_event_jsonl(p, e2) for item in read_audit_events_jsonl(p): assert isinstance(item, AuditEvent) def test_missing_file_raises_file_not_found(self, tmp_path): p = tmp_path / "missing.jsonl" with pytest.raises(FileNotFoundError): read_audit_events_jsonl(p) def test_empty_file_returns_empty_tuple(self, tmp_path): p = tmp_path / "audit.jsonl" p.write_text("", encoding="utf-8") result = read_audit_events_jsonl(p) assert result == () # --------------------------------------------------------------------------- # verify_audit_jsonl # --------------------------------------------------------------------------- class TestVerifyAuditJsonl: def test_valid_single_event_returns_true(self, tmp_path): p = tmp_path / "audit.jsonl" event = create_audit_event(_green_decision()) append_audit_event_jsonl(p, event) assert verify_audit_jsonl(p) is True def test_valid_three_event_chain_returns_true(self, tmp_path): p = tmp_path / "audit.jsonl" chain = append_to_audit_chain((), _green_decision()) chain = append_to_audit_chain(chain, _red_decision()) chain = append_to_audit_chain(chain, _green_decision()) for event in chain: append_audit_event_jsonl(p, event) assert verify_audit_jsonl(p) is True def test_missing_file_returns_false(self, tmp_path): p = tmp_path / "missing.jsonl" assert verify_audit_jsonl(p) is False def test_tampered_hash_field_returns_false(self, tmp_path): p = tmp_path / "audit.jsonl" e1 = create_audit_event(_green_decision()) e2 = create_audit_event(_red_decision(), previous_hash=e1.current_hash) append_audit_event_jsonl(p, e1) append_audit_event_jsonl(p, e2) lines = p.read_text("utf-8").splitlines() data = json.loads(lines[0]) data["final_gate_state"] = "RED" lines[0] = json.dumps(data, sort_keys=True, separators=(",", ":")) p.write_text("\n".join(lines) + "\n", encoding="utf-8") assert verify_audit_jsonl(p) is False def test_broken_chain_link_returns_false(self, tmp_path): p = tmp_path / "audit.jsonl" e1 = create_audit_event(_green_decision()) e2 = create_audit_event(_red_decision(), previous_hash=None) append_audit_event_jsonl(p, e1) append_audit_event_jsonl(p, e2) assert verify_audit_jsonl(p) is False def test_tampered_resonance_returns_false(self, tmp_path): p = tmp_path / "audit.jsonl" event = create_audit_event(_green_decision()) append_audit_event_jsonl(p, event) lines = p.read_text("utf-8").splitlines() data = json.loads(lines[0]) data["resonance"] = 0.0 lines[0] = json.dumps(data, sort_keys=True, separators=(",", ":")) p.write_text("\n".join(lines) + "\n", encoding="utf-8") assert verify_audit_jsonl(p) is False # --------------------------------------------------------------------------- # Source hygiene # --------------------------------------------------------------------------- class TestSourceHygiene: def test_no_coherence_terminology_in_source(self): import primordial_os.audit_store as mod source = inspect.getsource(mod) assert "coherence" not in source.lower(), ( "Banned term 'coherence' found in audit_store source" ) def test_no_forbidden_clinical_functionality_terms_in_source(self): import primordial_os.audit_store as mod source = inspect.getsource(mod) for term in ("diagnos", "therapeutic", "clinical_decision", "prescri"): assert term not in source.lower(), ( f"Forbidden clinical-functionality term '{term}' found in audit_store source" )