pypy/module/cpyext/test/test_pystate.py::AppTestThreads::()::test_thread_interrupt
self = <CallInfo when='call' exception: assert False>
func = <function <lambda> at 0x0c594510>, when = 'call'
def __init__(self, func, when):
#: context of invocation: one of "setup", "call",
#: "teardown", "memocollect"
self.when = when
self.start = time()
try:
> self.result = func()
../_pytest/runner.py:150:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
> return CallInfo(lambda: ihook(item=item, **kwds), when=when)
../_pytest/runner.py:138:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
self = <_HookCaller 'pytest_runtest_call'>
kwargs = {'__multicall__': <_MultiCall 0 results, 0 meths, kwargs={'item': <AppTestMethod 'test_thread_interrupt'>, '__multicall__': <_MultiCall 0 results, 0 meths, kwargs={...}>}>, 'item': <AppTestMethod 'test_thread_interrupt'>}
def __call__(self, **kwargs):
assert not self.is_historic()
> return self._hookexec(self, self._nonwrappers + self._wrappers, kwargs)
../_pytest/vendored_packages/pluggy.py:724:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
self = <_pytest.config.PytestPluginManager object at 0xf4809ef0>
hook = <_HookCaller 'pytest_runtest_call'>, methods = []
kwargs = {'__multicall__': <_MultiCall 0 results, 0 meths, kwargs={'item': <AppTestMethod 'test_thread_interrupt'>, '__multicall__': <_MultiCall 0 results, 0 meths, kwargs={...}>}>, 'item': <AppTestMethod 'test_thread_interrupt'>}
def _hookexec(self, hook, methods, kwargs):
# called from all hookcaller instances.
# enable_tracing will set its own wrapping function at self._inner_hookexec
> return self._inner_hookexec(hook, methods, kwargs)
../_pytest/vendored_packages/pluggy.py:338:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
hook = <_HookCaller 'pytest_runtest_call'>, methods = []
kwargs = {'__multicall__': <_MultiCall 0 results, 0 meths, kwargs={'item': <AppTestMethod 'test_thread_interrupt'>, '__multicall__': <_MultiCall 0 results, 0 meths, kwargs={...}>}>, 'item': <AppTestMethod 'test_thread_interrupt'>}
self._inner_hookexec = lambda hook, methods, kwargs: \
> _MultiCall(methods, kwargs, hook.spec_opts).execute()
../_pytest/vendored_packages/pluggy.py:333:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
self = <_MultiCall 0 results, 0 meths, kwargs={'item': <AppTestMethod 'test_thread_interrupt'>, '__multicall__': <_MultiCall 0 results, 0 meths, kwargs={...}>}>
def execute(self):
all_kwargs = self.kwargs
self.results = results = []
firstresult = self.specopts.get("firstresult")
while self.hook_impls:
hook_impl = self.hook_impls.pop()
args = [all_kwargs[argname] for argname in hook_impl.argnames]
if hook_impl.hookwrapper:
> return _wrapped_call(hook_impl.function(*args), self.execute)
../_pytest/vendored_packages/pluggy.py:595:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
wrap_controller = <generator object pytest_runtest_call at 0x0c89bc40>
func = <bound method _MultiCall.execute of <_MultiCall 0 results, 0 meths, kwargs={'i..._interrupt'>, '__multicall__': <_MultiCall 0 results, 0 meths, kwargs={...}>}>>
def _wrapped_call(wrap_controller, func):
""" Wrap calling to a function with a generator which needs to yield
exactly once. The yield point will trigger calling the wrapped function
and return its _CallOutcome to the yield point. The generator then needs
to finish (raise StopIteration) in order for the wrapped call to complete.
"""
try:
next(wrap_controller) # first yield
except StopIteration:
_raise_wrapfail(wrap_controller, "did not yield")
call_outcome = _CallOutcome(func)
try:
wrap_controller.send(call_outcome)
_raise_wrapfail(wrap_controller, "has second yield")
except StopIteration:
pass
> return call_outcome.get_result()
../_pytest/vendored_packages/pluggy.py:253:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
self = <_pytest.vendored_packages.pluggy._CallOutcome instance at 0x0cf60f64>
def get_result(self):
if self.excinfo is None:
return self.result
else:
ex = self.excinfo
if _py3:
raise ex[1].with_traceback(ex[2])
> _reraise(*ex) # noqa
../_pytest/vendored_packages/pluggy.py:279:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
self = <_pytest.vendored_packages.pluggy._CallOutcome instance at 0x0cf60f64>
func = <bound method _MultiCall.execute of <_MultiCall 0 results, 0 meths, kwargs={'i..._interrupt'>, '__multicall__': <_MultiCall 0 results, 0 meths, kwargs={...}>}>>
def __init__(self, func):
try:
> self.result = func()
../_pytest/vendored_packages/pluggy.py:264:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
self = <_MultiCall 0 results, 0 meths, kwargs={'item': <AppTestMethod 'test_thread_interrupt'>, '__multicall__': <_MultiCall 0 results, 0 meths, kwargs={...}>}>
def execute(self):
all_kwargs = self.kwargs
self.results = results = []
firstresult = self.specopts.get("firstresult")
while self.hook_impls:
hook_impl = self.hook_impls.pop()
args = [all_kwargs[argname] for argname in hook_impl.argnames]
if hook_impl.hookwrapper:
return _wrapped_call(hook_impl.function(*args), self.execute)
> res = hook_impl.function(*args)
../_pytest/vendored_packages/pluggy.py:596:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
item = <AppTestMethod 'test_thread_interrupt'>
def pytest_runtest_call(item):
try:
> item.runtest()
../_pytest/runner.py:91:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
self = <AppTestMethod 'test_thread_interrupt'>
def runtest(self):
target = self.obj
if self.config.option.runappdirect:
> return target()
tool/pytest/apptest.py:84:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
self = <pypy.module.cpyext.test.test_pystate.AppTestThreads object at 0x0bd736b0>
def test_thread_interrupt(self):
import signal # to initialize the signal infrastructure
import thread, time, posix as os
module = self.import_extension('disruptor', [
("surprise", "METH_VARARGS",
"""
long tid;
PyObject *exc;
if (!PyArg_ParseTuple(args, "lO", &tid, &exc))
return NULL;
return PyInt_FromLong(PyThreadState_SetAsyncExc(tid, exc));
"""),
("nosurprise", "METH_VARARGS",
"""
long tid;
if (!PyArg_ParseTuple(args, "l", &tid))
return NULL;
return PyInt_FromLong(PyThreadState_SetAsyncExc(tid, NULL));
"""),
])
# raise in another thread
def f():
try:
childstarted.append(thread.get_ident())
while 1:
print "waiting in child"
time.sleep(0.1)
except KeyError:
childstarted.pop()
else:
thread.interrupt_main() # to crash the test
cancelled_exception = False
for i in range(20):
print i
# first wait for the child to start
childstarted = []
thread.start_new_thread(f, ())
for i in range(500):
if childstarted:
break
time.sleep(0.1)
# then interrupt it from the main thread
res = module.surprise(childstarted[0], KeyError)
assert res == 1
# try to cancel the interrupt
# this is racing! the child could either have been terminated
# already or not
try:
res = module.nosurprise(childstarted[0])
assert res in (0, 1)
except IndexError:
pass
else:
if res == 1:
cancelled_exception = True
# actually send exception
res = module.surprise(childstarted[0], KeyError)
assert res in (0, 1)
# now wait for the exception to arrive in the child thread
# which empties childstarted again
for i in range(500):
if not childstarted:
break
time.sleep(0.1)
assert res == 1
> assert cancelled_exception # XXX I have no clue whether we can expect this to work or not
E assert False
module/cpyext/test/test_pystate.py:286: AssertionError
---------- Captured stdout call ----------
0
waiting in child
1
waiting in child
2
waiting in child
3
waiting in child
4
waiting in child
5
waiting in child
6
waiting in child
7
waiting in child
8
waiting in child
9
waiting in child
10
waiting in child
11
waiting in child
12
waiting in child
13
waiting in child
14
waiting in child
15
waiting in child
16
waiting in child
17
waiting in child
18
waiting in child
waiting in child
19
waiting in child
builder: pypy-c-jit-linux-x86-32 build #8326+
test: pypy/module/cpyext/test/test_pystate.py::AppTestThreads::()::test_thread_interrupt