openstack/trove
Code Issues Proposed changes

pkg: replace commands module with subprocess

Browse Source 
The commands is deprecated since Python 2.6 and was removed from
Python 3. This change replaces it with subprocess. By the way, it
also changes the code to avoid a shell and so indirectly makes the
code safer (shell injections are no more possibles).

tox: run test_backupagent and test_pkg on Python 3.4.

Partially implements: blueprint trove-python3
Change-Id: I3818fa1498819cc77a3da42c12245c50b96cad59
This commit is contained in:
Victor Stinner 2016-05-26 17:06:45 +02:00
parent a5f1ec074c
commit 4f888ddfa3
3 changed files with 46 additions and 35 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
tox.ini
View File

@ -36,6 +36,7 @@ commands =
trove/tests/unittests/api/test_versions.py \ trove/tests/unittests/api/test_versions.py \
trove/tests/unittests/backup/test_backup_controller.py \ trove/tests/unittests/backup/test_backup_controller.py \
trove/tests/unittests/backup/test_backup_models.py \ trove/tests/unittests/backup/test_backup_models.py \
trove/tests/unittests/backup/test_backupagent.py \
trove/tests/unittests/cluster/test_cassandra_cluster.py \ trove/tests/unittests/cluster/test_cassandra_cluster.py \
trove/tests/unittests/cluster/test_cluster.py \ trove/tests/unittests/cluster/test_cluster.py \
trove/tests/unittests/cluster/test_cluster_controller.py \ trove/tests/unittests/cluster/test_cluster_controller.py \
@ -70,6 +71,7 @@ commands =
trove/tests/unittests/guestagent/test_galera_cluster_api.py \ trove/tests/unittests/guestagent/test_galera_cluster_api.py \
trove/tests/unittests/guestagent/test_guestagent_utils.py \ trove/tests/unittests/guestagent/test_guestagent_utils.py \
trove/tests/unittests/guestagent/test_models.py \ trove/tests/unittests/guestagent/test_models.py \
trove/tests/unittests/guestagent/test_pkg.py \
trove/tests/unittests/guestagent/test_query.py \ trove/tests/unittests/guestagent/test_query.py \
trove/tests/unittests/guestagent/test_service.py \ trove/tests/unittests/guestagent/test_service.py \
trove/tests/unittests/guestagent/test_vertica_api.py \ trove/tests/unittests/guestagent/test_vertica_api.py \

36
trove/guestagent/pkg.py
View File

@ -16,7 +16,6 @@
""" """
Manages packages on the Guest VM. Manages packages on the Guest VM.
""" """
import commands
import os import os
import re import re
import subprocess import subprocess
@ -39,6 +38,23 @@ REINSTALL_FIRST = 2
CONFLICT_REMOVED = 3 CONFLICT_REMOVED = 3
def getoutput(*cmd):
"""Get the stdout+stderr of a command, ignore errors.
Similar to commands.getstatusoutput(cmd)[1] of Python 2.
"""
try:
proc = subprocess.Popen(cmd,
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT)
except OSError:
# ignore errors like program not found
return b''
stdout = proc.communicate()[0]
return stdout
class PkgAdminLockError(exception.TroveError): class PkgAdminLockError(exception.TroveError):
pass pass
@ -190,9 +206,7 @@ class RedhatPackagerMixin(BasePackagerMixin):
def pkg_is_installed(self, packages): def pkg_is_installed(self, packages):
packages = packages if isinstance(packages, list) else packages.split() packages = packages if isinstance(packages, list) else packages.split()
cmd = "rpm -qa" std_out = getoutput("rpm", "-qa")
p = commands.getstatusoutput(cmd)
std_out = p[1]
for pkg in packages: for pkg in packages:
found = False found = False
for line in std_out.split("\n"): for line in std_out.split("\n"):
@ -204,12 +218,11 @@ class RedhatPackagerMixin(BasePackagerMixin):
return True return True
def pkg_version(self, package_name): def pkg_version(self, package_name):
cmd_list = ["rpm", "-qa", "--qf", "'%{VERSION}-%{RELEASE}\n'", std_out = getoutput("rpm", "-qa",
package_name] "--qf", "'%{VERSION}-%{RELEASE}\n'",
p = commands.getstatusoutput(' '.join(cmd_list)) package_name)
# Need to capture the version string # Need to capture the version string
# check the command output # check the command output
std_out = p[1]
for line in std_out.split("\n"): for line in std_out.split("\n"):
regex = re.compile("[0-9.]+-.*") regex = re.compile("[0-9.]+-.*")
matches = regex.match(line) matches = regex.match(line)
@ -254,9 +267,7 @@ class DebianPackagerMixin(BasePackagerMixin):
package_name = m.group(1) package_name = m.group(1)
else: else:
package_name = package package_name = package
command = "sudo debconf-show %s" % package_name std_out = getoutput("sudo", "debconf-show", package_name)
p = commands.getstatusoutput(command)
std_out = p[1]
for line in std_out.split("\n"): for line in std_out.split("\n"):
for selection, value in config_opts.items(): for selection, value in config_opts.items():
m = re.match(".* (.*/%s):.*" % selection, line) m = re.match(".* (.*/%s):.*" % selection, line)
@ -368,8 +379,7 @@ class DebianPackagerMixin(BasePackagerMixin):
self._fix_package_selections(packages, config_opts) self._fix_package_selections(packages, config_opts)
def pkg_version(self, package_name): def pkg_version(self, package_name):
p = commands.getstatusoutput("apt-cache policy %s" % package_name) std_out = getoutput("apt-cache", "policy", package_name)
std_out = p[1]
for line in std_out.split("\n"): for line in std_out.split("\n"):
m = re.match("\s+Installed: (.*)", line) m = re.match("\s+Installed: (.*)", line)
if m: if m:

