1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
|
import math
import textwrap
import sys
import pytest
import threading
import traceback
import time
import random
import numpy as np
from numpy.testing import assert_, assert_equal, IS_PYPY
from . import util
class TestF77Callback(util.F2PyTest):
code = """
subroutine t(fun,a)
integer a
cf2py intent(out) a
external fun
call fun(a)
end
subroutine func(a)
cf2py intent(in,out) a
integer a
a = a + 11
end
subroutine func0(a)
cf2py intent(out) a
integer a
a = 11
end
subroutine t2(a)
cf2py intent(callback) fun
integer a
cf2py intent(out) a
external fun
call fun(a)
end
subroutine string_callback(callback, a)
external callback
double precision callback
double precision a
character*1 r
cf2py intent(out) a
r = 'r'
a = callback(r)
end
subroutine string_callback_array(callback, cu, lencu, a)
external callback
integer callback
integer lencu
character*8 cu(lencu)
integer a
cf2py intent(out) a
a = callback(cu, lencu)
end
"""
@pytest.mark.parametrize('name', 't,t2'.split(','))
def test_all(self, name):
self.check_function(name)
@pytest.mark.xfail(IS_PYPY,
reason="PyPy cannot modify tp_doc after PyType_Ready")
def test_docstring(self):
expected = textwrap.dedent("""\
a = t(fun,[fun_extra_args])
Wrapper for ``t``.
Parameters
----------
fun : call-back function
Other Parameters
----------------
fun_extra_args : input tuple, optional
Default: ()
Returns
-------
a : int
Notes
-----
Call-back functions::
def fun(): return a
Return objects:
a : int
""")
assert_equal(self.module.t.__doc__, expected)
def check_function(self, name):
t = getattr(self.module, name)
r = t(lambda: 4)
assert_(r == 4, repr(r))
r = t(lambda a: 5, fun_extra_args=(6,))
assert_(r == 5, repr(r))
r = t(lambda a: a, fun_extra_args=(6,))
assert_(r == 6, repr(r))
r = t(lambda a: 5 + a, fun_extra_args=(7,))
assert_(r == 12, repr(r))
r = t(lambda a: math.degrees(a), fun_extra_args=(math.pi,))
assert_(r == 180, repr(r))
r = t(math.degrees, fun_extra_args=(math.pi,))
assert_(r == 180, repr(r))
r = t(self.module.func, fun_extra_args=(6,))
assert_(r == 17, repr(r))
r = t(self.module.func0)
assert_(r == 11, repr(r))
r = t(self.module.func0._cpointer)
assert_(r == 11, repr(r))
class A:
def __call__(self):
return 7
def mth(self):
return 9
a = A()
r = t(a)
assert_(r == 7, repr(r))
r = t(a.mth)
assert_(r == 9, repr(r))
@pytest.mark.skipif(sys.platform=='win32',
reason='Fails with MinGW64 Gfortran (Issue #9673)')
def test_string_callback(self):
def callback(code):
if code == 'r':
return 0
else:
return 1
f = getattr(self.module, 'string_callback')
r = f(callback)
assert_(r == 0, repr(r))
@pytest.mark.skipif(sys.platform=='win32',
reason='Fails with MinGW64 Gfortran (Issue #9673)')
def test_string_callback_array(self):
# See gh-10027
cu = np.zeros((1, 8), 'S1')
def callback(cu, lencu):
if cu.shape != (lencu, 8):
return 1
if cu.dtype != 'S1':
return 2
if not np.all(cu == b''):
return 3
return 0
f = getattr(self.module, 'string_callback_array')
res = f(callback, cu, len(cu))
assert_(res == 0, repr(res))
def test_threadsafety(self):
# Segfaults if the callback handling is not threadsafe
errors = []
def cb():
# Sleep here to make it more likely for another thread
# to call their callback at the same time.
time.sleep(1e-3)
# Check reentrancy
r = self.module.t(lambda: 123)
assert_(r == 123)
return 42
def runner(name):
try:
for j in range(50):
r = self.module.t(cb)
assert_(r == 42)
self.check_function(name)
except Exception:
errors.append(traceback.format_exc())
threads = [threading.Thread(target=runner, args=(arg,))
for arg in ("t", "t2") for n in range(20)]
for t in threads:
t.start()
for t in threads:
t.join()
errors = "\n\n".join(errors)
if errors:
raise AssertionError(errors)
class TestF77CallbackPythonTLS(TestF77Callback):
"""
Callback tests using Python thread-local storage instead of
compiler-provided
"""
options = ["-DF2PY_USE_PYTHON_TLS"]
|
