def test_reset_ventilator(self):
        """Resetting ventilator after all items were ventilated will make it re-ventilate the same items"""
        items_count = 100
        for pool in [DummyPool(), ThreadPool(10)]:
            ventilator = ConcurrentVentilator(ventilate_fn=pool.ventilate,
                                              items_to_ventilate=[{'item': i} for i in range(items_count)],
                                              iterations=1)
            pool.start(IdentityWorker, ventilator=ventilator)

            # Readout all ventilated items
            for _ in range(items_count):
                pool.get_results()

            # Should fail reading the next, as all items were read by now
            with self.assertRaises(EmptyResultError):
                pool.get_results()

            # Resetting, hence will be read out the items all over again
            ventilator.reset()

            for _ in range(items_count):
                pool.get_results()

            with self.assertRaises(EmptyResultError):
                pool.get_results()

            pool.stop()
            pool.join()
Resetting ventilator after all items were ventilated will make it re-ventilate the same items
petastorm/workers_pool/tests/test_ventilator.py