runtime_tests_test_audit_store.py

"""
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"
            )