Skip to content
Permalink
Browse files

Wrap subprocess.Popen.communicate in order to allow encoding flag as …

…Python 3 version does

Change-Id: I0c90db90e706707a01605cf65149760021ab76fd
  • Loading branch information
si-23 committed Nov 7, 2019
1 parent e70b087 commit 979179b8698ce137e67b3c94ae0f048882fc4c09
@@ -0,0 +1,140 @@
#!/usr/bin/env python
# -*- encoding: utf-8; py-indent-offset: 4 -*-
# +------------------------------------------------------------------+
# | ____ _ _ __ __ _ __ |
# | / ___| |__ ___ ___| | __ | \/ | |/ / |
# | | | | '_ \ / _ \/ __| |/ / | |\/| | ' / |
# | | |___| | | | __/ (__| < | | | | . \ |
# | \____|_| |_|\___|\___|_|\_\___|_| |_|_|\_\ |
# | |
# | Copyright Mathias Kettner 2019 mk@mathias-kettner.de |
# +------------------------------------------------------------------+
#
# This file is part of Check_MK.
# The official homepage is at http://mathias-kettner.de/check_mk.
#
# check_mk is free software; you can redistribute it and/or modify it
# under the terms of the GNU General Public License as published by
# the Free Software Foundation in version 2. check_mk is distributed
# in the hope that it will be useful, but WITHOUT ANY WARRANTY; with-
# out even the implied warranty of MERCHANTABILITY or FITNESS FOR A
# PARTICULAR PURPOSE. See the GNU General Public License for more de-
# tails. You should have received a copy of the GNU General Public
# License along with GNU Make; see the file COPYING. If not, write
# to the Free Software Foundation, Inc., 51 Franklin St, Fifth Floor,
# Boston, MA 02110-1301 USA.
"""
This module is a little wrapper for the Python 2 subprocess.Popen and
communicate method in order to allow the flag 'encoding' as Python 3 version
does. This can be removed after Python 3 migration.
"""

import sys
import subprocess
import six

PIPE = subprocess.PIPE
STDOUT = subprocess.STDOUT
list2cmdline = subprocess.list2cmdline
call = subprocess.call
check_output = subprocess.check_output
CalledProcessError = subprocess.CalledProcessError

if sys.platform == "win32":
CREATE_NEW_PROCESS_GROUP = subprocess.CREATE_NEW_PROCESS_GROUP
else:
CREATE_NEW_PROCESS_GROUP = None

if six.PY3:
Popen = subprocess.Popen

else:

class Popen(subprocess.Popen):
def __init__(
self,
args,
bufsize=0,
executable=None,
stdin=None,
stdout=None,
stderr=None,
preexec_fn=None,
close_fds=False,
shell=False,
cwd=None,
env=None,
universal_newlines=False,
startupinfo=None,
creationflags=0,
encoding=None,
):
super(Popen, self).__init__(
args,
bufsize=bufsize,
executable=executable,
stdin=stdin,
stdout=stdout,
stderr=stderr,
preexec_fn=preexec_fn,
close_fds=close_fds,
shell=shell,
cwd=cwd,
env=env,
universal_newlines=universal_newlines,
startupinfo=startupinfo,
creationflags=creationflags,
)

# Cannot decode streams from args as Python 3 subprocess module does:
# py2 subprocess uses os.fdopen
# py3 subprocess uses io.open

if universal_newlines:
# We don't need this arg on Checkmk servers:
# Lines may be terminated by any of '\n', the Unix end-of-line
# convention, '\r', the old Macintosh convention or '\r\n', the
# Windows convention. All of these external representations are
# seen as '\n' by the Python program.
raise AttributeError("Do not use 'universal_newlines'")

self.encoding = encoding

def communicate(self, input=None): # pylint: disable=redefined-builtin
# Python 2:
# The optional input argument should be a
# string to be sent to the child process, or None, if no data
# should be sent to the child.

# Python 3:
# The optional "input" argument should be data to be sent to the
# child process (if self.universal_newlines is True, this should
# be a string; if it is False, "input" should be bytes), or
# None, if no data should be sent to the child.
# By default, all communication is in bytes, and therefore any
# "input" should be bytes, and the (stdout, stderr) will be bytes.
# If in text mode (indicated by self.text_mode), any "input" should
# be a string, and (stdout, stderr) will be strings decoded
# according to locale encoding, or by "encoding" if set. Text mode
# is triggered by setting any of text, encoding, errors or
# universal_newlines.

if input is not None:
if self.encoding:
if not isinstance(input, six.text_type):
# As Python 3 subprocess does:
raise AttributeError("'bytes' object has no attribute 'encode'")
input = input.encode(self.encoding)
else:
if not isinstance(input, six.binary_type):
# As Python 3 subprocess does:
raise TypeError("a bytes-like object is required, not 'str'")

