"""Signals and failures inside the caller-handoff initialization transaction."""

from __future__ import annotations

import os
import signal
import stat
from contextlib import ExitStack, nullcontext
from pathlib import Path
from unittest import TestCase
from unittest.mock import patch

import checkin_cli.weekly_operations_authority as authority_module
import checkin_cli.weekly_operations_publish as publisher
from checkin_cli.weekly_operations import WeeklyOperationsAuthorityCompromise, WeeklyOperationsCorruption
from checkin_cli.weekly_operations_authority import AuthorityId, begin_authority_initialization, construct_authority_binding, open_authority_root
from checkin_cli.weekly_operations_canonical_registry_history import REGISTRY_NAME
from checkin_cli.weekly_operations_layout import MARKER_NAME
from checkin_cli.weekly_operations_link import link_unnamed_file
from checkin_cli.weekly_operations_parent import WeeklyOperationsAuthorityBinding, acquire_parent_authority
from checkin_cli.weekly_operations_signals import InitializationCancellation, InitializationCancellationKind

ASSERTIONS = TestCase()
_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 _handoff_case(tmp_path: Path, seam: str, signal_number: signal.Signals, kind: InitializationCancellationKind, attempt: int) -> None:
    root = _root(tmp_path, f"{seam}-{kind.value}-{attempt}")
    parent = acquire_parent_authority(root)
    directory_fsyncs: list[int] = []
    real_fsync = os.fsync
    real_link = link_unnamed_file
    real_reconcile = publisher.reconcile_owned_marker
    real_binding = construct_authority_binding

    def observed_fsync(descriptor: int) -> None:
        info = os.fstat(descriptor)
        real_fsync(descriptor)
        if stat.S_ISDIR(info.st_mode) and info.st_ino == root.stat().st_ino:
            directory_fsyncs.append(descriptor)

    def cancel_after_link(source: int, directory: int, name: str) -> None:
        real_link(source, directory, name)
        os.kill(os.getpid(), signal_number)

    def cancel_before_reconcile(directory: int, expected: tuple[int, int], payload: bytes) -> os.stat_result:
        os.kill(os.getpid(), signal_number)
        return real_reconcile(directory, expected, payload)

    def cancel_before_binding(
        authority_id: AuthorityId, root_info: os.stat_result, marker: os.stat_result,
        marker_digest: str, registry: os.stat_result, registry_digest: str,
    ) -> WeeklyOperationsAuthorityBinding:
        os.kill(os.getpid(), signal_number)
        return real_binding(
            authority_id, root_info, marker, marker_digest, registry, registry_digest
        )

    captured = None
    with ExitStack() as stack:
        _ = stack.enter_context(patch("os.fsync", side_effect=observed_fsync))
        if seam == "after-link-pre-fsync":
            _ = stack.enter_context(patch.object(publisher, "link_unnamed_file", side_effect=cancel_after_link))
        elif seam == "post-fsync-pre-reconcile":
            _ = stack.enter_context(patch.object(publisher, "reconcile_owned_marker", side_effect=cancel_before_reconcile))
        elif seam == "post-reconcile-pre-binding":
            _ = stack.enter_context(patch.object(authority_module, "construct_authority_binding", side_effect=cancel_before_binding))
        else:
            _ = stack.enter_context(nullcontext())
        with ASSERTIONS.assertRaises(InitializationCancellation) as raised:
            with begin_authority_initialization(parent, AuthorityId("1" * 64)) as transaction:
                if seam == "after-binding":
                    os.kill(os.getpid(), signal_number)
                captured = transaction.binding
                transaction.acknowledge_binding()

    assert raised.exception.cancellation_kind is kind and directory_fsyncs and captured is not None
    reopened = open_authority_root(parent, captured)
    reopened.close()
    assert captured.marker_inode == (root / MARKER_NAME).stat().st_ino


