"""Unnamed-marker authority initialization fault tests."""

from __future__ import annotations

import errno
import os
import signal
from pathlib import Path
from unittest import TestCase
from unittest.mock import patch

import checkin_cli.weekly_operations_link as descriptor_link
import checkin_cli.weekly_operations_publish as publisher
from checkin_cli.weekly_operations import WeeklyOperationsAuthorityCompromise, WeeklyOperationsCorruption, WeeklyOperationsPlatformNotSupported
from checkin_cli.weekly_operations_authority import AuthorityId, begin_authority_initialization, 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 tests._weekly_operations_support import acquire_for_child, initialize_at

ASSERTIONS = TestCase()


def _empty_root(tmp_path: Path, suffix: str) -> Path:
    root = tmp_path / suffix / "authority"
    root.mkdir(parents=True, mode=0o700)
    return root


def _assert_clean(root: Path) -> None:
    assert root.is_dir() and tuple(root.iterdir()) == ()
    assert not tuple(root.glob(".weekly-operations-init-*"))


def _retry(root: Path, identity: AuthorityId) -> None:
    authority = initialize_at(root, identity)
    binding = authority.binding
    authority.close()
    reopened = open_authority_root(authority.parent, binding)
    assert {path.name for path in root.iterdir()} == {MARKER_NAME, REGISTRY_NAME}
    reopened.close()


def test_partial_unnamed_marker_write_failure_leaves_no_name_and_retries(tmp_path: Path) -> None:
    root = _empty_root(tmp_path, "write")
    real_write = os.write

    def partial_then_fail(descriptor: int, payload: bytes) -> None:
        _ = real_write(descriptor, payload[:7])
        raise OSError("injected partial marker write")

    with patch.object(publisher, "_write_all", side_effect=partial_then_fail):
        with ASSERTIONS.assertRaises(OSError):
            _ = initialize_at(root, AuthorityId("2" * 64))
    _assert_clean(root)
    _retry(root, AuthorityId("2" * 64))


def test_unnamed_marker_fsync_failure_leaves_no_name_and_retries(tmp_path: Path) -> None:
    root = _empty_root(tmp_path, "fsync")
    with patch("os.fsync", side_effect=OSError("injected unnamed fsync")):
        with ASSERTIONS.assertRaises(OSError):
            _ = initialize_at(root, AuthorityId("3" * 64))
    _assert_clean(root)
    _retry(root, AuthorityId("3" * 64))


def test_link_failure_leaves_no_name_and_retries(tmp_path: Path) -> None:
    root = _empty_root(tmp_path, "link")
    with patch.object(publisher, "link_unnamed_file", side_effect=OSError("injected descriptor link")):
        with ASSERTIONS.assertRaisesRegex(WeeklyOperationsCorruption, "marker link failed"):
            _ = initialize_at(root, AuthorityId("4" * 64))
    _assert_clean(root)
    _retry(root, AuthorityId("4" * 64))


def test_unsupported_otmpfile_is_typed_without_fallback(tmp_path: Path) -> None:
    root = _empty_root(tmp_path, "unsupported-open")
    real_open = os.open

    def reject_unnamed(path: str | bytes, flags: int, mode: int = 0o777, *, dir_fd: int | None = None) -> int:
        if flags & os.O_TMPFILE == os.O_TMPFILE:
            raise OSError(errno.EOPNOTSUPP, "injected unsupported O_TMPFILE")
        return real_open(path, flags, mode, dir_fd=dir_fd)

    with patch("os.open", side_effect=reject_unnamed):
        with ASSERTIONS.assertRaises(WeeklyOperationsPlatformNotSupported):
            _ = initialize_at(root, AuthorityId("5" * 64))
    _assert_clean(root)


def test_unsupported_descriptor_link_is_typed_without_fallback(tmp_path: Path) -> None:
    root = _empty_root(tmp_path, "unsupported-link")
    with patch.object(publisher, "link_unnamed_file", side_effect=WeeklyOperationsPlatformNotSupported("injected unsupported link")):
        with ASSERTIONS.assertRaises(WeeklyOperationsPlatformNotSupported):
            _ = initialize_at(root, AuthorityId("6" * 64))
    _assert_clean(root)