43
trove/tests/unittests/guestagent/test_pkg.py
View File

@ -13,7 +13,6 @@
# License for the specific language governing permissions and limitations # License for the specific language governing permissions and limitations
# under the License. # under the License.
import commands
import os import os
import re import re
import subprocess import subprocess
@ -265,27 +264,27 @@ class PkgDEBVersionTestCase(trove_testtools.TestCase):
super(PkgDEBVersionTestCase, self).setUp() super(PkgDEBVersionTestCase, self).setUp()
self.pkgName = 'mysql-server-5.5' self.pkgName = 'mysql-server-5.5'
self.pkgVersion = '5.5.28-0' self.pkgVersion = '5.5.28-0'
self.commands_output = commands.getstatusoutput self.getoutput = pkg.getoutput
def tearDown(self): def tearDown(self):
super(PkgDEBVersionTestCase, self).tearDown() super(PkgDEBVersionTestCase, self).tearDown()
commands.getstatusoutput = self.commands_output pkg.getoutput = self.getoutput
def test_version_success(self): def test_version_success(self):
cmd_out = "%s:\n Installed: %s\n" % (self.pkgName, self.pkgVersion) cmd_out = "%s:\n Installed: %s\n" % (self.pkgName, self.pkgVersion)
commands.getstatusoutput = Mock(return_value=(0, cmd_out)) pkg.getoutput = Mock(return_value=cmd_out)
version = pkg.DebianPackagerMixin().pkg_version(self.pkgName) version = pkg.DebianPackagerMixin().pkg_version(self.pkgName)
self.assertTrue(version) self.assertTrue(version)
self.assertEqual(self.pkgVersion, version) self.assertEqual(self.pkgVersion, version)
def test_version_unknown_package(self): def test_version_unknown_package(self):
cmd_out = "N: Unable to locate package %s" % self.pkgName cmd_out = "N: Unable to locate package %s" % self.pkgName
commands.getstatusoutput = Mock(return_value=(0, cmd_out)) pkg.getoutput = Mock(return_value=cmd_out)
self.assertFalse(pkg.DebianPackagerMixin().pkg_version(self.pkgName)) self.assertFalse(pkg.DebianPackagerMixin().pkg_version(self.pkgName))
def test_version_no_version(self): def test_version_no_version(self):
cmd_out = "%s:\n Installed: %s\n" % (self.pkgName, "(none)") cmd_out = "%s:\n Installed: %s\n" % (self.pkgName, "(none)")
commands.getstatusoutput = Mock(return_value=(0, cmd_out)) pkg.getoutput = Mock(return_value=cmd_out)
self.assertFalse(pkg.DebianPackagerMixin().pkg_version(self.pkgName)) self.assertFalse(pkg.DebianPackagerMixin().pkg_version(self.pkgName))
@ -295,21 +294,21 @@ class PkgRPMVersionTestCase(trove_testtools.TestCase):
super(PkgRPMVersionTestCase, self).setUp() super(PkgRPMVersionTestCase, self).setUp()
self.pkgName = 'python-requests' self.pkgName = 'python-requests'
self.pkgVersion = '0.14.2-1.el6' self.pkgVersion = '0.14.2-1.el6'
self.commands_output = commands.getstatusoutput self.getoutput = pkg.getoutput
def tearDown(self): def tearDown(self):
super(PkgRPMVersionTestCase, self).tearDown() super(PkgRPMVersionTestCase, self).tearDown()
commands.getstatusoutput = self.commands_output pkg.getoutput = self.getoutput
@patch('trove.guestagent.pkg.LOG') @patch('trove.guestagent.pkg.LOG')
def test_version_no_output(self, mock_logging): def test_version_no_output(self, mock_logging):
cmd_out = '' cmd_out = ''
commands.getstatusoutput = Mock(return_value=(0, cmd_out)) pkg.getoutput = Mock(return_value=cmd_out)
self.assertIsNone(pkg.RedhatPackagerMixin().pkg_version(self.pkgName)) self.assertIsNone(pkg.RedhatPackagerMixin().pkg_version(self.pkgName))
def test_version_success(self): def test_version_success(self):
cmd_out = self.pkgVersion cmd_out = self.pkgVersion
commands.getstatusoutput = Mock(return_value=(0, cmd_out)) pkg.getoutput = Mock(return_value=cmd_out)
version = pkg.RedhatPackagerMixin().pkg_version(self.pkgName) version = pkg.RedhatPackagerMixin().pkg_version(self.pkgName)
self.assertTrue(version) self.assertTrue(version)
self.assertEqual(self.pkgVersion, version) self.assertEqual(self.pkgVersion, version)
@ -320,7 +319,7 @@ class PkgRPMInstallTestCase(trove_testtools.TestCase):
def setUp(self): def setUp(self):
super(PkgRPMInstallTestCase, self).setUp() super(PkgRPMInstallTestCase, self).setUp()
self.pkg = pkg.RedhatPackagerMixin() self.pkg = pkg.RedhatPackagerMixin()
self.commands_output = commands.getstatusoutput self.getoutput = pkg.getoutput
self.pkgName = 'packageName' self.pkgName = 'packageName'
p0 = patch('pexpect.spawn') p0 = patch('pexpect.spawn')
@ -333,7 +332,7 @@ class PkgRPMInstallTestCase(trove_testtools.TestCase):
def tearDown(self): def tearDown(self):
super(PkgRPMInstallTestCase, self).tearDown() super(PkgRPMInstallTestCase, self).tearDown()
commands.getstatusoutput = self.commands_output pkg.getoutput = self.getoutput
def test_pkg_is_installed_no_packages(self): def test_pkg_is_installed_no_packages(self):
packages = [] packages = []
@ -341,14 +340,14 @@ class PkgRPMInstallTestCase(trove_testtools.TestCase):
def test_pkg_is_installed_yes(self): def test_pkg_is_installed_yes(self):
packages = ["package1=1.0", "package2"] packages = ["package1=1.0", "package2"]
with patch.object(commands, 'getstatusoutput', MagicMock( with patch.object(pkg, 'getoutput', MagicMock(
return_value={1: "package1=1.0\n" "package2=2.0"})): return_value="package1=1.0\n" "package2=2.0")):
self.assertTrue(self.pkg.pkg_is_installed(packages)) self.assertTrue(self.pkg.pkg_is_installed(packages))
def test_pkg_is_installed_no(self): def test_pkg_is_installed_no(self):
packages = ["package1=1.0", "package2", "package3=3.0"] packages = ["package1=1.0", "package2", "package3=3.0"]
with patch.object(commands, 'getstatusoutput', MagicMock( with patch.object(pkg, 'getoutput', MagicMock(
return_value={1: "package1=1.0\n" "package2=2.0"})): return_value="package1=1.0\n" "package2=2.0")):
self.assertFalse(self.pkg.pkg_is_installed(packages)) self.assertFalse(self.pkg.pkg_is_installed(packages))
def test_permission_error(self): def test_permission_error(self):
@ -520,11 +519,11 @@ class PkgDEBFixPackageSelections(trove_testtools.TestCase):
def setUp(self): def setUp(self):
super(PkgDEBFixPackageSelections, self).setUp() super(PkgDEBFixPackageSelections, self).setUp()
self.pkg = pkg.DebianPackagerMixin() self.pkg = pkg.DebianPackagerMixin()
self.commands_output = commands.getstatusoutput self.getoutput = pkg.getoutput
def tearDown(self): def tearDown(self):
super(PkgDEBFixPackageSelections, self).tearDown() super(PkgDEBFixPackageSelections, self).tearDown()
commands.getstatusoutput = self.commands_output pkg.getoutput = self.getoutput
@patch.object(os, 'remove') @patch.object(os, 'remove')
@patch.object(pkg, 'NamedTemporaryFile') @patch.object(pkg, 'NamedTemporaryFile')
@ -533,8 +532,8 @@ class PkgDEBFixPackageSelections(trove_testtools.TestCase):
mock_remove): mock_remove):
packages = ["package1"] packages = ["package1"]
config_opts = {'option': 'some_opt'} config_opts = {'option': 'some_opt'}
commands.getstatusoutput = Mock( pkg.getoutput = Mock(
return_value=(0, "* package1/option: some_opt")) return_value="* package1/option: some_opt")
self.pkg._fix_package_selections(packages, config_opts) self.pkg._fix_package_selections(packages, config_opts)
self.assertEqual(2, mock_execute.call_count) self.assertEqual(2, mock_execute.call_count)
self.assertEqual(1, mock_remove.call_count) self.assertEqual(1, mock_remove.call_count)
@ -547,8 +546,8 @@ class PkgDEBFixPackageSelections(trove_testtools.TestCase):
mock_remove): mock_remove):
packages = ["package1"] packages = ["package1"]
config_opts = {'option': 'some_opt'} config_opts = {'option': 'some_opt'}
commands.getstatusoutput = Mock( pkg.getoutput = Mock(
return_value=(0, "* package1/option: some_opt")) return_value="* package1/option: some_opt")
self.assertRaises(pkg.PkgConfigureError, self.assertRaises(pkg.PkgConfigureError,
self.pkg._fix_package_selections, self.pkg._fix_package_selections,
packages, config_opts) packages, config_opts)