"""Explicit caller binding-handoff transaction tests."""

from __future__ import annotations

import copy
import fcntl
import os
import signal
from concurrent.futures import ThreadPoolExecutor
from threading import Barrier
from contextlib import nullcontext
from pathlib import Path
from types import FrameType, TracebackType
from unittest import TestCase
from unittest.mock import patch

import checkin_cli.weekly_operations_authority as authority_module
from checkin_cli.weekly_operations import WeeklyOperationsAlreadyInitialized, WeeklyOperationsInputError
from checkin_cli.weekly_operations_authority import AuthorityId, WeeklyOperationsAuthorityRoot, WeeklyOperationsInitializationTransaction, begin_authority_initialization, open_authority_root
from checkin_cli.weekly_operations_parent import WeeklyOperationsAuthorityBinding, WeeklyOperationsParentAuthority, acquire_parent_authority
from checkin_cli.weekly_operations_signals import InitializationCancellation, InitializationCancellationKind, InitializationCommitContext, WeeklyOperationsInitializationCommitted
from tests._weekly_operations_support import initialize_for_parent

ASSERTIONS = TestCase()


class CallerPersistenceFailure(RuntimeError):
    """Injected caller-side binding persistence failure."""


_SIGNALS = ((signal.SIGTERM, InitializationCancellationKind.SIGTERM), (signal.SIGINT, InitializationCancellationKind.SIGINT))


def _root(tmp_path: Path, name: str) -> Path:
    root = tmp_path / name
    root.mkdir(mode=0o700)
    return root


def _fresh_open(parent: WeeklyOperationsParentAuthority, binding: WeeklyOperationsAuthorityBinding) -> None:
    reopened = open_authority_root(parent, binding)
    assert reopened.binding == binding
    reopened.close()


def test_r12_finally_seam_hands_binding_to_caller_before_unmask_twice(tmp_path: Path) -> None:
    for signal_number, kind in _SIGNALS:
        for attempt in (1, 2):
            root = _root(tmp_path, f"return-seam-{kind.value}-{attempt}")
            parent = acquire_parent_authority(root)
            real_close = os.close
            sent = False
            captured = None

            def close_then_cancel(descriptor: int) -> None:
                nonlocal sent
                info = os.fstat(descriptor)
                access = fcntl.fcntl(descriptor, fcntl.F_GETFL) & os.O_ACCMODE
                real_close(descriptor)
                if not sent and info.st_nlink == 1 and access == os.O_RDWR:
                    sent = True
                    os.kill(os.getpid(), signal_number)

            with ASSERTIONS.assertRaises(InitializationCancellation) as raised:
                with patch("os.close", side_effect=close_then_cancel):
                    with begin_authority_initialization(parent, AuthorityId("1" * 64)) as transaction:
                        captured = transaction.binding
                        transaction.acknowledge_binding()
            assert raised.exception.cancellation_kind is kind and captured is not None
            _fresh_open(parent, captured)
            with ASSERTIONS.assertRaises(WeeklyOperationsAlreadyInitialized):
                _ = initialize_for_parent(parent, AuthorityId("9" * 64))


def _acknowledged_signal_position(tmp_path: Path, position: str) -> None:
    for signal_number, kind in _SIGNALS:
        for attempt in (1, 2):
            root = _root(tmp_path, f"{position}-{kind.value}-{attempt}")
            parent = acquire_parent_authority(root)
            captured = None
            real_exit = WeeklyOperationsInitializationTransaction.__exit__

            def cancel_at_exit(transaction: WeeklyOperationsInitializationTransaction, error_type: type[BaseException] | None, error: BaseException | None, traceback: TracebackType | None) -> bool:
                os.kill(os.getpid(), signal_number)
                return real_exit(transaction, error_type, error, traceback)

            manager = patch.object(WeeklyOperationsInitializationTransaction, "__exit__", new=cancel_at_exit) if position == "first-exit-line" else nullcontext()
            with ASSERTIONS.assertRaises(InitializationCancellation) as raised:
                with manager:
                    with begin_authority_initialization(parent, AuthorityId("2" * 64)) as transaction:
                        if position == "first-body-line":
                            os.kill(os.getpid(), signal_number)
                        captured = transaction.binding
                        if position == "before-ack":
                            os.kill(os.getpid(), signal_number)
                        transaction.acknowledge_binding()
                        if position == "after-ack":
                            os.kill(os.getpid(), signal_number)
            assert raised.exception.cancellation_kind is kind and captured is not None
            _fresh_open(parent, captured)


