from __future__ import annotations

import ctypes
import importlib.util
import os
import select
import signal
import sys
from collections.abc import Callable
from pathlib import Path
from types import ModuleType
from typing import Protocol, cast

import pytest

ROOT = Path(__file__).resolve().parent


class _Syscall(Protocol):
    argtypes: list[object]
    restype: object

    def __call__(self, number: int, pid: int, flags: int, /) -> int: ...


def _protocol() -> ModuleType:
    spec = importlib.util.spec_from_file_location(
        "timeout_protocol", ROOT / "task22_child_protocol.py"
    )
    if spec is None or spec.loader is None:
        raise AssertionError("completion protocol loader missing")
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    return module


def _pidfd_open(pid: int) -> int:
    syscall = cast(_Syscall, cast(object, ctypes.CDLL(None, use_errno=True).syscall))
    syscall.argtypes = [ctypes.c_long, ctypes.c_int, ctypes.c_uint]
    syscall.restype = ctypes.c_long
    descriptor = syscall(434, pid, 0)
    if descriptor < 0:
        error = ctypes.get_errno()
        raise OSError(error, os.strerror(error))
    return descriptor


def _libc_kill(pidfd: int) -> None:
    function = ctypes.CDLL(None, use_errno=True).pidfd_send_signal
    function.argtypes = [ctypes.c_int, ctypes.c_int, ctypes.c_void_p, ctypes.c_uint]
    function.restype = ctypes.c_int
    if function(pidfd, signal.SIGKILL, None, 0) != 0:
        error = ctypes.get_errno()
        raise OSError(error, os.strerror(error))

class _FakeSender:
    def __init__(self, error: int = 0) -> None:
        self.argtypes: list[object] = []
        self.restype: object = None
        self.error: int = error
        self.calls: list[tuple[int, int, None, int]] = []

    def __call__(self, pidfd: int, sig: int, info: None, flags: int, /) -> int:
        self.calls.append((pidfd, sig, info, flags))
        if self.error:
            _ = ctypes.set_errno(self.error)
            return -1
        return 0


class _FakeLibrary:
    def __init__(self, sender: _FakeSender) -> None:
        self.pidfd_send_signal: _FakeSender = sender


def _hanging_child(receipt: bytes = b"") -> tuple[int, int, int, int, int]:
    ready_read, ready_write = os.pipe2(os.O_CLOEXEC)
    gate_read, gate_write = os.pipe2(os.O_CLOEXEC)
    receipt_read, receipt_write = os.pipe2(os.O_CLOEXEC)
    pid = os.fork()
    if pid == 0:
        os.close(ready_read)
        os.close(gate_write)
        os.close(receipt_read)
        if receipt:
            _ = os.write(receipt_write, receipt)
        _ = os.write(ready_write, b"1")
        _ = os.read(gate_read, 1)
        os._exit(98)
    os.close(ready_write)
    os.close(gate_read)
    os.close(receipt_write)
    pidfd = _pidfd_open(pid)
    ready = select.poll()
    ready.register(ready_read, select.POLLIN)
    assert ready.poll(2_000)
    assert os.read(ready_read, 1) == b"1"
    return pid, pidfd, receipt_read, gate_write, ready_read


def _close_child(resources: tuple[int, int, int, int, int]) -> None:
    pid, pidfd, receipt_read, gate_write, ready_read = resources
    try:
        try:
            waited, _ = os.waitpid(pid, os.WNOHANG)
        except ChildProcessError:
            waited = pid
        if waited == 0:
            _libc_kill(pidfd)
            reaper = select.poll()
            reaper.register(pidfd, select.POLLIN)
            assert reaper.poll(2_000)
            waited, _ = os.waitpid(pid, 0)
        assert waited == pid
    finally:
        os.close(pidfd)
        os.close(receipt_read)
        os.close(gate_write)
        os.close(ready_read)