def test_keyboard_interrupt_before_link_leaves_no_name(tmp_path: Path) -> None:
    root = _empty_root(tmp_path, "keyboard")
    with patch.object(publisher, "link_unnamed_file", side_effect=KeyboardInterrupt):
        with ASSERTIONS.assertRaises(KeyboardInterrupt):
            _ = initialize_at(root, AuthorityId("7" * 64))
    _assert_clean(root)


def test_sigterm_after_unnamed_write_leaves_no_name(tmp_path: Path) -> None:
    root = _empty_root(tmp_path, "sigterm")
    def write_then_terminate(descriptor: int, payload: bytes) -> None:
        _ = os.write(descriptor, payload)
        os.kill(os.getpid(), signal.SIGTERM)

    parent = acquire_for_child(root)
    binding = None
    with patch.object(publisher, "_write_all", side_effect=write_then_terminate):
        with ASSERTIONS.assertRaisesRegex(WeeklyOperationsCorruption, "cancelled by sigterm"):
            with begin_authority_initialization(parent, AuthorityId("8" * 64)) as transaction:
                binding = transaction.binding
                transaction.acknowledge_binding()
    assert binding is not None
    reopened = open_authority_root(parent, binding)
    reopened.close()


def test_system_exit_before_link_leaves_no_name(tmp_path: Path) -> None:
    root = _empty_root(tmp_path, "system-exit")
    with patch.object(publisher, "link_unnamed_file", side_effect=SystemExit(143)):
        with ASSERTIONS.assertRaises(SystemExit):
            _ = initialize_at(root, AuthorityId("9" * 64))
    _assert_clean(root)


def test_link_report_failure_after_commit_recovers_exact_binding(tmp_path: Path) -> None:
    root = _empty_root(tmp_path, "reported-link")

    def link_then_fail(source: int, directory: int, name: str) -> None:
        link_unnamed_file(source, directory, name)
        raise OSError("injected report failure after descriptor link")

    with patch.object(publisher, "link_unnamed_file", side_effect=link_then_fail):
        authority = initialize_at(root, AuthorityId("a" * 64))
    binding = authority.binding
    authority.close()
    reopened = open_authority_root(authority.parent, binding)
    assert reopened.binding == binding and {path.name for path in root.iterdir()} == {MARKER_NAME, REGISTRY_NAME}
    reopened.close()


def test_directory_fsync_report_failure_after_link_is_noncommitted(tmp_path: Path) -> None:
    root = _empty_root(tmp_path, "reported-fsync")
    real_fsync = os.fsync
    calls = 0

    def fail_after_directory_fsync(descriptor: int) -> None:
        nonlocal calls
        calls += 1
        real_fsync(descriptor)
        if calls == 2:
            raise OSError("injected report failure after directory fsync")

    with patch("os.fsync", side_effect=fail_after_directory_fsync):
        with ASSERTIONS.assertRaisesRegex(WeeklyOperationsCorruption, "registry directory fsync failed"):
            _ = initialize_at(root, AuthorityId("b" * 64))
    assert tuple(root.iterdir()) == ()


def test_malformed_marker_preoccupation_is_preserved(tmp_path: Path) -> None:
    root = _empty_root(tmp_path, "preoccupied")
    marker = root / MARKER_NAME
    _ = marker.write_bytes(b"unowned")
    marker.chmod(0o600)
    with ASSERTIONS.assertRaises(WeeklyOperationsAuthorityCompromise):
        _ = initialize_at(root, AuthorityId("c" * 64))
    assert marker.read_bytes() == b"unowned"


def test_valid_marker_with_invalid_preoccupied_inventory_is_typed_incident(tmp_path: Path) -> None:
    root = _empty_root(tmp_path, "preoccupied-inventory")
    authority = initialize_at(root, AuthorityId("d" * 64))
    injected = root / "unowned-entry"
    _ = injected.write_bytes(b"preserve")
    injected.chmod(0o600)
    with ASSERTIONS.assertRaises(WeeklyOperationsAuthorityCompromise):
        _ = initialize_at(root, AuthorityId("e" * 64))
    assert injected.read_bytes() == b"preserve"
    authority.close()


def test_descriptor_link_wrapper_converts_unsupported_errno() -> None:
    with patch.object(descriptor_link, "_LINKAT", return_value=-1), patch("ctypes.get_errno", return_value=errno.EOPNOTSUPP):
        with ASSERTIONS.assertRaises(WeeklyOperationsPlatformNotSupported):
            descriptor_link.link_unnamed_file(10, 11, MARKER_NAME)