def test_signal_on_first_body_line_cannot_prevent_binding_capture(tmp_path: Path) -> None:
    _acknowledged_signal_position(tmp_path, "first-body-line")


def test_signal_after_capture_before_ack_cannot_lose_binding(tmp_path: Path) -> None:
    _acknowledged_signal_position(tmp_path, "before-ack")


def test_signal_after_ack_delivers_only_after_handoff(tmp_path: Path) -> None:
    _acknowledged_signal_position(tmp_path, "after-ack")


def test_signal_on_first_exit_line_delivers_only_after_handoff(tmp_path: Path) -> None:
    _acknowledged_signal_position(tmp_path, "first-exit-line")


def test_constructor_is_inert_and_preenter_signals_are_noncommitted(tmp_path: Path) -> None:
    for signal_number, kind in _SIGNALS:
        root = _root(tmp_path, f"before-enter-{kind.value}")
        parent = acquire_parent_authority(root)
        before = signal.pthread_sigmask(signal.SIG_BLOCK, frozenset())
        transaction = begin_authority_initialization(parent, AuthorityId("3" * 64))
        assert not hasattr(authority_module, "initialize_authority_root")
        assert signal.pthread_sigmask(signal.SIG_BLOCK, frozenset()) == before

        def cancel(observed: int, _frame: FrameType | None) -> None:
            observed_kind = InitializationCancellationKind.SIGTERM if observed == signal.SIGTERM else InitializationCancellationKind.SIGINT
            raise InitializationCancellation(observed_kind)

        previous = signal.signal(signal_number, cancel)
        try:
            with ASSERTIONS.assertRaises(InitializationCancellation) as raised:
                os.kill(os.getpid(), signal_number)
        finally:
            _ = signal.signal(signal_number, previous)
        assert raised.exception.cancellation_kind is kind and tuple(root.iterdir()) == ()
        with transaction as entered:
            _ = entered.binding
            entered.acknowledge_binding()


def test_nested_prior_mask_is_restored_with_binding_exposed(tmp_path: Path) -> None:
    root = _root(tmp_path, "nested-mask")
    parent = acquire_parent_authority(root)
    def cancel(_signal_number: int, _frame: FrameType | None) -> None:
        raise InitializationCancellation(InitializationCancellationKind.SIGTERM)

    previous_handler = signal.signal(signal.SIGTERM, cancel)
    prior = signal.pthread_sigmask(signal.SIG_BLOCK, {signal.SIGTERM, signal.SIGINT})
    captured = None
    try:
        os.kill(os.getpid(), signal.SIGTERM)
        with begin_authority_initialization(parent, AuthorityId("4" * 64)) as transaction:
            captured = transaction.binding
            transaction.acknowledge_binding()
        current = signal.pthread_sigmask(signal.SIG_BLOCK, frozenset())
        assert signal.SIGTERM in current and signal.SIGINT in current
        assert captured is not None
        _fresh_open(parent, captured)
        assert signal.SIGTERM in signal.sigpending()
        _ = signal.pthread_sigmask(signal.SIG_UNBLOCK, {signal.SIGTERM})
    except InitializationCancellation as cancellation:
        assert cancellation.cancellation_kind is InitializationCancellationKind.SIGTERM and captured is not None
    finally:
        _ = signal.pthread_sigmask(signal.SIG_SETMASK, prior)
        _ = signal.signal(signal.SIGTERM, previous_handler)


def test_missing_ack_raises_committed_outcome_with_binding(tmp_path: Path) -> None:
    root = _root(tmp_path, "missing-ack")
    parent = acquire_parent_authority(root)
    binding: WeeklyOperationsAuthorityBinding | None = None
    with ASSERTIONS.assertRaises(WeeklyOperationsInitializationCommitted) as raised:
        with begin_authority_initialization(parent, AuthorityId("5" * 64)) as transaction:
            binding = transaction.binding
    assert binding is not None and raised.exception.binding == binding
    assert raised.exception.exit_context is InitializationCommitContext.MISSING_ACKNOWLEDGEMENT
    _fresh_open(parent, raised.exception.binding)


def test_pending_cancellation_without_ack_raises_binding_outcome(tmp_path: Path) -> None:
    for signal_number, kind in _SIGNALS:
        root = _root(tmp_path, f"unacknowledged-{kind.value}")
        parent = acquire_parent_authority(root)
        binding: WeeklyOperationsAuthorityBinding | None = None
        with ASSERTIONS.assertRaises(WeeklyOperationsInitializationCommitted) as raised:
            with begin_authority_initialization(parent, AuthorityId("a" * 64)) as transaction:
                binding = transaction.binding
                os.kill(os.getpid(), signal_number)
        assert binding is not None and raised.exception.binding == binding
        assert raised.exception.exit_context is InitializationCommitContext.CANCELLATION
        assert raised.exception.cancellation_kind is kind
        _fresh_open(parent, binding)


