Repository: RustPython/RustPython
File: Lib/test/test_threading.py (Python)

Code before:
            tid = _thread.start_new_thread(f, (mutex,))
            # Wait for the thread to finish.
            mutex.acquire()
        if error is not None:
            raise error
        self.assertEqual(tid, dummy_thread.ident)
        # Issue gh-106236:
        with self.assertRaises(RuntimeError):
            dummy_thread.join()
        dummy_thread._started.clear()
        with self.assertRaises(RuntimeError):
            dummy_thread.is_alive()
        # Busy wait for the following condition: after the thread dies, the
        # related dummy thread must be removed from threading._active.
        timeout = 5
        timeout_at = time.monotonic() + timeout
        while time.monotonic() < timeout_at:
            if threading._active.get(dummy_thread.ident) is not dummy_thread:
                break
            time.sleep(.1)
        else:
            self.fail('It was expected that the created threading._DummyThread was removed from threading._active.')

    # PyThreadState_SetAsyncExc() is a CPython-only gimmick, not (currently)
    # exposed at the Python level.  This test relies on ctypes to get at it.
    @unittest.expectedFailure # TODO: RUSTPYTHON; expects @cpython_only
    def test_PyThreadState_SetAsyncExc(self):
        ctypes = import_module("ctypes")

        set_async_exc = ctypes.pythonapi.PyThreadState_SetAsyncExc
        set_async_exc.argtypes = (ctypes.c_ulong, ctypes.py_object)

        class AsyncExc(Exception):
            pass

        exception = ctypes.py_object(AsyncExc)

        # First check it works when setting the exception from the same thread.
        tid = threading.get_ident()
        self.assertIsInstance(tid, int)
        self.assertGreater(tid, 0)

        try:
            result = set_async_exc(tid, exception)
            # The exception is async, so we might have to keep the VM busy until
            # it notices.
            while True:
                pass
        except AsyncExc:
            pass
        else:

Review comment (suggestion):
This has already been fixed upstream in python/cpython#143002, so it may be reasonable to do the same here:
```suggestion
    @cpython_only
```

Code after:
            tid = _thread.start_new_thread(f, (mutex,))
            # Wait for the thread to finish.
            mutex.acquire()
        if error is not None:
            raise error
        self.assertEqual(tid, dummy_thread.ident)
        # Issue gh-106236:
        with self.assertRaises(RuntimeError):
            dummy_thread.join()
        dummy_thread._started.clear()
        with self.assertRaises(RuntimeError):
            dummy_thread.is_alive()
        # Busy wait for the following condition: after the thread dies, the
        # related dummy thread must be removed from threading._active.
        timeout = 5
        timeout_at = time.monotonic() + timeout
        while time.monotonic() < timeout_at:
            if threading._active.get(dummy_thread.ident) is not dummy_thread:
                break
            time.sleep(.1)
        else:
            self.fail('It was expected that the created threading._DummyThread was removed from threading._active.')

    # PyThreadState_SetAsyncExc() is a CPython-only gimmick, not (currently)
    # exposed at the Python level.  This test relies on ctypes to get at it.
    @cpython_only
    def test_PyThreadState_SetAsyncExc(self):
        ctypes = import_module("ctypes")

        set_async_exc = ctypes.pythonapi.PyThreadState_SetAsyncExc
        set_async_exc.argtypes = (ctypes.c_ulong, ctypes.py_object)

        class AsyncExc(Exception):
            pass

        exception = ctypes.py_object(AsyncExc)

        # First check it works when setting the exception from the same thread.
        tid = threading.get_ident()
        self.assertIsInstance(tid, int)
        self.assertGreater(tid, 0)

        try:
            result = set_async_exc(tid, exception)
            # The exception is async, so we might have to keep the VM busy until
            # it notices.
            while True:
                pass
        except AsyncExc:
            pass
        else: