(self)
| 1235 | b.wait() |
| 1236 | |
| 1237 | def test_repr(self): |
| 1238 | barrier = self.barriertype(3) |
| 1239 | timeout = support.LONG_TIMEOUT |
| 1240 | self.assertRegex(repr(barrier), r"<\w+\.Barrier at .*: waiters=0/3>") |
| 1241 | def f(): |
| 1242 | barrier.wait(timeout) |
| 1243 | |
| 1244 | N = 2 |
| 1245 | with Bunch(f, N): |
| 1246 | # Threads blocked on barrier.wait() |
| 1247 | for _ in support.sleeping_retry(support.SHORT_TIMEOUT): |
| 1248 | if barrier.n_waiting >= N: |
| 1249 | break |
| 1250 | self.assertRegex(repr(barrier), |
| 1251 | r"<\w+\.Barrier at .*: waiters=2/3>") |
| 1252 | |
| 1253 | # Threads unblocked |
| 1254 | barrier.wait(timeout) |
| 1255 | |
| 1256 | self.assertRegex(repr(barrier), |
| 1257 | r"<\w+\.Barrier at .*: waiters=0/3>") |
| 1258 | |
| 1259 | # Abort the barrier |
| 1260 | barrier.abort() |
| 1261 | self.assertRegex(repr(barrier), |
| 1262 | r"<\w+\.Barrier at .*: broken>") |
nothing calls this directly
no test coverage detected