Method _shutdown_put_join

Lib/test/test_queue.py:576–607

(self, immediate)

        return self._shutdown_join(False)

    def _shutdown_put_join(self, immediate):
        q = self.type2test(2)
        results = []
        go = threading.Event()
        q.put("Y")
        # queue not fulled

        thrds = (
            (self._put_shutdown, (q, "E", go, results)),
            (self._join, (q, results)),
        )
        threads = []
        for func, params in thrds:
            threads.append(threading.Thread(target=func, args=params))
            threads[-1].start()
        self.assertEqual(q.unfinished_tasks, 1)

        q.shutdown(immediate)
        go.set()

        if immediate:
            with self.assertRaises(self.queue.ShutDown):
                q.get_nowait()
        else:
            result = q.get()
            self.assertEqual(result, "Y")
            q.task_done()

        for t in threads:
            t.join()

        self.assertEqual(results, [True]*len(thrds))

    def test_shutdown_immediate_put_join(self):
        return self._shutdown_put_join(True)