from __future__ import annotations

import multiprocessing
import sys
import sysconfig
import threading
import time

import pytest

import env
from pybind11_tests import gil_scoped as m

# Test collection seems to hold the gil
# These tests have rare flakes in nogil; since they
# are testing the gil, they are skipped at the moment.
skipif_not_free_threaded = pytest.mark.skipif(
    sysconfig.get_config_var("Py_GIL_DISABLED"),
    reason="Flaky without the GIL",
)


class ExtendedVirtClass(m.VirtClass):
    def virtual_func(self):
        pass

    def pure_virtual_func(self):
        pass


def test_callback_py_obj():
    m.test_callback_py_obj(lambda: None)


def test_callback_std_func():
    m.test_callback_std_func(lambda: None)


def test_callback_virtual_func():
    extended = ExtendedVirtClass()
    m.test_callback_virtual_func(extended)


def test_callback_pure_virtual_func():
    extended = ExtendedVirtClass()
    m.test_callback_pure_virtual_func(extended)


def test_cross_module_gil_released():
    """Makes sure that the GIL can be acquired by another module from a GIL-released state."""
    m.test_cross_module_gil_released()  # Should not raise a SIGSEGV


def test_cross_module_gil_acquired():
    """Makes sure that the GIL can be acquired by another module from a GIL-acquired state."""
    m.test_cross_module_gil_acquired()  # Should not raise a SIGSEGV


def test_cross_module_gil_inner_custom_released():
    """Makes sure that the GIL can be acquired/released by another module
    from a GIL-released state using custom locking logic."""
    m.test_cross_module_gil_inner_custom_released()


def test_cross_module_gil_inner_custom_acquired():
    """Makes sure that the GIL can be acquired/acquired by another module
    from a GIL-acquired state using custom locking logic."""
    m.test_cross_module_gil_inner_custom_acquired()


def test_cross_module_gil_inner_pybind11_released():
    """Makes sure that the GIL can be acquired/released by another module
    from a GIL-released state using pybind11 locking logic."""
    m.test_cross_module_gil_inner_pybind11_released()


def test_cross_module_gil_inner_pybind11_acquired():
    """Makes sure that the GIL can be acquired/acquired by another module
    from a GIL-acquired state using pybind11 locking logic."""
    m.test_cross_module_gil_inner_pybind11_acquired()


@pytest.mark.skipif(sys.platform.startswith("emscripten"), reason="Requires threads")
def test_cross_module_gil_nested_custom_released():
    """Makes sure that the GIL can be nested acquired/released by another module
    from a GIL-released state using custom locking logic."""
    m.test_cross_module_gil_nested_custom_released()


@pytest.mark.skipif(sys.platform.startswith("emscripten"), reason="Requires threads")
def test_cross_module_gil_nested_custom_acquired():
    """Makes sure that the GIL can be nested acquired/acquired by another module
    from a GIL-acquired state using custom locking logic."""
    m.test_cross_module_gil_nested_custom_acquired()


@pytest.mark.skipif(sys.platform.startswith("emscripten"), reason="Requires threads")
def test_cross_module_gil_nested_pybind11_released():
    """Makes sure that the GIL can be nested acquired/released by another module
    from a GIL-released state using pybind11 locking logic."""
    m.test_cross_module_gil_nested_pybind11_released()


@pytest.mark.skipif(sys.platform.startswith("emscripten"), reason="Requires threads")
def test_cross_module_gil_nested_pybind11_acquired():
    """Makes sure that the GIL can be nested acquired/acquired by another module
    from a GIL-acquired state using pybind11 locking logic."""
    m.test_cross_module_gil_nested_pybind11_acquired()


def test_release_acquire():
    assert m.test_release_acquire(0xAB) == "171"


def test_nested_acquire():
    assert m.test_nested_acquire(0xAB) == "171"


@pytest.mark.skipif(sys.platform.startswith("emscripten"), reason="Requires threads")
@pytest.mark.skipif(
    env.GRAALPY and sys.platform == "darwin",
    reason="Transiently crashes on GraalPy on OS X",
)
def test_multi_acquire_release_cross_module():
    for bits in range(16 * 8):
        internals_ids = m.test_multi_acquire_release_cross_module(bits)
        assert len(internals_ids) == 2 if bits % 8 else 1


# Intentionally putting human review in the loop here, to guard against accidents.
VARS_BEFORE_ALL_BASIC_TESTS = dict(vars())  # Make a copy of the dict (critical).
ALL_BASIC_TESTS = (
    test_callback_py_obj,
    test_callback_std_func,
    test_callback_virtual_func,
    test_callback_pure_virtual_func,
    test_cross_module_gil_released,
    test_cross_module_gil_acquired,
    test_cross_module_gil_inner_custom_released,
    test_cross_module_gil_inner_custom_acquired,
    test_cross_module_gil_inner_pybind11_released,
    test_cross_module_gil_inner_pybind11_acquired,
    test_cross_module_gil_nested_custom_released,
    test_cross_module_gil_nested_custom_acquired,
    test_cross_module_gil_nested_pybind11_released,
    test_cross_module_gil_nested_pybind11_acquired,
    test_release_acquire,
    test_nested_acquire,
    test_multi_acquire_release_cross_module,
)


def test_all_basic_tests_completeness():
    num_found = 0
    for key, value in VARS_BEFORE_ALL_BASIC_TESTS.items():
        if not key.startswith("test_"):
            continue
        assert value in ALL_BASIC_TESTS
        num_found += 1
    assert len(ALL_BASIC_TESTS) == num_found


