(self)
| 228 | pass |
| 229 | |
| 230 | def test_timeout(self): |
| 231 | lock = self.locktype() |
| 232 | # Can't set timeout if not blocking |
| 233 | self.assertRaises(ValueError, lock.acquire, False, 1) |
| 234 | # Invalid timeout values |
| 235 | self.assertRaises(ValueError, lock.acquire, timeout=-100) |
| 236 | self.assertRaises(OverflowError, lock.acquire, timeout=1e100) |
| 237 | self.assertRaises(OverflowError, lock.acquire, timeout=TIMEOUT_MAX + 1) |
| 238 | # TIMEOUT_MAX is ok |
| 239 | lock.acquire(timeout=TIMEOUT_MAX) |
| 240 | lock.release() |
| 241 | t1 = time.monotonic() |
| 242 | self.assertTrue(lock.acquire(timeout=5)) |
| 243 | t2 = time.monotonic() |
| 244 | # Just a sanity test that it didn't actually wait for the timeout. |
| 245 | self.assertLess(t2 - t1, 5) |
| 246 | results = [] |
| 247 | def f(): |
| 248 | t1 = time.monotonic() |
| 249 | results.append(lock.acquire(timeout=0.5)) |
| 250 | t2 = time.monotonic() |
| 251 | results.append(t2 - t1) |
| 252 | with Bunch(f, 1): |
| 253 | pass |
| 254 | self.assertFalse(results[0]) |
| 255 | self.assertTimeout(results[1], 0.5) |
| 256 | |
| 257 | def test_weakref_exists(self): |
| 258 | lock = self.locktype() |
nothing calls this directly
no test coverage detected