| 106 | thread.stack_size(0) |
| 107 | |
| 108 | def test__count(self): |
| 109 | # Test the _count() function. |
| 110 | orig = thread._count() |
| 111 | mut = thread.allocate_lock() |
| 112 | mut.acquire() |
| 113 | started = [] |
| 114 | |
| 115 | def task(): |
| 116 | started.append(None) |
| 117 | mut.acquire() |
| 118 | mut.release() |
| 119 | |
| 120 | with threading_helper.wait_threads_exit(): |
| 121 | thread.start_new_thread(task, ()) |
| 122 | while not started: |
| 123 | time.sleep(POLL_SLEEP) |
| 124 | self.assertEqual(thread._count(), orig + 1) |
| 125 | # Allow the task to finish. |
| 126 | mut.release() |
| 127 | # The only reliable way to be sure that the thread ended from the |
| 128 | # interpreter's point of view is to wait for the function object to be |
| 129 | # destroyed. |
| 130 | done = [] |
| 131 | wr = weakref.ref(task, lambda _: done.append(None)) |
| 132 | del task |
| 133 | while not done: |
| 134 | time.sleep(POLL_SLEEP) |
| 135 | support.gc_collect() # For PyPy or other GCs. |
| 136 | self.assertEqual(thread._count(), orig) |
| 137 | |
| 138 | def test_unraisable_exception(self): |
| 139 | def task(): |