def _repeat_handoff_seam(tmp_path: Path, seam: str) -> None:
    for signal_number, kind in _SIGNALS:
        for attempt in (1, 2):
            _handoff_case(tmp_path, seam, signal_number, kind, attempt)


def test_signals_after_link_before_fsync_wait_for_handoff_twice(tmp_path: Path) -> None:
    _repeat_handoff_seam(tmp_path, "after-link-pre-fsync")


def test_signals_post_fsync_before_reconcile_wait_for_handoff_twice(tmp_path: Path) -> None:
    _repeat_handoff_seam(tmp_path, "post-fsync-pre-reconcile")


def test_signals_post_reconcile_before_binding_wait_for_handoff_twice(tmp_path: Path) -> None:
    _repeat_handoff_seam(tmp_path, "post-reconcile-pre-binding")


def test_signals_after_binding_wait_for_acknowledged_exit_twice(tmp_path: Path) -> None:
    _repeat_handoff_seam(tmp_path, "after-binding")


def test_pending_signal_cannot_convert_directory_fsync_failure_to_handoff(tmp_path: Path) -> None:
    root = _root(tmp_path, "fsync-failure")
    parent = acquire_parent_authority(root)
    real_fsync = os.fsync

    def link_then_cancel(source: int, directory: int, name: str) -> None:
        link_unnamed_file(source, directory, name)
        os.kill(os.getpid(), signal.SIGTERM)

    def fail_directory_fsync(descriptor: int) -> None:
        if stat.S_ISDIR(os.fstat(descriptor).st_mode):
            raise OSError("injected directory fsync failure")
        real_fsync(descriptor)

    with patch.object(publisher, "link_unnamed_file", side_effect=link_then_cancel), patch("os.fsync", side_effect=fail_directory_fsync):
        with ASSERTIONS.assertRaisesRegex(WeeklyOperationsCorruption, "registry directory fsync failed"):
            with begin_authority_initialization(parent, AuthorityId("4" * 64)):
                raise AssertionError("enter unexpectedly succeeded")
    assert tuple(root.iterdir()) == ()


def test_pending_signal_cannot_convert_reconcile_failure_to_handoff(tmp_path: Path) -> None:
    root = _root(tmp_path, "reconcile-failure")
    parent = acquire_parent_authority(root)

    def link_then_cancel(source: int, directory: int, name: str) -> None:
        link_unnamed_file(source, directory, name)
        os.kill(os.getpid(), signal.SIGINT)

    with patch.object(publisher, "link_unnamed_file", side_effect=link_then_cancel), patch.object(publisher, "reconcile_owned_marker", side_effect=WeeklyOperationsAuthorityCompromise("injected reconcile failure")):
        with ASSERTIONS.assertRaisesRegex(WeeklyOperationsAuthorityCompromise, "reconcile failure"):
            with begin_authority_initialization(parent, AuthorityId("5" * 64)):
                raise AssertionError("enter unexpectedly succeeded")
    assert {path.name for path in root.iterdir()} == {MARKER_NAME, REGISTRY_NAME}


def test_pending_signal_cannot_convert_binding_failure_to_handoff(tmp_path: Path) -> None:
    root = _root(tmp_path, "binding-failure")
    parent = acquire_parent_authority(root)

    def link_then_cancel(source: int, directory: int, name: str) -> None:
        link_unnamed_file(source, directory, name)
        os.kill(os.getpid(), signal.SIGTERM)

    with patch.object(publisher, "link_unnamed_file", side_effect=link_then_cancel), patch.object(authority_module, "construct_authority_binding", side_effect=WeeklyOperationsCorruption("injected binding failure")):
        with ASSERTIONS.assertRaisesRegex(WeeklyOperationsCorruption, "binding failure"):
            with begin_authority_initialization(parent, AuthorityId("6" * 64)):
                raise AssertionError("enter unexpectedly succeeded")
    assert {path.name for path in root.iterdir()} == {MARKER_NAME, REGISTRY_NAME}