Dec 10, 2024 · Dec 8, 2024 · Dec 9, 2024 · Dec 9, 2024 · Dec 9, 2024 · Dec 9, 2024
diff --git a/testgres/operations/local_ops.py b/testgres/operations/local_ops.py
        if not get_process:
            input_prepared = Helpers.PrepareProcessInput(input, encoding)  # throw

        assert input_prepared is None or (type(input_prepared) == bytes)  # noqa: E721

        process = subprocess.Popen(
            cmd,
            shell=shell,
            stdin=stdin or subprocess.PIPE if input is not None else None,
            stdout=stdout or subprocess.PIPE,
            stderr=stderr or subprocess.PIPE,
        )
        assert not (process is None)
        if get_process:
            return process, None, None
        try:
            output, error = process.communicate(input=input_prepared, timeout=timeout)
            if encoding:
                output = output.decode(encoding)
                error = error.decode(encoding)
            return process, output, error
        except subprocess.TimeoutExpired:
            process.kill()
            raise ExecUtilException("Command timed out after {} seconds.".format(timeout))

        assert type(output) == bytes  # noqa: E721
        assert type(error) == bytes  # noqa: E721

        if encoding:
            output = output.decode(encoding)
            error = error.decode(encoding)
        return process, output, error

    def _run_command(self, cmd, shell, input, stdin, stdout, stderr, get_process, timeout, encoding):
        """Execute a command and return the process and its output."""
        if os.name == 'nt' and stdout is None:  # Windows
diff --git a/testgres/operations/remote_ops.py b/testgres/operations/remote_ops.py
        if not get_process:
            input_prepared = Helpers.PrepareProcessInput(input, encoding)  # throw

        assert input_prepared is None or (type(input_prepared) == bytes)  # noqa: E721

        ssh_cmd = []
        if isinstance(cmd, str):
            ssh_cmd = ['ssh', self.ssh_dest] + self.ssh_args + [cmd]
        elif isinstance(cmd, list):
            ssh_cmd = ['ssh', self.ssh_dest] + self.ssh_args + cmd
        process = subprocess.Popen(ssh_cmd, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
        assert not (process is None)
        if get_process:
            return process
Original file line number	Diff line number	Diff line change
Expand Up		@@ -86,25 +86,32 @@ def _run_command__generic(self, cmd, shell, input, stdin, stdout, stderr, get_pr
		if not get_process:
		input_prepared = Helpers.PrepareProcessInput(input, encoding) # throw

		assert input_prepared is None or (type(input_prepared) == bytes) # noqa: E721

		process = subprocess.Popen(
		cmd,
		shell=shell,
		stdin=stdin or subprocess.PIPE if input is not None else None,
		stdout=stdout or subprocess.PIPE,
		stderr=stderr or subprocess.PIPE,
		)
		assert not (process is None)
		if get_process:
		return process, None, None
		try:
		output, error = process.communicate(input=input_prepared, timeout=timeout)
		if encoding:
		output = output.decode(encoding)
		error = error.decode(encoding)
		return process, output, error
		except subprocess.TimeoutExpired:
		process.kill()
		raise ExecUtilException("Command timed out after {} seconds.".format(timeout))

		assert type(output) == bytes # noqa: E721
		assert type(error) == bytes # noqa: E721

		if encoding:
		output = output.decode(encoding)
		error = error.decode(encoding)
		return process, output, error

		def _run_command(self, cmd, shell, input, stdin, stdout, stderr, get_process, timeout, encoding):
		"""Execute a command and return the process and its output."""
		if os.name == 'nt' and stdout is None: # Windows
Expand Down
Original file line number	Diff line number	Diff line change
Expand Up		@@ -75,12 +75,15 @@ def exec_command(self, cmd, wait_exit=False, verbose=False, expect_error=False,
		if not get_process:
		input_prepared = Helpers.PrepareProcessInput(input, encoding) # throw

		assert input_prepared is None or (type(input_prepared) == bytes) # noqa: E721

		ssh_cmd = []
		if isinstance(cmd, str):
		ssh_cmd = ['ssh', self.ssh_dest] + self.ssh_args + [cmd]
		elif isinstance(cmd, list):
		ssh_cmd = ['ssh', self.ssh_dest] + self.ssh_args + cmd
		process = subprocess.Popen(ssh_cmd, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
		assert not (process is None)
		if get_process:
		return process

Expand Down