(self)
| 108 | |
| 109 | @gen_test |
| 110 | def test_get_timeout(self): |
| 111 | q = queues.Queue() # type: queues.Queue[int] |
| 112 | get_timeout = q.get(timeout=timedelta(seconds=0.01)) |
| 113 | get = q.get() |
| 114 | with self.assertRaises(TimeoutError): |
| 115 | yield get_timeout |
| 116 | |
| 117 | q.put_nowait(0) |
| 118 | self.assertEqual(0, (yield get)) |
| 119 | |
| 120 | @gen_test |
| 121 | def test_get_timeout_preempted(self): |
nothing calls this directly
no test coverage detected