import pytest

from pybind11_tests import callbacks as m

from threading import Thread

def test_callbacks():

    from functools import partial

    def func1():

        return "func1"

    def func2(a, b, c, d):

        return "func2", a, b, c, d

    def func3(a):

        return "func3({})".format(a)

    assert m.test_callback1(func1) == "func1"

    assert m.test_callback2(func2) == ("func2", "Hello", "x", True, 5)

    assert m.test_callback1(partial(func2, 1, 2, 3, 4)) == ("func2", 1, 2, 3, 4)

    assert m.test_callback1(partial(func3, "partial")) == "func3(partial)"

    assert m.test_callback3(lambda i: i + 1) == "func(43) = 44"

    f = m.test_callback4()

    assert f(43) == 44

    f = m.test_callback5()

    assert f(number=43) == 44

def test_bound_method_callback():

    # Bound Python method:

    class MyClass:

        def double(self, val):

            return 2 * val

    z = MyClass()

    assert m.test_callback3(z.double) == "func(43) = 86"

    z = m.CppBoundMethodTest()

    assert m.test_callback3(z.triple) == "func(43) = 129"

def test_keyword_args_and_generalized_unpacking():

    def f(*args, **kwargs):

        return args, kwargs

    assert m.test_tuple_unpacking(f) == (("positional", 1, 2, 3, 4, 5, 6), {})

    assert m.test_dict_unpacking(f) == (("positional", 1), {"key": "value", "a": 1, "b": 2})

    assert m.test_keyword_args(f) == ((), {"x": 10, "y": 20})

    assert m.test_unpacking_and_keywords1(f) == ((1, 2), {"c": 3, "d": 4})

    assert m.test_unpacking_and_keywords2(f) == (

        ("positional", 1, 2, 3, 4, 5),

        {"key": "value", "a": 1, "b": 2, "c": 3, "d": 4, "e": 5}

    )

    with pytest.raises(TypeError) as excinfo:

        m.test_unpacking_error1(f)

    assert "Got multiple values for keyword argument" in str(excinfo.value)

    with pytest.raises(TypeError) as excinfo:

        m.test_unpacking_error2(f)

    assert "Got multiple values for keyword argument" in str(excinfo.value)

    with pytest.raises(RuntimeError) as excinfo:

        m.test_arg_conversion_error1(f)

    assert "Unable to convert call argument" in str(excinfo.value)

    with pytest.raises(RuntimeError) as excinfo:

        m.test_arg_conversion_error2(f)

    assert "Unable to convert call argument" in str(excinfo.value)

def test_lambda_closure_cleanup():

    m.test_cleanup()

    cstats = m.payload_cstats()

    assert cstats.alive() == 0

    assert cstats.copy_constructions == 1

    assert cstats.move_constructions >= 1

def test_cpp_function_roundtrip():

    """Test if passing a function pointer from C++ -> Python -> C++ yields the original pointer"""

    assert m.test_dummy_function(m.dummy_function) == "matches dummy_function: eval(1) = 2"

    assert (m.test_dummy_function(m.roundtrip(m.dummy_function)) ==

            "matches dummy_function: eval(1) = 2")

    assert m.roundtrip(None, expect_none=True) is None

    assert (m.test_dummy_function(lambda x: x + 2) ==

            "can't convert to function pointer: eval(1) = 3")

    with pytest.raises(TypeError) as excinfo:

        m.test_dummy_function(m.dummy_function2)

    assert "incompatible function arguments" in str(excinfo.value)

    with pytest.raises(TypeError) as excinfo:

        m.test_dummy_function(lambda x, y: x + y)

    assert any(s in str(excinfo.value) for s in ("missing 1 required positional argument",

                                                 "takes exactly 2 arguments"))

def test_function_signatures(doc):

    assert doc(m.test_callback3) == "test_callback3(arg0: Callable[[int], int]) -> str"

    assert doc(m.test_callback4) == "test_callback4() -> Callable[[int], int]"

def test_movable_object():

    assert m.callback_with_movable(lambda _: None) is True

def test_async_callbacks():

    # serves as state for async callback

    class Item:

        def __init__(self, value):

            self.value = value

    res = []

    # generate stateful lambda that will store result in `res`

    def gen_f():

        s = Item(3)

        return lambda j: res.append(s.value + j)

    # do some work async

    work = [1, 2, 3, 4]

    m.test_async_callback(gen_f(), work)

    # wait until work is done

    from time import sleep

    sleep(0.5)

    assert sum(res) == sum([x + 3 for x in work])

def test_async_async_callbacks():

    t = Thread(target=test_async_callbacks)

    t.start()

    t.join()