1import pytest
2from pybind11_tests import callbacks as m
3from threading import Thread
4
5
6def test_callbacks():
7    from functools import partial
8
9    def func1():
10        return "func1"
11
12    def func2(a, b, c, d):
13        return "func2", a, b, c, d
14
15    def func3(a):
16        return "func3({})".format(a)
17
18    assert m.test_callback1(func1) == "func1"
19    assert m.test_callback2(func2) == ("func2", "Hello", "x", True, 5)
20    assert m.test_callback1(partial(func2, 1, 2, 3, 4)) == ("func2", 1, 2, 3, 4)
21    assert m.test_callback1(partial(func3, "partial")) == "func3(partial)"
22    assert m.test_callback3(lambda i: i + 1) == "func(43) = 44"
23
24    f = m.test_callback4()
25    assert f(43) == 44
26    f = m.test_callback5()
27    assert f(number=43) == 44
28
29
30def test_bound_method_callback():
31    # Bound Python method:
32    class MyClass:
33        def double(self, val):
34            return 2 * val
35
36    z = MyClass()
37    assert m.test_callback3(z.double) == "func(43) = 86"
38
39    z = m.CppBoundMethodTest()
40    assert m.test_callback3(z.triple) == "func(43) = 129"
41
42
43def test_keyword_args_and_generalized_unpacking():
44
45    def f(*args, **kwargs):
46        return args, kwargs
47
48    assert m.test_tuple_unpacking(f) == (("positional", 1, 2, 3, 4, 5, 6), {})
49    assert m.test_dict_unpacking(f) == (("positional", 1), {"key": "value", "a": 1, "b": 2})
50    assert m.test_keyword_args(f) == ((), {"x": 10, "y": 20})
51    assert m.test_unpacking_and_keywords1(f) == ((1, 2), {"c": 3, "d": 4})
52    assert m.test_unpacking_and_keywords2(f) == (
53        ("positional", 1, 2, 3, 4, 5),
54        {"key": "value", "a": 1, "b": 2, "c": 3, "d": 4, "e": 5}
55    )
56
57    with pytest.raises(TypeError) as excinfo:
58        m.test_unpacking_error1(f)
59    assert "Got multiple values for keyword argument" in str(excinfo.value)
60
61    with pytest.raises(TypeError) as excinfo:
62        m.test_unpacking_error2(f)
63    assert "Got multiple values for keyword argument" in str(excinfo.value)
64
65    with pytest.raises(RuntimeError) as excinfo:
66        m.test_arg_conversion_error1(f)
67    assert "Unable to convert call argument" in str(excinfo.value)
68
69    with pytest.raises(RuntimeError) as excinfo:
70        m.test_arg_conversion_error2(f)
71    assert "Unable to convert call argument" in str(excinfo.value)
72
73
74def test_lambda_closure_cleanup():
75    m.test_cleanup()
76    cstats = m.payload_cstats()
77    assert cstats.alive() == 0
78    assert cstats.copy_constructions == 1
79    assert cstats.move_constructions >= 1
80
81
82def test_cpp_function_roundtrip():
83    """Test if passing a function pointer from C++ -> Python -> C++ yields the original pointer"""
84
85    assert m.test_dummy_function(m.dummy_function) == "matches dummy_function: eval(1) = 2"
86    assert (m.test_dummy_function(m.roundtrip(m.dummy_function)) ==
87            "matches dummy_function: eval(1) = 2")
88    assert m.roundtrip(None, expect_none=True) is None
89    assert (m.test_dummy_function(lambda x: x + 2) ==
90            "can't convert to function pointer: eval(1) = 3")
91
92    with pytest.raises(TypeError) as excinfo:
93        m.test_dummy_function(m.dummy_function2)
94    assert "incompatible function arguments" in str(excinfo.value)
95
96    with pytest.raises(TypeError) as excinfo:
97        m.test_dummy_function(lambda x, y: x + y)
98    assert any(s in str(excinfo.value) for s in ("missing 1 required positional argument",
99                                                 "takes exactly 2 arguments"))
100
101
102def test_function_signatures(doc):
103    assert doc(m.test_callback3) == "test_callback3(arg0: Callable[[int], int]) -> str"
104    assert doc(m.test_callback4) == "test_callback4() -> Callable[[int], int]"
105
106
107def test_movable_object():
108    assert m.callback_with_movable(lambda _: None) is True
109
110
111def test_async_callbacks():
112    # serves as state for async callback
113    class Item:
114        def __init__(self, value):
115            self.value = value
116
117    res = []
118
119    # generate stateful lambda that will store result in `res`
120    def gen_f():
121        s = Item(3)
122        return lambda j: res.append(s.value + j)
123
124    # do some work async
125    work = [1, 2, 3, 4]
126    m.test_async_callback(gen_f(), work)
127    # wait until work is done
128    from time import sleep
129    sleep(0.5)
130    assert sum(res) == sum([x + 3 for x in work])
131
132
133def test_async_async_callbacks():
134    t = Thread(target=test_async_callbacks)
135    t.start()
136    t.join()
137