(self)
| 1196 | |
| 1197 | @warnings_helper.ignore_fork_in_thread_deprecation_warnings() |
| 1198 | def test_put(self): |
| 1199 | MAXSIZE = 6 |
| 1200 | queue = self.Queue(maxsize=MAXSIZE) |
| 1201 | child_can_start = self.Event() |
| 1202 | parent_can_continue = self.Event() |
| 1203 | |
| 1204 | proc = self.Process( |
| 1205 | target=self._test_put, |
| 1206 | args=(queue, child_can_start, parent_can_continue) |
| 1207 | ) |
| 1208 | proc.daemon = True |
| 1209 | proc.start() |
| 1210 | |
| 1211 | self.assertEqual(queue_empty(queue), True) |
| 1212 | self.assertEqual(queue_full(queue, MAXSIZE), False) |
| 1213 | |
| 1214 | queue.put(1) |
| 1215 | queue.put(2, True) |
| 1216 | queue.put(3, True, None) |
| 1217 | queue.put(4, False) |
| 1218 | queue.put(5, False, None) |
| 1219 | queue.put_nowait(6) |
| 1220 | |
| 1221 | # the values may be in buffer but not yet in pipe so sleep a bit |
| 1222 | time.sleep(DELTA) |
| 1223 | |
| 1224 | self.assertEqual(queue_empty(queue), False) |
| 1225 | self.assertEqual(queue_full(queue, MAXSIZE), True) |
| 1226 | |
| 1227 | put = TimingWrapper(queue.put) |
| 1228 | put_nowait = TimingWrapper(queue.put_nowait) |
| 1229 | |
| 1230 | self.assertRaises(pyqueue.Full, put, 7, False) |
| 1231 | self.assertTimingAlmostEqual(put.elapsed, 0) |
| 1232 | |
| 1233 | self.assertRaises(pyqueue.Full, put, 7, False, None) |
| 1234 | self.assertTimingAlmostEqual(put.elapsed, 0) |
| 1235 | |
| 1236 | self.assertRaises(pyqueue.Full, put_nowait, 7) |
| 1237 | self.assertTimingAlmostEqual(put_nowait.elapsed, 0) |
| 1238 | |
| 1239 | self.assertRaises(pyqueue.Full, put, 7, True, TIMEOUT1) |
| 1240 | self.assertTimingAlmostEqual(put.elapsed, TIMEOUT1) |
| 1241 | |
| 1242 | self.assertRaises(pyqueue.Full, put, 7, False, TIMEOUT2) |
| 1243 | self.assertTimingAlmostEqual(put.elapsed, 0) |
| 1244 | |
| 1245 | self.assertRaises(pyqueue.Full, put, 7, True, timeout=TIMEOUT3) |
| 1246 | self.assertTimingAlmostEqual(put.elapsed, TIMEOUT3) |
| 1247 | |
| 1248 | child_can_start.set() |
| 1249 | parent_can_continue.wait() |
| 1250 | |
| 1251 | self.assertEqual(queue_empty(queue), True) |
| 1252 | self.assertEqual(queue_full(queue, MAXSIZE), False) |
| 1253 | |
| 1254 | proc.join() |
| 1255 | close_queue(queue) |
nothing calls this directly
no test coverage detected