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