bikoret/azazel_processor/test/test_protocol_restriction.py

"""
Tests for the protocol envelope (property 8) — protocol-restricted
access. Vayikra 17:7: lo yizbechu od et zivchehem la'sa'irim — the
same destination outside the Yom Kippur envelope is avodah zarah.

These tests deliberately do NOT auto-open the envelope (unlike
test_azazel.py). They exercise the gate directly: verify dispatch
refuses outside the envelope, accepts inside the Yom Kippur window,
accepts inside an authorize_disposal context, and that the
authorization scope is per-thread and properly nested.
"""

import pathlib
import tempfile
import threading
from datetime import date, timedelta

import pytest

from azazel_processor.azazel import AzazelInput, dispatch_for_disposal
from azazel_processor.protocol import (
    ProtocolViolation,
    assert_protocol_envelope_active,
    authorize_disposal,
    clear_yom_kippur_window,
    has_explicit_disposal_authorization,
    is_yom_kippur_protocol_active,
    set_yom_kippur_window,
)
from azazel_processor.shaliach import IshItti
from azazel_processor.vidui import confess_empty_vital


@pytest.fixture(autouse=True)
def _isolate_protocol_state():
    """Ensure each test starts with no Yom Kippur window and no
    leftover thread-local authorization. Module-level state in
    protocol.py persists across tests if we don't reset."""
    clear_yom_kippur_window()
    yield
    clear_yom_kippur_window()


@pytest.fixture
def tmp_in_flight():
    with tempfile.TemporaryDirectory(prefix="azazel-protocol-test-") as d:
        yield pathlib.Path(d)


@pytest.fixture
def active_handler():
    h = IshItti("aharon")
    h.activate()
    yield h
    if h.is_active():
        h.tevilah()


def _input(case_ref="proto-001"):
    return AzazelInput(
        case_ref=case_ref, payload={}, classification="empty_vital"
    )


def _vidui():
    return confess_empty_vital("placeholder", confessor="kohen-gadol")


# ============================================================
# Default state: refused
# ============================================================


def test_default_state_refuses_dispatch(active_handler, tmp_in_flight):
    """No Yom Kippur window configured + no authorization → ProtocolViolation."""
    with pytest.raises(ProtocolViolation) as exc:
        dispatch_for_disposal(
            _input(),
            handler=active_handler,
            confess_with=_vidui(),
            authorized_by="kohen-gadol",
            in_flight_dir=tmp_in_flight,
        )
    msg = str(exc.value)
    # Error message names both legitimate paths
    assert "yom kippur" in msg.lower() or "set_yom_kippur_window" in msg
    assert "authorize_disposal" in msg


def test_default_state_assert_raises():
    with pytest.raises(ProtocolViolation):
        assert_protocol_envelope_active()


def test_default_state_neither_predicate_active():
    assert is_yom_kippur_protocol_active() is False
    assert has_explicit_disposal_authorization() is False


# ============================================================
# Yom Kippur window
# ============================================================


def test_yom_kippur_window_today_inside_active():
    today = date(2026, 9, 21)
    set_yom_kippur_window(today - timedelta(days=3), today + timedelta(days=3))
    assert is_yom_kippur_protocol_active(today=today) is True


def test_yom_kippur_window_today_outside_inactive():
    set_yom_kippur_window(date(2026, 9, 1), date(2026, 9, 5))
    assert is_yom_kippur_protocol_active(today=date(2026, 12, 1)) is False


def test_yom_kippur_window_inverted_clears():
    """Setting a window where start > end clears it (no valid window)."""
    set_yom_kippur_window(date(2026, 9, 5), date(2026, 9, 1))
    assert is_yom_kippur_protocol_active(today=date(2026, 9, 3)) is False


def test_yom_kippur_window_inclusive_at_boundaries():
    set_yom_kippur_window(date(2026, 9, 1), date(2026, 9, 5))
    assert is_yom_kippur_protocol_active(today=date(2026, 9, 1)) is True
    assert is_yom_kippur_protocol_active(today=date(2026, 9, 5)) is True
    assert is_yom_kippur_protocol_active(today=date(2026, 8, 31)) is False
    assert is_yom_kippur_protocol_active(today=date(2026, 9, 6)) is False