stdout, stderr = super(Popen, self).communicate(input=input)

if self.encoding:
if stdout is not None:
stdout = stdout.decode(self.encoding)
if stderr is not None:
stderr = stderr.decode(self.encoding)
return (stdout, stderr)
@@ -0,0 +1,75 @@
# -*- encoding: utf-8; py-indent-offset: 4 -*-
import pytest
import six
import cmk.utils.cmk_subprocess as subprocess


def _check_type_of_stdout_and_stderr(p_communicate, encoding):
stdout, stderr = p_communicate
if stdout:
if encoding:
assert isinstance(stdout, six.text_type)
else:
assert isinstance(stdout, six.binary_type)
if stderr:
if encoding:
assert isinstance(stdout, six.text_type)
else:
assert isinstance(stdout, six.binary_type)


@pytest.mark.parametrize("command, encoding", [
(['ls'], None),
(['ls'], "utf-8"),
])
def test_cmk_subprocess_no_input(command, encoding):
p = subprocess.Popen(
command,
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
encoding=encoding,
)
_check_type_of_stdout_and_stderr(p.communicate(), encoding)


@pytest.mark.parametrize("command, input_, encoding", [
(['grep', 'f.*'], b'foo', None),
(['grep', 'f.*'], b'f\xc3\xb6\xc3\xb6', None),
(['grep', 'f.*'], u'foo', "utf-8"),
(['grep', 'f.*'], u'föö', "utf-8"),
])
def test_cmk_subprocess_input_no_errors(command, input_, encoding):
p = subprocess.Popen(
command,
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
encoding=encoding,
)
_check_type_of_stdout_and_stderr(p.communicate(), encoding)


@pytest.mark.parametrize("command, input_, encoding, py3_error", [
(['grep', 'f.*'], b'foo', "utf-8", AttributeError),
(['grep', 'f.*'], b'f\xc3\xb6\xc3\xb6', "utf-8", AttributeError),
(['grep', 'f.*'], u'foo', None, TypeError),
(['grep', 'f.*'], u'föö', None, TypeError),
])
def test_cmk_subprocess_input_errors(command, input_, encoding, py3_error):
p = subprocess.Popen(
command,
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
encoding=encoding,
)
if six.PY3:
# from Python 3 subprocess:
# AttributeError: 'bytes' object has no attribute 'encode'
# TypeError: memoryview: a bytes-like object is required, not 'str'
with pytest.raises(py3_error):
p.communicate(input_)
else:
with pytest.raises(AssertionError):
p.communicate(input_)
@@ -0,0 +1,71 @@
# -*- encoding: utf-8; py-indent-offset: 4 -*-
import pytest
import six
import cmk.utils.cmk_subprocess as subprocess


def _check_type_of_stdout_and_stderr(p_communicate, encoding):
stdout, stderr = p_communicate
if stdout:
if encoding:
assert isinstance(stdout, six.text_type)
else:
assert isinstance(stdout, six.binary_type)
if stderr:
if encoding:
assert isinstance(stdout, six.text_type)
else:
assert isinstance(stdout, six.binary_type)


@pytest.mark.parametrize("command, encoding", [
(['ls'], None),
(['ls'], "utf-8"),
])
def test_cmk_subprocess_no_input(command, encoding):
p = subprocess.Popen(
command,
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
encoding=encoding,
)
_check_type_of_stdout_and_stderr(p.communicate(), encoding)


@pytest.mark.parametrize("command, input_, encoding", [
(['grep', 'f.*'], b'foo', None),
(['grep', 'f.*'], b'f\xc3\xb6\xc3\xb6', None),
(['grep', 'f.*'], u'foo', "utf-8"),
(['grep', 'f.*'], u'föö', "utf-8"),
])
def test_cmk_subprocess_input_no_errors(command, input_, encoding):
p = subprocess.Popen(
command,
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
encoding=encoding,
)
_check_type_of_stdout_and_stderr(p.communicate(), encoding)


@pytest.mark.parametrize("command, input_, encoding, py3_error", [
(['grep', 'f.*'], b'foo', "utf-8", AttributeError),
(['grep', 'f.*'], b'f\xc3\xb6\xc3\xb6', "utf-8", AttributeError),
(['grep', 'f.*'], u'foo', None, TypeError),
(['grep', 'f.*'], u'föö', None, TypeError),
])
def test_cmk_subprocess_input_errors(command, input_, encoding, py3_error):
p = subprocess.Popen(
command,
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
encoding=encoding,
)
# from Python 3 subprocess:
# AttributeError: 'bytes' object has no attribute 'encode'
# TypeError: memoryview: a bytes-like object is required, not 'str'
with pytest.raises(py3_error):
p.communicate(input_)

0 comments on commit 979179b

Please sign in to comment.
You can’t perform that action at this time.