(self)
| 270 | threading.stack_size(0) |
| 271 | |
| 272 | def test_foreign_thread(self): |
| 273 | # Check that a "foreign" thread can use the threading module. |
| 274 | dummy_thread = None |
| 275 | error = None |
| 276 | def f(mutex): |
| 277 | try: |
| 278 | nonlocal dummy_thread |
| 279 | nonlocal error |
| 280 | # Calling current_thread() forces an entry for the foreign |
| 281 | # thread to get made in the threading._active map. |
| 282 | dummy_thread = threading.current_thread() |
| 283 | tid = dummy_thread.ident |
| 284 | self.assertIn(tid, threading._active) |
| 285 | self.assertIsInstance(dummy_thread, threading._DummyThread) |
| 286 | self.assertIs(threading._active.get(tid), dummy_thread) |
| 287 | # gh-29376 |
| 288 | self.assertTrue( |
| 289 | dummy_thread.is_alive(), |
| 290 | 'Expected _DummyThread to be considered alive.' |
| 291 | ) |
| 292 | self.assertIn('_DummyThread', repr(dummy_thread)) |
| 293 | except BaseException as e: |
| 294 | error = e |
| 295 | finally: |
| 296 | mutex.release() |
| 297 | |
| 298 | mutex = threading.Lock() |
| 299 | mutex.acquire() |
| 300 | with threading_helper.wait_threads_exit(): |
| 301 | tid = _thread.start_new_thread(f, (mutex,)) |
| 302 | # Wait for the thread to finish. |
| 303 | mutex.acquire() |
| 304 | if error is not None: |
| 305 | raise error |
| 306 | self.assertEqual(tid, dummy_thread.ident) |
| 307 | # Issue gh-106236: |
| 308 | with self.assertRaises(RuntimeError): |
| 309 | dummy_thread.join() |
| 310 | dummy_thread._started.clear() |
| 311 | with self.assertRaises(RuntimeError): |
| 312 | dummy_thread.is_alive() |
| 313 | # Busy wait for the following condition: after the thread dies, the |
| 314 | # related dummy thread must be removed from threading._active. |
| 315 | timeout = 5 |
| 316 | timeout_at = time.monotonic() + timeout |
| 317 | while time.monotonic() < timeout_at: |
| 318 | if threading._active.get(dummy_thread.ident) is not dummy_thread: |
| 319 | break |
| 320 | time.sleep(.1) |
| 321 | else: |
| 322 | self.fail('It was expected that the created threading._DummyThread was removed from threading._active.') |
| 323 | |
| 324 | # PyThreadState_SetAsyncExc() is a CPython-only gimmick, not (currently) |
| 325 | # exposed at the Python level. This test relies on ctypes to get at it. |
nothing calls this directly
no test coverage detected