(self)
| 1048 | parent_can_continue.set() |
| 1049 | |
| 1050 | def test_put(self): |
| 1051 | MAXSIZE = 6 |
| 1052 | queue = self.Queue(maxsize=MAXSIZE) |
| 1053 | child_can_start = self.Event() |
| 1054 | parent_can_continue = self.Event() |
| 1055 | |
| 1056 | proc = self.Process( |
| 1057 | target=self._test_put, |
| 1058 | args=(queue, child_can_start, parent_can_continue) |
| 1059 | ) |
| 1060 | proc.daemon = True |
| 1061 | proc.start() |
| 1062 | |
| 1063 | self.assertEqual(queue_empty(queue), True) |
| 1064 | self.assertEqual(queue_full(queue, MAXSIZE), False) |
| 1065 | |
| 1066 | queue.put(1) |
| 1067 | queue.put(2, True) |
| 1068 | queue.put(3, True, None) |
| 1069 | queue.put(4, False) |
| 1070 | queue.put(5, False, None) |
| 1071 | queue.put_nowait(6) |
| 1072 | |
| 1073 | # the values may be in buffer but not yet in pipe so sleep a bit |
| 1074 | time.sleep(DELTA) |
| 1075 | |
| 1076 | self.assertEqual(queue_empty(queue), False) |
| 1077 | self.assertEqual(queue_full(queue, MAXSIZE), True) |
| 1078 | |
| 1079 | put = TimingWrapper(queue.put) |
| 1080 | put_nowait = TimingWrapper(queue.put_nowait) |
| 1081 | |
| 1082 | self.assertRaises(pyqueue.Full, put, 7, False) |
| 1083 | self.assertTimingAlmostEqual(put.elapsed, 0) |
| 1084 | |
| 1085 | self.assertRaises(pyqueue.Full, put, 7, False, None) |
| 1086 | self.assertTimingAlmostEqual(put.elapsed, 0) |
| 1087 | |
| 1088 | self.assertRaises(pyqueue.Full, put_nowait, 7) |
| 1089 | self.assertTimingAlmostEqual(put_nowait.elapsed, 0) |
| 1090 | |
| 1091 | self.assertRaises(pyqueue.Full, put, 7, True, TIMEOUT1) |
| 1092 | self.assertTimingAlmostEqual(put.elapsed, TIMEOUT1) |
| 1093 | |
| 1094 | self.assertRaises(pyqueue.Full, put, 7, False, TIMEOUT2) |
| 1095 | self.assertTimingAlmostEqual(put.elapsed, 0) |
| 1096 | |
| 1097 | self.assertRaises(pyqueue.Full, put, 7, True, timeout=TIMEOUT3) |
| 1098 | self.assertTimingAlmostEqual(put.elapsed, TIMEOUT3) |
| 1099 | |
| 1100 | child_can_start.set() |
| 1101 | parent_can_continue.wait() |
| 1102 | |
| 1103 | self.assertEqual(queue_empty(queue), True) |
| 1104 | self.assertEqual(queue_full(queue, MAXSIZE), False) |
| 1105 | |
| 1106 | proc.join() |
| 1107 | close_queue(queue) |
nothing calls this directly
no test coverage detected