pypy/module/cpyext/test/test_pystate.py::AppTestThreads::()::test_thread_interrupt

self = <pypy.interpreter.typedef.W_ObjectObjectUserDictWeakrefable object at 0x7fed987b0450>

    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           (application-level) AssertionError: assert False

[/build_dir/own-linux-x86-64/build/pypy/module/cpyext/test/test_pystate.py:218]:69: AssertionError
---------- Captured stdout call ----------
NOTE: invoke_after_thread_switch() is meant to be translated 
and not called directly.  Using some emulation.
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
19
waiting in child
---------- Captured stderr call ----------
<function _Py_InitPyPyModule at 0x7fed98c94550>  DONE
<function PyInt_AsLong at 0x7fed98c40250>  DONE
<function PyThreadState_SetAsyncExc at 0x7fed98a8a1d0>  DONE
<function PyInt_AsLong at 0x7fed98c40250>  DONE
<function PyThreadState_SetAsyncExc at 0x7fed98a8a1d0>  DONE
<function PyInt_AsLong at 0x7fed98c40250>  DONE
<function PyThreadState_SetAsyncExc at 0x7fed98a8a1d0>  DONE
<function PyInt_AsLong at 0x7fed98c40250>  DONE
<function PyThreadState_SetAsyncExc at 0x7fed98a8a1d0>  DONE
<function PyInt_AsLong at 0x7fed98c40250>  DONE
<function PyThreadState_SetAsyncExc at 0x7fed98a8a1d0>  DONE
<function PyInt_AsLong at 0x7fed98c40250>  DONE
<function PyThreadState_SetAsyncExc at 0x7fed98a8a1d0>  DONE
<function PyInt_AsLong at 0x7fed98c40250>  DONE
<function PyThreadState_SetAsyncExc at 0x7fed98a8a1d0>  DONE
<function PyInt_AsLong at 0x7fed98c40250>  DONE
<function PyThreadState_SetAsyncExc at 0x7fed98a8a1d0>  DONE
<function PyInt_AsLong at 0x7fed98c40250>  DONE
<function PyThreadState_SetAsyncExc at 0x7fed98a8a1d0>  DONE
<function PyInt_AsLong at 0x7fed98c40250>  DONE
<function PyThreadState_SetAsyncExc at 0x7fed98a8a1d0>  DONE
<function PyInt_AsLong at 0x7fed98c40250>  DONE
<function PyThreadState_SetAsyncExc at 0x7fed98a8a1d0>  DONE
<function PyInt_AsLong at 0x7fed98c40250>  DONE
<function PyThreadState_SetAsyncExc at 0x7fed98a8a1d0>  DONE
<function PyInt_AsLong at 0x7fed98c40250>  DONE
<function PyThreadState_SetAsyncExc at 0x7fed98a8a1d0>  DONE
<function PyInt_AsLong at 0x7fed98c40250>  DONE
<function PyThreadState_SetAsyncExc at 0x7fed98a8a1d0>  DONE
<function PyInt_AsLong at 0x7fed98c40250>  DONE
<function PyThreadState_SetAsyncExc at 0x7fed98a8a1d0>  DONE
<function PyInt_AsLong at 0x7fed98c40250>  DONE
<function PyThreadState_SetAsyncExc at 0x7fed98a8a1d0>  DONE
<function PyInt_AsLong at 0x7fed98c40250>  DONE
<function PyThreadState_SetAsyncExc at 0x7fed98a8a1d0>  DONE
<function PyInt_AsLong at 0x7fed98c40250>  DONE
<function PyThreadState_SetAsyncExc at 0x7fed98a8a1d0>  DONE
<function PyInt_AsLong at 0x7fed98c40250>  DONE
<function PyThreadState_SetAsyncExc at 0x7fed98a8a1d0>  DONE
<function PyInt_AsLong at 0x7fed98c40250>  DONE
<function PyThreadState_SetAsyncExc at 0x7fed98a8a1d0>  DONE

builder: own-linux-x86-64 build #10816

test: pypy/module/cpyext/test/test_pystate/py/AppTestThreads/()/test_thread_interrupt