(self, q)
| 162 | q.task_done() |
| 163 | |
| 164 | def queue_join_test(self, q): |
| 165 | self.cum = 0 |
| 166 | threads = [] |
| 167 | for i in (0,1): |
| 168 | thread = threading.Thread(target=self.worker, args=(q,)) |
| 169 | thread.start() |
| 170 | threads.append(thread) |
| 171 | for i in range(100): |
| 172 | q.put(i) |
| 173 | q.join() |
| 174 | self.assertEqual(self.cum, sum(range(100)), |
| 175 | "q.join() did not block until all tasks were done") |
| 176 | for i in (0,1): |
| 177 | q.put(-1) # instruct the threads to close |
| 178 | q.join() # verify that you can join twice |
| 179 | for thread in threads: |
| 180 | thread.join() |
| 181 | |
| 182 | def test_queue_task_done(self): |
| 183 | # Test to make sure a queue task completed successfully. |
no test coverage detected