def test_clear_yom_kippur_window_disables():
    set_yom_kippur_window(date(2026, 1, 1), date(2026, 12, 31))
    assert is_yom_kippur_protocol_active(today=date(2026, 6, 1)) is True
    clear_yom_kippur_window()
    assert is_yom_kippur_protocol_active(today=date(2026, 6, 1)) is False


# ============================================================
# Explicit-authorization context manager
# ============================================================


def test_authorize_disposal_inside_block_active():
    assert has_explicit_disposal_authorization() is False
    with authorize_disposal():
        assert has_explicit_disposal_authorization() is True
    assert has_explicit_disposal_authorization() is False


def test_authorize_disposal_nested_blocks_decrement_correctly():
    """Reentrancy: nested with-blocks each increment, each exit
    decrements. Authorization stays active while at least one is open."""
    assert has_explicit_disposal_authorization() is False
    with authorize_disposal():
        assert has_explicit_disposal_authorization() is True
        with authorize_disposal():
            assert has_explicit_disposal_authorization() is True
        # Inner exited; outer still holds
        assert has_explicit_disposal_authorization() is True
    assert has_explicit_disposal_authorization() is False


def test_authorize_disposal_exception_inside_still_clears():
    """If an exception fires inside the with-block, the authorization
    is still cleared on exit (try/finally guarantee)."""
    with pytest.raises(RuntimeError):
        with authorize_disposal():
            assert has_explicit_disposal_authorization() is True
            raise RuntimeError("boom")
    assert has_explicit_disposal_authorization() is False


def test_authorize_disposal_is_per_thread():
    """One thread's authorization MUST NOT leak to another. Per-thread
    isolation is the load-bearing scoping property."""
    sibling_authorized = []

    def sibling():
        sibling_authorized.append(has_explicit_disposal_authorization())

    with authorize_disposal():
        t = threading.Thread(target=sibling)
        t.start()
        t.join()

    assert sibling_authorized == [False]


# ============================================================
# dispatch_for_disposal honors the gate
# ============================================================


def test_dispatch_proceeds_inside_authorize_disposal(
    active_handler, tmp_in_flight
):
    with authorize_disposal():
        record = dispatch_for_disposal(
            _input(),
            handler=active_handler,
            confess_with=_vidui(),
            authorized_by="kohen-gadol",
            in_flight_dir=tmp_in_flight,
        )
    assert record is not None


def test_dispatch_proceeds_inside_yom_kippur_window(
    active_handler, tmp_in_flight
):
    today = date.today()
    set_yom_kippur_window(today - timedelta(days=1), today + timedelta(days=1))
    record = dispatch_for_disposal(
        _input(),
        handler=active_handler,
        confess_with=_vidui(),
        authorized_by="kohen-gadol",
        in_flight_dir=tmp_in_flight,
    )
    assert record is not None


def test_dispatch_outside_envelope_writes_no_in_flight_file(
    active_handler, tmp_in_flight
):
    """Property 8 + 5 interaction: when the gate refuses, NO
    side-effect occurs. No checkpoint persisted, no payload held,
    no L3 record created."""
    try:
        dispatch_for_disposal(
            _input(case_ref="must-not-persist-001"),
            handler=active_handler,
            confess_with=_vidui(),
            authorized_by="kohen-gadol",
            in_flight_dir=tmp_in_flight,
        )
    except ProtocolViolation:
        pass

    # No file should have been created
    assert not (tmp_in_flight / "must-not-persist-001.json").exists()
    # The shaliach should not have held the payload
    assert active_handler.held_payload_count() == 0


def test_protocol_violation_raised_before_any_other_check(
    active_handler, tmp_in_flight
):
    """The protocol envelope is checked FIRST — even before the role
    gate. An inactive handler invoked outside the envelope should
    raise ProtocolViolation, not ShaliachError. This ordering matters
    because protocol restriction is the avodah-zarah-shape defense:
    we refuse on protocol grounds before doing any other validation."""
    h = IshItti("aharon")
    # Don't activate
    with pytest.raises(ProtocolViolation):
        dispatch_for_disposal(
            _input(),
            handler=h,
            confess_with=_vidui(),
            authorized_by="kohen-gadol",
            in_flight_dir=tmp_in_flight,
        )