def _intentional_deadlock():
    m.intentional_deadlock()


ALL_BASIC_TESTS_PLUS_INTENTIONAL_DEADLOCK = (*ALL_BASIC_TESTS, _intentional_deadlock)


def _run_in_process(target, *args, **kwargs):
    if env.ANDROID or env.IOS or sys.platform.startswith("emscripten"):
        pytest.skip("Requires subprocess support")

    test_fn = target if len(args) == 0 else args[0]
    # Do not need to wait much, 10s should be more than enough.
    timeout = 0.1 if test_fn is _intentional_deadlock else 10
    process = multiprocessing.Process(target=target, args=args, kwargs=kwargs)
    process.daemon = True
    try:
        t_start = time.time()
        process.start()
        if timeout >= 100:  # For debugging.
            print(
                "\nprocess.pid STARTED", process.pid, (sys.argv, target, args, kwargs)
            )
            print(f"COPY-PASTE-THIS: gdb {sys.argv[0]} -p {process.pid}", flush=True)
        process.join(timeout=timeout)
        if timeout >= 100:
            print("\nprocess.pid JOINED", process.pid, flush=True)
        t_delta = time.time() - t_start
        if process.exitcode == 66 and m.defined_THREAD_SANITIZER:  # Issue #2754
            # WOULD-BE-NICE-TO-HAVE: Check that the message below is actually in the output.
            # Maybe this could work:
            # https://gist.github.com/alexeygrigorev/01ce847f2e721b513b42ea4a6c96905e
            pytest.skip(
                "ThreadSanitizer: starting new threads after multi-threaded fork is not supported."
            )
        elif test_fn is _intentional_deadlock:
            assert process.exitcode is None
            return 0

        if process.exitcode is None:
            assert t_delta > 0.9 * timeout
            msg = "DEADLOCK, most likely, exactly what this test is meant to detect."
            soabi = sysconfig.get_config_var("SOABI")
            if env.WIN and env.PYPY:
                pytest.xfail(f"[TEST-GIL-SCOPED] {soabi} PyPy: " + msg)
            if env.MACOS:
                if not env.sys_is_gil_enabled():
                    pytest.xfail(f"[TEST-GIL-SCOPED] {soabi} with GIL disabled: " + msg)
                if env.PY_GIL_DISABLED:
                    pytest.xfail(f"[TEST-GIL-SCOPED] {soabi}: " + msg)
            raise RuntimeError(msg)
        return process.exitcode
    finally:
        if process.is_alive():
            process.terminate()


def _run_in_threads(test_fn, num_threads, parallel):
    threads = []
    for _ in range(num_threads):
        thread = threading.Thread(target=test_fn)
        thread.daemon = True
        thread.start()
        if parallel:
            threads.append(thread)
        else:
            thread.join()
    for thread in threads:
        thread.join()


@pytest.mark.skipif(sys.platform.startswith("emscripten"), reason="Requires threads")
@pytest.mark.parametrize("test_fn", ALL_BASIC_TESTS_PLUS_INTENTIONAL_DEADLOCK)
@pytest.mark.skipif(
    "env.GRAALPY",
    reason="GraalPy transiently complains about unfinished threads at process exit",
)
def test_run_in_process_one_thread(test_fn):
    """Makes sure there is no GIL deadlock when running in a thread.

    It runs in a separate process to be able to stop and assert if it deadlocks.
    """
    assert _run_in_process(_run_in_threads, test_fn, num_threads=1, parallel=False) == 0


@skipif_not_free_threaded
@pytest.mark.skipif(sys.platform.startswith("emscripten"), reason="Requires threads")
@pytest.mark.parametrize("test_fn", ALL_BASIC_TESTS_PLUS_INTENTIONAL_DEADLOCK)
@pytest.mark.skipif(
    "env.GRAALPY",
    reason="GraalPy transiently complains about unfinished threads at process exit",
)
def test_run_in_process_multiple_threads_parallel(test_fn):
    """Makes sure there is no GIL deadlock when running in a thread multiple times in parallel.

    It runs in a separate process to be able to stop and assert if it deadlocks.
    """
    assert _run_in_process(_run_in_threads, test_fn, num_threads=8, parallel=True) == 0


@pytest.mark.skipif(sys.platform.startswith("emscripten"), reason="Requires threads")
@pytest.mark.parametrize("test_fn", ALL_BASIC_TESTS_PLUS_INTENTIONAL_DEADLOCK)
@pytest.mark.skipif(
    "env.GRAALPY",
    reason="GraalPy transiently complains about unfinished threads at process exit",
)
def test_run_in_process_multiple_threads_sequential(test_fn):
    """Makes sure there is no GIL deadlock when running in a thread multiple times sequentially.

    It runs in a separate process to be able to stop and assert if it deadlocks.
    """
    assert _run_in_process(_run_in_threads, test_fn, num_threads=8, parallel=False) == 0


@pytest.mark.skipif(sys.platform.startswith("emscripten"), reason="Requires threads")
@pytest.mark.parametrize(
    "test_fn",
    [
        *ALL_BASIC_TESTS,
        pytest.param(_intentional_deadlock, marks=skipif_not_free_threaded),
    ],
)
@pytest.mark.skipif(
    "env.GRAALPY",
    reason="GraalPy transiently complains about unfinished threads at process exit",
)
def test_run_in_process_direct(test_fn):
    """Makes sure there is no GIL deadlock when using processes.

    This test is for completion, but it was never an issue.
    """
    assert _run_in_process(test_fn) == 0