Function test_api_check2

dali/test/python/test_pipeline.py:1246–1276

()

def test_api_check2():
    batch_size = 1

    class TestPipeline(Pipeline):
        def __init__(self, batch_size, num_threads, device_id, num_gpus):
            super(TestPipeline, self).__init__(batch_size, num_threads, device_id)
            self.input = ops.readers.Caffe(
                path=caffe_db_folder, shard_id=device_id, num_shards=num_gpus
            )

        def define_graph(self):
            inputs, labels = self.input(name="Reader")
            return inputs

    pipe = TestPipeline(batch_size=batch_size, num_threads=1, device_id=0, num_gpus=1)
    pipe.schedule_run()
    pipe.share_outputs()
    pipe.release_outputs()
    pipe.schedule_run()
    pipe.outputs()
    with assert_raises(
        RuntimeError,
        glob=(
            "Mixing pipeline API type. Currently used: PipelineAPIType.SCHEDULED,"
            " but trying to use PipelineAPIType.BASIC"
        ),
    ):
        pipe.run()
    # disable check
    pipe.enable_api_check(False)
    pipe.run()


class DupPipeline(Pipeline):