How to use the renku.core.models.cwl.command_line_tool.CommandLineToolFactory function in renku

def test_05_stdout(instance_path):
    """Test stdout mapping."""
    output = Path(instance_path) / 'output.txt'
    output.touch()

    argv = ['echo', 'Hello world!']
    factory = CommandLineToolFactory(
        argv,
        directory=instance_path,
        working_dir=instance_path,
        stdout='output.txt',
    )

    assert factory.stdout == 'output.txt'
    assert factory.outputs[0].type == 'stdout'

    tool = factory.generate_tool()
    assert tool.to_argv() == argv

def test_stdout_with_conflicting_arg(instance_path):
    """Test stdout with conflicting argument value."""
    output = Path(instance_path) / 'lalala'
    output.touch()

    argv = ['echo', 'lalala']
    factory = CommandLineToolFactory(
        argv,
        directory=instance_path,
        working_dir=instance_path,
        stdout='lalala',
    )

    assert factory.inputs[0].default == 'lalala'
    assert factory.inputs[0].type == 'string'
    assert factory.stdout == 'lalala'
    assert factory.outputs[0].type == 'stdout'

    tool = factory.generate_tool()
    assert tool.to_argv() == argv

def test_base_command_as_file_input(instance_path):
    """Test base command detection when it is a script file."""
    cwd = Path(instance_path)
    script = cwd / 'script.py'
    script.touch()

    input_file = cwd / 'input.csv'
    input_file.touch()

    argv = ['script.py', 'input.csv']
    tool = CommandLineToolFactory(
        argv,
        directory=instance_path,
        working_dir=instance_path,
    ).generate_tool()

    assert not tool.baseCommand
    assert 2 == len(tool.inputs)

def test_04_output(instance_path):
    """Test describtion of outputs from a command."""
    hello = Path(instance_path) / 'hello.tar'
    hello.touch()

    argv = ['tar', 'xf', 'hello.tar']
    factory = CommandLineToolFactory(
        argv, directory=instance_path, working_dir=instance_path
    )

    # simulate run

    output = Path(instance_path) / 'hello.txt'
    output.touch()

    parameters = list(factory.guess_outputs([output]))

    assert parameters[0][0].type == 'File'
    assert parameters[0][0].outputBinding.glob == 'hello.txt'

    tool = factory.generate_tool()
    assert tool.to_argv() == argv

def test_base_command_detection(instance_path):
    """Test base command detection."""
    hello = Path(instance_path) / 'hello.tar'
    hello.touch()

    argv = ['tar', 'xf', 'hello.tar']
    tool = CommandLineToolFactory(
        argv, directory=instance_path, working_dir=instance_path
    ).generate_tool()

    assert tool.baseCommand == ['tar', 'xf']
    assert tool.inputs[0].default.path.samefile(hello)
    assert tool.inputs[0].type == 'File'
    assert tool.inputs[0].inputBinding.prefix is None
    assert tool.inputs[0].inputBinding.separate is True

    assert tool.to_argv() == argv

def test_06_params(instance_path):
    """Test referencing input parameters in other fields."""
    hello = Path(instance_path) / 'hello.tar'
    hello.touch()

    argv = ['tar', 'xf', 'hello.tar', 'goodbye.txt']
    factory = CommandLineToolFactory(
        argv,
        directory=instance_path,
        working_dir=instance_path,
    )

    assert factory.inputs[1].default == 'goodbye.txt'
    assert factory.inputs[1].type == 'string'
    assert factory.inputs[1].inputBinding.position == 2

    goodbye_id = factory.inputs[1].id

    # simulate run

    output = Path(instance_path) / 'goodbye.txt'
    output.touch()

def test_stdin_and_stdout(argv, instance_path):
    """Test stdout mapping."""
    input_ = Path(instance_path) / 'input.txt'
    input_.touch()
    output = Path(instance_path) / 'output.txt'
    output.touch()

    factory = CommandLineToolFactory(
        argv,
        directory=instance_path,
        working_dir=instance_path,
        stdin='input.txt',
        stdout='output.txt',
        stderr='error.log',
    )

    assert factory.stdin
    if len(argv) > 1:
        assert factory.arguments

    assert factory.stdout == 'output.txt'
    assert factory.outputs[0].type == 'stdout'

    tool = factory.generate_tool()

def test_input_directory(instance_path):
    """Test input directory."""
    cwd = Path(instance_path)
    src = cwd / 'src'
    src.mkdir(parents=True)

    for i in range(5):
        (src / str(i)).touch()

    argv = ['tar', 'czvf', 'src.tar', 'src']
    factory = CommandLineToolFactory(
        argv,
        directory=instance_path,
        working_dir=instance_path,
    )

    src_tar = src / 'src.tar'
    src_tar.touch()

    tool = factory.generate_tool()
    assert tool.to_argv() == argv

    assert tool.inputs[0].type == 'string'
    assert tool.inputs[0].default == src_tar.name
    assert tool.inputs[1].type == 'Directory'
    assert tool.inputs[1].default.path.samefile(src)

def test_exitings_output_directory(client):
    """Test creation of InitialWorkDirRequirement for output directory."""
    instance_path = client.path
    output = client.path / 'output'

    argv = ['script', 'output']
    factory = CommandLineToolFactory(
        argv,
        directory=instance_path,
        working_dir=instance_path,
    )

    with factory.watch(client, no_output=True) as tool:
        # Script creates the directory.
        output.mkdir(parents=True)

    initial_work_dir_requirement = [
        r for r in tool.requirements
        if r.__class__.__name__ == 'InitialWorkDirRequirement'
    ]
    assert 0 == len(initial_work_dir_requirement)

    output.mkdir(parents=True, exist_ok=True)

def run(
    client, inputs, outputs, no_output, success_codes, isolation, command_line
):
    """Tracking work on a specific problem."""
    working_dir = client.repo.working_dir
    mapped_std = _mapped_std_streams(client.candidate_paths)
    factory = CommandLineToolFactory(
        command_line=command_line,
        explicit_inputs=inputs,
        explicit_outputs=outputs,
        directory=os.getcwd(),
        working_dir=working_dir,
        successCodes=success_codes,
        **{
            name: os.path.relpath(path, working_dir)
            for name, path in mapped_std.items()
        }
    )
    with client.with_workflow_storage() as wf:
        with factory.watch(client, no_output=no_output) as tool:
            # Don't compute paths if storage is disabled.
            if client.has_external_storage:
                # Make sure all inputs are pulled from a storage.