Back to home page

EIC code displayed by LXR

 
 

    


File indexing completed on 2025-01-18 10:17:52

0001 import time
0002 from threading import Thread
0003 
0004 import pytest
0005 
0006 import env  # noqa: F401
0007 from pybind11_tests import callbacks as m
0008 
0009 
0010 def test_callbacks():
0011     from functools import partial
0012 
0013     def func1():
0014         return "func1"
0015 
0016     def func2(a, b, c, d):
0017         return "func2", a, b, c, d
0018 
0019     def func3(a):
0020         return f"func3({a})"
0021 
0022     assert m.test_callback1(func1) == "func1"
0023     assert m.test_callback2(func2) == ("func2", "Hello", "x", True, 5)
0024     assert m.test_callback1(partial(func2, 1, 2, 3, 4)) == ("func2", 1, 2, 3, 4)
0025     assert m.test_callback1(partial(func3, "partial")) == "func3(partial)"
0026     assert m.test_callback3(lambda i: i + 1) == "func(43) = 44"
0027 
0028     f = m.test_callback4()
0029     assert f(43) == 44
0030     f = m.test_callback5()
0031     assert f(number=43) == 44
0032 
0033 
0034 def test_bound_method_callback():
0035     # Bound Python method:
0036     class MyClass:
0037         def double(self, val):
0038             return 2 * val
0039 
0040     z = MyClass()
0041     assert m.test_callback3(z.double) == "func(43) = 86"
0042 
0043     z = m.CppBoundMethodTest()
0044     assert m.test_callback3(z.triple) == "func(43) = 129"
0045 
0046 
0047 def test_keyword_args_and_generalized_unpacking():
0048     def f(*args, **kwargs):
0049         return args, kwargs
0050 
0051     assert m.test_tuple_unpacking(f) == (("positional", 1, 2, 3, 4, 5, 6), {})
0052     assert m.test_dict_unpacking(f) == (
0053         ("positional", 1),
0054         {"key": "value", "a": 1, "b": 2},
0055     )
0056     assert m.test_keyword_args(f) == ((), {"x": 10, "y": 20})
0057     assert m.test_unpacking_and_keywords1(f) == ((1, 2), {"c": 3, "d": 4})
0058     assert m.test_unpacking_and_keywords2(f) == (
0059         ("positional", 1, 2, 3, 4, 5),
0060         {"key": "value", "a": 1, "b": 2, "c": 3, "d": 4, "e": 5},
0061     )
0062 
0063     with pytest.raises(TypeError) as excinfo:
0064         m.test_unpacking_error1(f)
0065     assert "Got multiple values for keyword argument" in str(excinfo.value)
0066 
0067     with pytest.raises(TypeError) as excinfo:
0068         m.test_unpacking_error2(f)
0069     assert "Got multiple values for keyword argument" in str(excinfo.value)
0070 
0071     with pytest.raises(RuntimeError) as excinfo:
0072         m.test_arg_conversion_error1(f)
0073     assert "Unable to convert call argument" in str(excinfo.value)
0074 
0075     with pytest.raises(RuntimeError) as excinfo:
0076         m.test_arg_conversion_error2(f)
0077     assert "Unable to convert call argument" in str(excinfo.value)
0078 
0079 
0080 def test_lambda_closure_cleanup():
0081     m.test_lambda_closure_cleanup()
0082     cstats = m.payload_cstats()
0083     assert cstats.alive() == 0
0084     assert cstats.copy_constructions == 1
0085     assert cstats.move_constructions >= 1
0086 
0087 
0088 def test_cpp_callable_cleanup():
0089     alive_counts = m.test_cpp_callable_cleanup()
0090     assert alive_counts == [0, 1, 2, 1, 2, 1, 0]
0091 
0092 
0093 def test_cpp_function_roundtrip():
0094     """Test if passing a function pointer from C++ -> Python -> C++ yields the original pointer"""
0095 
0096     assert (
0097         m.test_dummy_function(m.dummy_function) == "matches dummy_function: eval(1) = 2"
0098     )
0099     assert (
0100         m.test_dummy_function(m.roundtrip(m.dummy_function))
0101         == "matches dummy_function: eval(1) = 2"
0102     )
0103     assert (
0104         m.test_dummy_function(m.dummy_function_overloaded)
0105         == "matches dummy_function: eval(1) = 2"
0106     )
0107     assert m.roundtrip(None, expect_none=True) is None
0108     assert (
0109         m.test_dummy_function(lambda x: x + 2)
0110         == "can't convert to function pointer: eval(1) = 3"
0111     )
0112 
0113     with pytest.raises(TypeError) as excinfo:
0114         m.test_dummy_function(m.dummy_function2)
0115     assert "incompatible function arguments" in str(excinfo.value)
0116 
0117     with pytest.raises(TypeError) as excinfo:
0118         m.test_dummy_function(lambda x, y: x + y)
0119     assert any(
0120         s in str(excinfo.value)
0121         for s in ("missing 1 required positional argument", "takes exactly 2 arguments")
0122     )
0123 
0124 
0125 def test_function_signatures(doc):
0126     assert doc(m.test_callback3) == "test_callback3(arg0: Callable[[int], int]) -> str"
0127     assert doc(m.test_callback4) == "test_callback4() -> Callable[[int], int]"
0128 
0129 
0130 def test_movable_object():
0131     assert m.callback_with_movable(lambda _: None) is True
0132 
0133 
0134 @pytest.mark.skipif(
0135     "env.PYPY",
0136     reason="PyPy segfaults on here. See discussion on #1413.",
0137 )
0138 def test_python_builtins():
0139     """Test if python builtins like sum() can be used as callbacks"""
0140     assert m.test_sum_builtin(sum, [1, 2, 3]) == 6
0141     assert m.test_sum_builtin(sum, []) == 0
0142 
0143 
0144 def test_async_callbacks():
0145     # serves as state for async callback
0146     class Item:
0147         def __init__(self, value):
0148             self.value = value
0149 
0150     res = []
0151 
0152     # generate stateful lambda that will store result in `res`
0153     def gen_f():
0154         s = Item(3)
0155         return lambda j: res.append(s.value + j)
0156 
0157     # do some work async
0158     work = [1, 2, 3, 4]
0159     m.test_async_callback(gen_f(), work)
0160     # wait until work is done
0161     from time import sleep
0162 
0163     sleep(0.5)
0164     assert sum(res) == sum(x + 3 for x in work)
0165 
0166 
0167 def test_async_async_callbacks():
0168     t = Thread(target=test_async_callbacks)
0169     t.start()
0170     t.join()
0171 
0172 
0173 def test_callback_num_times():
0174     # Super-simple micro-benchmarking related to PR #2919.
0175     # Example runtimes (Intel Xeon 2.2GHz, fully optimized):
0176     #   num_millions  1, repeats  2:  0.1 secs
0177     #   num_millions 20, repeats 10: 11.5 secs
0178     one_million = 1000000
0179     num_millions = 1  # Try 20 for actual micro-benchmarking.
0180     repeats = 2  # Try 10.
0181     rates = []
0182     for rep in range(repeats):
0183         t0 = time.time()
0184         m.callback_num_times(lambda: None, num_millions * one_million)
0185         td = time.time() - t0
0186         rate = num_millions / td if td else 0
0187         rates.append(rate)
0188         if not rep:
0189             print()
0190         print(
0191             f"callback_num_times: {num_millions:d} million / {td:.3f} seconds = {rate:.3f} million / second"
0192         )
0193     if len(rates) > 1:
0194         print("Min    Mean   Max")
0195         print(f"{min(rates):6.3f} {sum(rates) / len(rates):6.3f} {max(rates):6.3f}")
0196 
0197 
0198 def test_custom_func():
0199     assert m.custom_function(4) == 36
0200     assert m.roundtrip(m.custom_function)(4) == 36
0201 
0202 
0203 @pytest.mark.skipif(
0204     m.custom_function2 is None, reason="Current PYBIND11_INTERNALS_VERSION too low"
0205 )
0206 def test_custom_func2():
0207     assert m.custom_function2(3) == 27
0208     assert m.roundtrip(m.custom_function2)(3) == 27