Consistently check subprocess output

In some places we used universal_newlines and in others we didn't. In
some places we decoded from utf8 and in others we didn't. Switch to not
using universal_newlines at all because this implies getting back a str
instance on python 3 not a binary string which can't be decoded. Then
consistently decode using the users preferred local as that is what
universal_newlines will attempt to do.

We also update things to treat all exception output as binary initially,
convert to user's locale and then parse. This makes error handling
consistent with normal situation handling across the board for python2
and python3 as well.

Note that we don't appear to do any line splitting on newlines here
anyways so this shouldn't affect anything, also these all run on Linux
where you should get unix newlines anyways.

Change-Id: I8dc5ee0b14906324169def2cdc35fbcd4f776423
This commit is contained in:
Clark Boylan 2017-04-10 16:25:48 -07:00
parent 891894266f
commit 428ae68863
2 changed files with 27 additions and 37 deletions

View File

@ -15,6 +15,7 @@
# See the License for the specific language governing permissions and
# limitations under the License.
from locale import getpreferredencoding
import logging
import os.path
from parsley import makeGrammar
@ -226,7 +227,7 @@ class Depends(object):
def platform_profiles(self):
output = subprocess.check_output(
["lsb_release", "-cirs"],
stderr=subprocess.STDOUT).decode('utf-8')
stderr=subprocess.STDOUT).decode(getpreferredencoding(False))
lsbinfo = output.lower().split()
# NOTE(toabctl): distro can be more than one string (i.e. "SUSE LINUX")
codename = lsbinfo[len(lsbinfo) - 1:len(lsbinfo)][0]
@ -279,12 +280,12 @@ class Dpkg(Platform):
output = subprocess.check_output(
["dpkg-query", "-W", "-f", "${Package} ${Status} ${Version}\n",
pkg_name],
stderr=subprocess.STDOUT,
universal_newlines=True).decode('utf-8')
stderr=subprocess.STDOUT).decode(getpreferredencoding(False))
except subprocess.CalledProcessError as e:
eoutput = e.output.decode(getpreferredencoding(False))
if (e.returncode == 1 and
(e.output.startswith('dpkg-query: no packages found') or
e.output.startswith('No packages found matching'))):
(eoutput.startswith('dpkg-query: no packages found') or
eoutput.startswith('No packages found matching'))):
return None
raise
# output looks like
@ -309,11 +310,11 @@ class Rpm(Platform):
["rpm", "--qf",
"%{NAME} %|EPOCH?{%{EPOCH}:}|%{VERSION}-%{RELEASE}\n", "-q",
pkg_name],
stderr=subprocess.STDOUT,
universal_newlines=True).decode('utf-8')
stderr=subprocess.STDOUT).decode(getpreferredencoding(False))
except subprocess.CalledProcessError as e:
eoutput = e.output.decode(getpreferredencoding(False))
if (e.returncode == 1 and
e.output.strip().endswith('is not installed')):
eoutput.strip().endswith('is not installed')):
return None
raise
# output looks like
@ -335,8 +336,7 @@ class Emerge(Platform):
try:
output = subprocess.check_output(
['equery', 'l', '--format=\'$version\'', pkg_name],
stderr=subprocess.STDOUT,
universal_newlines=True).decode('utf-8')
stderr=subprocess.STDOUT).decode(getpreferredencoding(False))
except subprocess.CalledProcessError as e:
if e.returncode == 3:
return None
@ -358,10 +358,10 @@ class Pacman(Platform):
try:
output = subprocess.check_output(
['pacman', '-Q', pkg_name],
stderr=subprocess.STDOUT,
universal_newlines=True)
stderr=subprocess.STDOUT).decode(getpreferredencoding(False))
except subprocess.CalledProcessError as e:
if e.returncode == 1 and e.output.endswith('was not found'):
eoutput = e.output.decode(getpreferredencoding(False))
if e.returncode == 1 and eoutput.endswith('was not found'):
return None
raise
# output looks like

View File