def test_body_exception_before_ack_is_wrapped_with_exact_binding(tmp_path: Path) -> None:
    root = _root(tmp_path, "body-error")
    parent = acquire_parent_authority(root)
    binding: WeeklyOperationsAuthorityBinding | None = None
    with ASSERTIONS.assertRaises(WeeklyOperationsInitializationCommitted) as raised:
        with begin_authority_initialization(parent, AuthorityId("6" * 64)) as transaction:
            binding = transaction.binding
            raise CallerPersistenceFailure("injected caller persistence failure")
    assert binding is not None and raised.exception.binding == binding
    assert raised.exception.exit_context is InitializationCommitContext.BODY_EXCEPTION
    assert isinstance(raised.exception.original_context, CallerPersistenceFailure)


def test_single_enter_idempotent_ack_no_copy_and_active_binding_scope(tmp_path: Path) -> None:
    root = _root(tmp_path, "lifecycle")
    transaction = begin_authority_initialization(acquire_parent_authority(root), AuthorityId("7" * 64))
    with ASSERTIONS.assertRaises(WeeklyOperationsInputError):
        _ = transaction.binding
    with ASSERTIONS.assertRaises(WeeklyOperationsInputError):
        _ = copy.copy(transaction)
    with ASSERTIONS.assertRaises(WeeklyOperationsInputError):
        _ = copy.deepcopy(transaction)
    with transaction:
        _ = transaction.binding
        with ASSERTIONS.assertRaises(WeeklyOperationsInputError):
            _ = transaction.__enter__()
        transaction.acknowledge_binding()
        transaction.acknowledge_binding()
    with ASSERTIONS.assertRaises(WeeklyOperationsInputError):
        _ = transaction.binding
    with ASSERTIONS.assertRaises(WeeklyOperationsInputError):
        _ = transaction.__enter__()


def test_concurrent_winner_and_loser_restore_each_thread_mask(tmp_path: Path) -> None:
    root = _root(tmp_path, "concurrent-mask")
    gate = Barrier(2)

    def initialize(identity: AuthorityId) -> tuple[WeeklyOperationsAuthorityRoot | WeeklyOperationsAlreadyInitialized, bool]:
        parent = acquire_parent_authority(root)
        before = signal.pthread_sigmask(signal.SIG_BLOCK, frozenset())
        _ = gate.wait(timeout=5)
        try:
            outcome: WeeklyOperationsAuthorityRoot | WeeklyOperationsAlreadyInitialized = initialize_for_parent(parent, identity)
        except WeeklyOperationsAlreadyInitialized as error:
            outcome = error
        after = signal.pthread_sigmask(signal.SIG_BLOCK, frozenset())
        return outcome, after == before

    with ThreadPoolExecutor(max_workers=2) as executor:
        results = tuple(executor.map(initialize, (AuthorityId("b" * 64), AuthorityId("c" * 64))))
    outcomes = tuple(result[0] for result in results)
    assert sum(isinstance(value, WeeklyOperationsAuthorityRoot) for value in outcomes) == 1
    assert sum(isinstance(value, WeeklyOperationsAlreadyInitialized) for value in outcomes) == 1
    assert all(result[1] for result in results)


def test_wrong_thread_access_is_rejected_without_mask_leak(tmp_path: Path) -> None:
    root = _root(tmp_path, "wrong-thread")
    before = signal.pthread_sigmask(signal.SIG_BLOCK, frozenset())
    with begin_authority_initialization(acquire_parent_authority(root), AuthorityId("8" * 64)) as transaction:
        def misuse() -> tuple[type[WeeklyOperationsInputError], ...]:
            errors: list[type[WeeklyOperationsInputError]] = []
            try:
                transaction.acknowledge_binding()
            except WeeklyOperationsInputError as error:
                errors.append(type(error))
            try:
                _ = transaction.__exit__(None, None, None)
            except WeeklyOperationsInputError as error:
                errors.append(type(error))
            return tuple(errors)

        with ThreadPoolExecutor(max_workers=1) as executor:
            kinds = executor.submit(misuse).result(timeout=5)
        assert kinds == (WeeklyOperationsInputError, WeeklyOperationsInputError)
        transaction.acknowledge_binding()
    assert signal.pthread_sigmask(signal.SIG_BLOCK, frozenset()) == before