(self)
| 63 | self.assertIs(self.lock.acquire(), True) |
| 64 | |
| 65 | def test_uncond_acquire_blocking(self): |
| 66 | #Make sure that unconditional acquiring of a locked lock blocks. |
| 67 | def delay_unlock(to_unlock, delay): |
| 68 | """Hold on to lock for a set amount of time before unlocking.""" |
| 69 | time.sleep(delay) |
| 70 | to_unlock.release() |
| 71 | |
| 72 | self.lock.acquire() |
| 73 | start_time = int(time.monotonic()) |
| 74 | _thread.start_new_thread(delay_unlock,(self.lock, DELAY)) |
| 75 | if support.verbose: |
| 76 | print() |
| 77 | print("*** Waiting for thread to release the lock "\ |
| 78 | "(approx. %s sec.) ***" % DELAY) |
| 79 | self.lock.acquire() |
| 80 | end_time = int(time.monotonic()) |
| 81 | if support.verbose: |
| 82 | print("done") |
| 83 | self.assertGreaterEqual(end_time - start_time, DELAY, |
| 84 | "Blocking by unconditional acquiring failed.") |
| 85 | |
| 86 | @mock.patch('time.sleep') |
| 87 | def test_acquire_timeout(self, mock_sleep): |
nothing calls this directly
no test coverage detected