(self, loop, debug)
| 336 | self.loop.call_at(None, cb) |
| 337 | |
| 338 | def check_thread(self, loop, debug): |
| 339 | def cb(): |
| 340 | pass |
| 341 | |
| 342 | loop.set_debug(debug) |
| 343 | if debug: |
| 344 | msg = ("Non-thread-safe operation invoked on an event loop other " |
| 345 | "than the current one") |
| 346 | with self.assertRaisesRegex(RuntimeError, msg): |
| 347 | loop.call_soon(cb) |
| 348 | with self.assertRaisesRegex(RuntimeError, msg): |
| 349 | loop.call_later(60, cb) |
| 350 | with self.assertRaisesRegex(RuntimeError, msg): |
| 351 | loop.call_at(loop.time() + 60, cb) |
| 352 | else: |
| 353 | loop.call_soon(cb) |
| 354 | loop.call_later(60, cb) |
| 355 | loop.call_at(loop.time() + 60, cb) |
| 356 | |
| 357 | def test_check_thread(self): |
| 358 | def check_in_thread(loop, event, debug, create_loop, fut): |
no test coverage detected