Function test_api_check1

dali/test/python/test_pipeline.py:1215–1243

()

def test_api_check1():
    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.run()
    for method in [pipe.schedule_run, pipe.share_outputs, pipe.release_outputs, pipe.outputs]:
        with assert_raises(
            RuntimeError,
            glob=(
                "Mixing pipeline API type. Currently used: PipelineAPIType.BASIC,"
                " but trying to use PipelineAPIType.SCHEDULED"
            ),
        ):
            method()
    # disable check
    pipe.enable_api_check(False)
    for method in [pipe.schedule_run, pipe.share_outputs, pipe.release_outputs, pipe.outputs]:
        method()


def test_api_check2():