MCPcopy Index your code
hub / github.com/RustPython/RustPython / test_threads_join

Method test_threads_join

Lib/test/test_threading.py:1541–1567  ·  view source on GitHub ↗
(self)

Source from the content-addressed store, hash-verified

1539
1540 @unittest.expectedFailure # TODO: RUSTPYTHON
1541 def test_threads_join(self):
1542 # Non-daemon threads should be joined at subinterpreter shutdown
1543 # (issue #18808)
1544 r, w = self.pipe()
1545 code = textwrap.dedent(r"""
1546 import os
1547 import random
1548 import threading
1549 import time
1550
1551 def random_sleep():
1552 seconds = random.random() * 0.010
1553 time.sleep(seconds)
1554
1555 def f():
1556 # Sleep a bit so that the thread is still running when
1557 # Py_EndInterpreter is called.
1558 random_sleep()
1559 os.write(%d, b"x")
1560
1561 threading.Thread(target=f).start()
1562 random_sleep()
1563 """ % (w,))
1564 ret = test.support.run_in_subinterp(code)
1565 self.assertEqual(ret, 0)
1566 # The thread was joined properly.
1567 self.assertEqual(os.read(r, 1), b"x")
1568
1569 @unittest.expectedFailure # TODO: RUSTPYTHON
1570 def test_threads_join_2(self):

Callers

nothing calls this directly

Calls 4

pipeMethod · 0.95
dedentMethod · 0.80
assertEqualMethod · 0.45
readMethod · 0.45

Tested by

no test coverage detected