@ -361,8 +361,7 @@ class TestDpkg(TestCase):
mock_checkoutput.assert_called_once_with(
["dpkg-query", "-W", "-f",
"${Package} ${Status} ${Version}\n", "foo"],
stderr=subprocess.STDOUT,
universal_newlines=True)
stderr=subprocess.STDOUT)
def test_unknown_package(self):
platform = Dpkg()
@ -371,15 +370,14 @@ class TestDpkg(TestCase):
def _side_effect_raise(*args, **kwargs):
raise subprocess.CalledProcessError(
1, [], "dpkg-query: no packages found matching foo\n")
1, [], b"dpkg-query: no packages found matching foo\n")
mock_checkoutput.side_effect = _side_effect_raise
self.assertEqual(None, platform.get_pkg_version("foo"))
mock_checkoutput.assert_called_once_with(
["dpkg-query", "-W", "-f",
"${Package} ${Status} ${Version}\n", "foo"],
stderr=subprocess.STDOUT,
universal_newlines=True)
stderr=subprocess.STDOUT)
def test_installed_version(self):
platform = Dpkg()
@ -393,8 +391,7 @@ class TestDpkg(TestCase):
mocked_checkoutput.assert_called_once_with(
["dpkg-query", "-W", "-f",
"${Package} ${Status} ${Version}\n", "foo"],
stderr=subprocess.STDOUT,
universal_newlines=True)
stderr=subprocess.STDOUT)
class TestEmerge(TestCase):
@ -412,8 +409,7 @@ class TestEmerge(TestCase):
self.assertEqual(None, platform.get_pkg_version("foo"))
mocked_checkoutput.assert_called_once_with(
['equery', 'l', '--format=\'$version\'', 'foo'],
stderr=subprocess.STDOUT,
universal_newlines=True)
stderr=subprocess.STDOUT)
def test_unknown_package(self):
platform = Emerge()
@ -428,8 +424,7 @@ class TestEmerge(TestCase):
self.assertEqual(None, platform.get_pkg_version("foo"))
mocked_checkoutput.assert_called_once_with(
['equery', 'l', '--format=\'$version\'', 'foo'],
stderr=subprocess.STDOUT,
universal_newlines=True)
stderr=subprocess.STDOUT)
def test_installed_version(self):
platform = Emerge()
@ -439,8 +434,7 @@ class TestEmerge(TestCase):
self.assertEqual("4.0.0", platform.get_pkg_version("foo"))
mock_checkoutput.assert_called_once_with(
['equery', 'l', '--format=\'$version\'', 'foo'],
stderr=subprocess.STDOUT,
universal_newlines=True)
stderr=subprocess.STDOUT)
class TestPacman(TestCase):
@ -450,7 +444,7 @@ class TestPacman(TestCase):
def _side_effect_raise(*args, **kwargs):
raise subprocess.CalledProcessError(
1, [], "error: package 'foo' was not found")
1, [], b"error: package 'foo' was not found")
mock_checkoutput = self.useFixture(
fixtures.MockPatchObject(subprocess, "check_output")).mock
@ -459,20 +453,18 @@ class TestPacman(TestCase):
self.assertEqual(None, platform.get_pkg_version("foo"))
mock_checkoutput.assert_called_once_with(
['pacman', '-Q', 'foo'],
stderr=subprocess.STDOUT,
universal_newlines=True)
stderr=subprocess.STDOUT)
self.assertEqual(None, platform.get_pkg_version("foo"))
def test_installed_version(self):
platform = Pacman()
mock_checkoutput = self.useFixture(
fixtures.MockPatchObject(subprocess, "check_output")).mock
mock_checkoutput.return_value = 'foo 4.0.0-2'
mock_checkoutput.return_value = b'foo 4.0.0-2'
self.assertEqual("4.0.0-2", platform.get_pkg_version("foo"))
mock_checkoutput.assert_called_once_with(
['pacman', '-Q', 'foo'],
stderr=subprocess.STDOUT,
universal_newlines=True)
stderr=subprocess.STDOUT)
class TestRpm(TestCase):
@ -485,7 +477,7 @@ class TestRpm(TestCase):
def _side_effect_raise(*args, **kwargs):
raise subprocess.CalledProcessError(
1, [], "package foo is not installed\n")
1, [], b"package foo is not installed\n")
mock_checkoutput = self.useFixture(
fixtures.MockPatchObject(subprocess, 'check_output')).mock
@ -495,8 +487,7 @@ class TestRpm(TestCase):
["rpm", "--qf",
"%{NAME} %|EPOCH?{%{EPOCH}:}|%{VERSION}-%{RELEASE}\n", "-q",
"foo"],
stderr=subprocess.STDOUT,
universal_newlines=True)
stderr=subprocess.STDOUT)
self.assertEqual(None, platform.get_pkg_version("foo"))
def test_installed_version(self):
@ -509,8 +500,7 @@ class TestRpm(TestCase):
["rpm", "--qf",
"%{NAME} %|EPOCH?{%{EPOCH}:}|%{VERSION}-%{RELEASE}\n", "-q",
"foo"],
stderr=subprocess.STDOUT,
universal_newlines=True)
stderr=subprocess.STDOUT)
class TestEval(TestCase):