Method test_pool_one_task

dali/test/python/test_pool.py:126–138

(self, start_method)

    @params(*start_methods)
    def test_pool_one_task(self, start_method):
        groups = [MockGroup.from_callback(simple_callback)]
        with create_pool(
            groups, keep_alive_queue_size=1, num_workers=1, start_method=start_method
        ) as pool:
            pids = get_pids(pool)
            pid = pids[0]
            tasks = [(SampleInfo(0, 0, 0, 0),)]
            work_batch = TaskArgs.make_sample(SampleRange(0, 1, 0, 0))
            pool.schedule_batch(context_i=0, work_batch=work_batch)
            batch = pool.receive_batch(context_i=0)
            for task, sample in zip(tasks, batch):
                np.testing.assert_array_equal(answer(pid, *task), sample)

    @params(*start_methods)
    def test_pool_multi_task(self, start_method):