def test_pidfd_binding_validates_types_signal_flags_and_exact_fd(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    protocol = _protocol()
    resources = _hanging_child()
    pid, pidfd, _, _, _ = resources
    sender = _FakeSender()
    library = _FakeLibrary(sender)
    def load_library(*_args: object, **_kwargs: object) -> object:
        return library

    monkeypatch.setattr(ctypes, "CDLL", load_library)
    send = cast(Callable[[int, int], None], getattr(protocol, "_send_pidfd_sigkill"))
    try:
        send(pid, pidfd)
        assert sender.argtypes == [
            ctypes.c_int, ctypes.c_int, ctypes.c_void_p, ctypes.c_uint
        ]
        assert sender.restype is ctypes.c_int
        assert sender.calls == [(pidfd, int(signal.SIGKILL), None, 0)]
    finally:
        monkeypatch.undo()
        _close_child(resources)


def test_invalid_or_mismatched_pidfd_fails_closed() -> None:
    protocol = _protocol()
    send = cast(Callable[[int, int], None], getattr(protocol, "_send_pidfd_sigkill"))
    read_fd, write_fd = os.pipe2(os.O_CLOEXEC)
    try:
        with pytest.raises(RuntimeError, match="pidfd"):
            send(os.getpid(), read_fd)
        with pytest.raises(ValueError, match="pidfd"):
            send(os.getpid(), -1)
    finally:
        os.close(read_fd)
        os.close(write_fd)
    resources = _hanging_child()
    pid, pidfd, _, _, _ = resources
    try:
        with pytest.raises(RuntimeError, match="requested child"):
            send(pid + 1, pidfd)
    finally:
        _close_child(resources)


def test_missing_libc_symbol_fails_closed(monkeypatch: pytest.MonkeyPatch) -> None:
    protocol = _protocol()
    def load_library(*_args: object, **_kwargs: object) -> object:
        return object()

    monkeypatch.setattr(ctypes, "CDLL", load_library)
    load = cast(Callable[[], object], getattr(protocol, "_load_pidfd_sender"))
    with pytest.raises(RuntimeError, match="unavailable"):
        _ = load()
    monkeypatch.undo()
    monkeypatch.setattr(sys, "platform", "darwin")
    with pytest.raises(RuntimeError, match="requires Linux"):
        _ = load()


def test_libc_errno_is_preserved(monkeypatch: pytest.MonkeyPatch) -> None:
    protocol = _protocol()
    resources = _hanging_child()
    pid, pidfd, _, _, _ = resources
    sender = _FakeSender(1)
    def load_library(*_args: object, **_kwargs: object) -> object:
        return _FakeLibrary(sender)

    monkeypatch.setattr(ctypes, "CDLL", load_library)
    send = cast(Callable[[int, int], None], getattr(protocol, "_send_pidfd_sigkill"))
    try:
        with pytest.raises(PermissionError) as raised:
            send(pid, pidfd)
        assert raised.value.errno == 1
    finally:
        monkeypatch.undo()
        _close_child(resources)


def test_already_exited_child_race_is_reaped() -> None:
    protocol = _protocol()
    kill = cast(Callable[[int, int], None], getattr(protocol, "_kill_pidfd"))
    pid = os.fork()
    if pid == 0:
        os._exit(17)
    pidfd = _pidfd_open(pid)
    try:
        ready = select.poll()
        ready.register(pidfd, select.POLLIN)
        assert ready.poll(2_000)
        kill(pid, pidfd)
        with pytest.raises(ChildProcessError):
            _ = os.waitpid(pid, os.WNOHANG)
    finally:
        os.close(pidfd)


def test_receipt_at_timeout_boundary_is_not_forced_timeout() -> None:
    protocol = _protocol()
    frame = cast(Callable[[str, int], bytes], getattr(protocol, "frame"))
    await_child = cast(
        Callable[[int, int, int, int], int], getattr(protocol, "await_child")
    )
    resources = _hanging_child(frame("exit", 0))
    pid, pidfd, receipt_read, _, _ = resources
    try:
        with pytest.raises(RuntimeError, match="requires no completion receipt"):
            _ = await_child(pid, pidfd, receipt_read, 1)
    finally:
        _close_child(resources)


@pytest.mark.parametrize("iteration", range(3))
def test_hanging_timeout_is_repeatable_without_orphan(iteration: int) -> None:
    del iteration
    protocol = _protocol()
    await_child = cast(
        Callable[[int, int, int, int], int], getattr(protocol, "await_child")
    )
    resources = _hanging_child()
    pid, pidfd, receipt_read, _, _ = resources
    try:
        with pytest.raises(RuntimeError, match="bounded execution window"):
            _ = await_child(pid, pidfd, receipt_read, 1)
    finally:
        _close_child(resources)