(self)
| 168 | lock.release() |
| 169 | |
| 170 | def test_acquire_contended(self): |
| 171 | lock = self.locktype() |
| 172 | lock.acquire() |
| 173 | def f(): |
| 174 | lock.acquire() |
| 175 | lock.release() |
| 176 | |
| 177 | N = 5 |
| 178 | with Bunch(f, N) as bunch: |
| 179 | # Threads block on lock.acquire() |
| 180 | wait_threads_blocked(N) |
| 181 | self.assertEqual(len(bunch.finished), 0) |
| 182 | |
| 183 | # Threads unblocked |
| 184 | lock.release() |
| 185 | |
| 186 | self.assertEqual(len(bunch.finished), N) |
| 187 | |
| 188 | def test_with(self): |
| 189 | lock = self.locktype() |
nothing calls this directly
no test coverage detected