Function test_wrong_source

dali/test/python/test_external_source_parallel.py:113–135

()

def test_wrong_source():
    callable_msg = (
        "Callable passed to External Source in parallel mode (when `parallel=True`) "
        "must accept exactly one argument*. Got {} instead."
    )
    batch_required_msg = "Parallel external source with {} must be run in a batch mode"
    disallowed_sources = [
        (no_arg_fun, (TypeError, callable_msg.format("a callable that does not accept arguments"))),
        (
            multi_arg_fun,
            (
                TypeError,
                "The `source` callable must accept either 0 or 1 positional arguments "
                "to indicate whether it accepts the batch or sample indexing information. "
                "Found more than one positional argument, which is not allowed.",
            ),
        ),
        (Iterable(), (TypeError, batch_required_msg.format("an iterable"))),
        (generator_fun, (TypeError, batch_required_msg.format("a generator function"))),
        (generator_fun(), (TypeError, batch_required_msg.format("an iterable"))),
    ]
    for source, (error_type, error_msg) in disallowed_sources:
        raises(error_type, error_msg)(check_source_build)(source)


# Test that we can launch several CPU-only pipelines by fork as we don't touch CUDA context.