(self)
| 192 | self.fail("Did not detect task count going negative") |
| 193 | |
| 194 | def test_queue_join(self): |
| 195 | # Test that a queue join()s successfully, and before anything else |
| 196 | # (done twice for insurance). |
| 197 | q = self.type2test() |
| 198 | self.queue_join_test(q) |
| 199 | self.queue_join_test(q) |
| 200 | try: |
| 201 | q.task_done() |
| 202 | except ValueError: |
| 203 | pass |
| 204 | else: |
| 205 | self.fail("Did not detect task count going negative") |
| 206 | |
| 207 | def test_basic(self): |
| 208 | # Do it a couple of times on the same queue. |
nothing calls this directly
no test coverage detected