Method test_acquire_contended

Lib/test/lock_tests.py:166–182

(self)

    @unittest.skip("TODO: RUSTPYTHON; sometimes hangs")
    def test_acquire_contended(self):
        lock = self.locktype()
        lock.acquire()
        def f():
            lock.acquire()
            lock.release()

        N = 5
        with Bunch(f, N) as bunch:
            # Threads block on lock.acquire()
            wait_threads_blocked(N)
            self.assertEqual(len(bunch.finished), 0)

            # Threads unblocked
            lock.release()

        self.assertEqual(len(bunch.finished), N)

    def test_with(self):
        lock = self.locktype()