from _pydevd_bundle import pydevd_timeout
from _pydevd_bundle.pydevd_constants import IS_PYPY, IS_CPYTHON
from _pydevd_bundle.pydevd_timeout import create_interrupt_this_thread_callback
from tests_python.debugger_unittest import wait_for_condition
import pytest
import threading
import time


@pytest.yield_fixture(autouse=True)
def _enable_debug_msgs():
    original = pydevd_timeout._DEBUG
    pydevd_timeout._DEBUG = True
    yield
    pydevd_timeout._DEBUG = original


def test_timeout():
    class _DummyPyDb(object):
        def __init__(self):
            self.created_pydb_daemon_threads = {}

    raise_timeout = pydevd_timeout.create_interrupt_this_thread_callback()

    def on_timeout(arg):
        assert arg == 1
        raise_timeout()

    py_db = _DummyPyDb()
    timeout_tracker = pydevd_timeout.TimeoutTracker(py_db)
    try:
        if IS_PYPY:
            timeout = 2
        else:
            timeout = 20
        with timeout_tracker.call_on_timeout(1, on_timeout, kwargs={"arg": 1}):
            time.sleep(timeout)
    except KeyboardInterrupt:
        pass


def test_timeout_0_time():
    class _DummyPyDb(object):
        def __init__(self):
            self.created_pydb_daemon_threads = {}

    raise_timeout = pydevd_timeout.create_interrupt_this_thread_callback()

    def on_timeout(arg):
        assert arg == 1
        raise_timeout()

    py_db = _DummyPyDb()
    timeout_tracker = pydevd_timeout.TimeoutTracker(py_db)
    try:
        with timeout_tracker.call_on_timeout(0, on_timeout, kwargs={"arg": 1}):
            time.sleep(1)
    except KeyboardInterrupt:
        pass


@pytest.mark.skipif(not IS_CPYTHON, reason="This only works in CPython.")
def test_create_interrupt_this_thread_callback():
    class MyThread(threading.Thread):
        def __init__(self):
            threading.Thread.__init__(self)
            self.finished = False
            self.daemon = True
            self.interrupt_thread = None
            self.interrupted = False

        def run(self):
            try:
                self.interrupt_thread = create_interrupt_this_thread_callback()
                while True:
                    time.sleep(0.2)
            except KeyboardInterrupt:
                self.interrupted = True
            finally:
                self.finished = True

    t = MyThread()
    t.start()
    wait_for_condition(lambda: t.interrupt_thread is not None)

    t.interrupt_thread()

    wait_for_condition(lambda: t.finished)

    assert t.interrupted


@pytest.mark.skipif(True, reason="Skipping because running this test can interrupt the test suite execution.")
def test_interrupt_main_thread():
    from _pydevd_bundle.pydevd_constants import IS_PY39_OR_GREATER

    class MyThread(threading.Thread):
        def __init__(self, interrupt_thread_callback):
            threading.Thread.__init__(self)
            self.interrupt_thread_callback = interrupt_thread_callback

        def run(self):
            time.sleep(0.5)
            self.interrupt_thread_callback()

    initial_time = time.time()
    interrupt_only_on_callback = IS_PYPY or IS_PY39_OR_GREATER
    if interrupt_only_on_callback:
        timeout = 2
    else:
        timeout = 20
    try:
        t = MyThread(create_interrupt_this_thread_callback())
        t.start()
        time.sleep(timeout)
    except KeyboardInterrupt:
        if not interrupt_only_on_callback:
            assert time.time() - initial_time < timeout
    else:
        raise AssertionError("Expected main thread to be interrupted.")