Method test__count

Lib/test/test_thread.py:118–151

(self)

        thread.stack_size(0)

    def test__count(self):
        # Test the _count() function.
        orig = thread._count()
        mut = thread.allocate_lock()
        mut.acquire()
        started = []

        def task():
            started.append(None)
            mut.acquire()
            mut.release()

        with threading_helper.wait_threads_exit():
            thread.start_new_thread(task, ())
            for _ in support.sleeping_retry(support.LONG_TIMEOUT):
                if started:
                    break
            self.assertEqual(thread._count(), orig + 1)

            # Allow the task to finish.
            mut.release()

            # The only reliable way to be sure that the thread ended from the
            # interpreter's point of view is to wait for the function object to
            # be destroyed.
            done = []
            wr = weakref.ref(task, lambda _: done.append(None))
            del task

            for _ in support.sleeping_retry(support.LONG_TIMEOUT):
                if done:
                    break
                support.gc_collect()  # For PyPy or other GCs.
            self.assertEqual(thread._count(), orig)

    def test_unraisable_exception(self):
        def task():