openstack
/
trove
Code Issues Proposed changes

Abstract 'mkdir' shell commands in guestagent 

The guestagent commonly executes basic shell commands
such as 'mkdir' across multiple files and datastores.
Each datastore currently scripts its own shell calls.
Apart from cluttering the Python code with shell constructs,
it also makes maintenance of the shell calls more difficult by
requiring the same change be made across multiple lines,
source files and/or potentially even datastores.

The goal of this patch set is to provide a simple straightforward
interface for most common shell calls used across the guestagent
code while encapsulating the related shell code into a single module.

This patch set includes refactoring of 'mkdir' and related 'chown'
calls.
It also introduces unit tests ('guestagent/test_operating_system')
for the above implementations.

The shell code was moved into 'common/operating_system' module.
The replacing interface is:

def create_directory(dir_path, user=None, group=None, force=True):
    """Create a given directory and update its ownership
    (recursively) to the given user and group if any."""

The existing interface for 'chown' has been extended to maintain
consistency with 'mkdir'.

def chown(path, user, group, recursive=True, force=False):
    """Changes the owner and group of a given file.
    """

The calls also take optional keyword arguments:
    :param as_root:          Execute as root.
    :type as_root:           boolean

    :param timeout:          Number of seconds if specified,
                             default if not.
                             There is no timeout if set to None.
    :type timeout:           integer

Most occurrences of the above two shell calls ('mkdir' and 'chown')
in the guestagent code have been replaced by the above abstraction.
I did not replace native Python os.chown() and os.mkdir() calls
and calls that are executed as a part of a larger shell script
in this patch set.

Authored-By: Petr Malik <pmalik@tesora.com>
Change-Id: I0fe96301087e3bd49d937c7c489006fd9d84ca1e
Depends-On: Ib969a111d102b817c9f18437c1f36bd17e3a8a4b
Closes-Bug: 1431618
Related-Bug: 1438430
This commit is contained in:
Jenkins 2015-03-12 10:20:51 +00:00 committed by Petr Malik
parent 77df95af1c
commit 0397f9c6a0
18 changed files with 323 additions and 81 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

86
trove/guestagent/common/operating_system.py
View File

@ -200,12 +200,90 @@ def service_discovery(service_candidates):
return result
def update_owner(user, group, path):
def create_directory(dir_path, user=None, group=None, force=True, **kwargs):
"""Create a given directory and update its ownership
(recursively) to the given user and group if any.
seealso:: _execute_shell_cmd for valid optional keyword arguments.
:param dir_path: Path to the created directory.
:type dir_path: string
:param user: Owner.
:type user: string
:param group: Group.
:type group: string
:param force: No error if existing, make parent directories
as needed.
:type force: boolean
:raises: :class:`UnprocessableEntity` if dir_path not given.
"""
Changes the owner and group for the path (recursively)
if dir_path:
_create_directory(dir_path, force, **kwargs)
if user or group:
chown(dir_path, user, group, **kwargs)
else:
raise exception.UnprocessableEntity(
_("Cannot create a blank directory."))
def chown(path, user, group, recursive=True, force=False, **kwargs):
"""Changes the owner and group of a given file.
seealso:: _execute_shell_cmd for valid optional keyword arguments.
:param path: Path to the modified file.
:type path: string
:param user: Owner.
:type user: string
:param group: Group.
:type group: string
:param recursive: Operate on files and directories recursively.
:type recursive: boolean
:param force: Suppress most error messages.
:type force: boolean
:raises: :class:`UnprocessableEntity` if path not given.
:raises: :class:`UnprocessableEntity` if owner/group not given.
"""
utils.execute_with_timeout("chown", "-R", "%s:%s" % (user, group), path,
run_as_root=True, root_helper="sudo")
if not path:
raise exception.UnprocessableEntity(
_("Cannot change ownership of a blank file or directory."))
if not user and not group:
raise exception.UnprocessableEntity(
_("Please specify owner or group, or both."))
owner_group_modifier = _build_user_group_pair(user, group)
options = (('f', force), ('R', recursive))
_execute_shell_cmd('chown', options, owner_group_modifier, path, **kwargs)
def _build_user_group_pair(user, group):
return "%s:%s" % tuple((v if v else '') for v in (user, group))
def _create_directory(dir_path, force=True, **kwargs):
"""Create a given directory.
:param dir_path: Path to the created directory.
:type dir_path: string
:param force: No error if existing, make parent directories
as needed.
:type force: boolean
"""
options = (('p', force),)
_execute_shell_cmd('mkdir', options, dir_path, **kwargs)
def chmod(path, mode, recursive=True, force=False, **kwargs):

8
trove/guestagent/datastore/experimental/cassandra/service.py
View File

@ -64,8 +64,7 @@ class CassandraApp(object):
def init_storage_structure(self, mount_point):
try:
cmd = system.INIT_FS % mount_point
utils.execute_with_timeout(cmd, shell=True)
operating_system.create_directory(mount_point, as_root=True)
except exception.ProcessExecutionError:
LOG.exception(_("Error while initiating storage structure."))
@ -142,8 +141,9 @@ class CassandraApp(object):
#TODO(denis_makogon): figure out the dynamic way to discover
# configs owner since it can cause errors if there is
# no cassandra user in operating system
execute_function("sudo", "chown",
"cassandra:cassandra", system.CASSANDRA_CONF)
operating_system.chown(system.CASSANDRA_CONF,
'cassandra', 'cassandra', recursive=False,
as_root=True)
operating_system.chmod(system.CASSANDRA_CONF,
FileMode.ADD_READ_ALL, as_root=True)
except Exception:

1
trove/guestagent/datastore/experimental/cassandra/system.py
View File

@ -22,7 +22,6 @@ CASSANDRA_CONF = "/etc/cassandra/cassandra.yaml"
CASSANDRA_TEMP_CONF = "/tmp/cassandra.yaml"
CASSANDRA_TEMP_DIR = "/tmp/cassandra"
INIT_FS = "sudo mkdir -p %s"
ENABLE_CASSANDRA_ON_BOOT = "sudo update-rc.d cassandra enable"
DISABLE_CASSANDRA_ON_BOOT = "sudo update-rc.d cassandra disable"

17
trove/guestagent/datastore/experimental/couchbase/service.py
View File

@ -68,9 +68,8 @@ class CouchbaseApp(object):
mount_point = CONF.couchbase.mount_point
try:
LOG.info(_('Couchbase Server change data dir path.'))
operating_system.update_owner('couchbase',
'couchbase',
mount_point)
operating_system.chown(mount_point, 'couchbase', 'couchbase',
as_root=True)
pwd = CouchbaseRootAccess.get_password()
utils.execute_with_timeout(
(system.cmd_node_init
@ -104,11 +103,8 @@ class CouchbaseApp(object):
"""
LOG.debug('Installing Couchbase Server. Creating %s' %
system.COUCHBASE_CONF_DIR)
utils.execute_with_timeout('mkdir',
'-p',
system.COUCHBASE_CONF_DIR,
run_as_root=True,
root_helper='sudo')
operating_system.create_directory(system.COUCHBASE_CONF_DIR,
as_root=True)
pkg_opts = {}
packager.pkg_install(packages, pkg_opts, system.TIME_OUT)
self.start_db()
@ -316,9 +312,8 @@ class CouchbaseRootAccess(object):
self.write_password_to_file(root_password)
def write_password_to_file(self, root_password):
utils.execute_with_timeout('mkdir', '-p', system.COUCHBASE_CONF_DIR,
run_as_root=True, root_helper='sudo')
operating_system.create_directory(system.COUCHBASE_CONF_DIR,
as_root=True)
try:
tempfd, tempname = tempfile.mkstemp()
os.fchmod(tempfd, stat.S_IRUSR | stat.S_IWUSR)

16
trove/guestagent/datastore/experimental/couchdb/service.py
View File

@ -78,17 +78,17 @@ class CouchDBApp(object):
"""
try:
LOG.debug("Changing permissions.")
operating_system.update_owner(
'couchdb', 'couchdb', COUCHDB_LIB_DIR
operating_system.chown(
COUCHDB_LIB_DIR, 'couchdb', 'couchdb', as_root=True
)
operating_system.update_owner(
'couchdb', 'couchdb', COUCHDB_LOG_DIR
operating_system.chown(
COUCHDB_LOG_DIR, 'couchdb', 'couchdb', as_root=True
)
operating_system.update_owner(
'couchdb', 'couchdb', COUCHDB_BIN_DIR
operating_system.chown(
COUCHDB_BIN_DIR, 'couchdb', 'couchdb', as_root=True
)
operating_system.update_owner(
'couchdb', 'couchdb', COUCHDB_CONFIG_DIR
operating_system.chown(
COUCHDB_CONFIG_DIR, 'couchdb', 'couchdb', as_root=True
)
operating_system.chmod(COUCHDB_LIB_DIR, FileMode.ADD_GRP_RW,
as_root=True)

11
trove/guestagent/datastore/experimental/db2/service.py
View File

@ -17,6 +17,7 @@ from trove.common import cfg
from trove.common import exception
from trove.common import instance as rd_instance
from trove.common import utils as utils
from trove.guestagent.common import operating_system
from trove.guestagent.datastore import service
from trove.guestagent.datastore.experimental.db2 import system
from trove.guestagent.db import models
@ -55,12 +56,10 @@ class DB2App(object):
"""
LOG.debug("Changing ownership of the DB2 data directory.")
try:
utils.execute_with_timeout(
system.CHANGE_DB_DIR_OWNER % {'datadir': mount_point},
shell=True)
utils.execute_with_timeout(
system.CHANGE_DB_DIR_GROUP_OWNER % {'datadir': mount_point},
shell=True)
operating_system.chown(mount_point,
system.DB2_INSTANCE_OWNER,
system.DB2_INSTANCE_OWNER,
recursive=False, as_root=True)
except exception.ProcessExecutionError:
raise RuntimeError(_(
"Command to change ownership of DB2 data directory failed."))

3
trove/guestagent/datastore/experimental/db2/system.py
View File

@ -23,9 +23,6 @@ START_DB2 = "db2start"
STOP_DB2 = "db2 force application all; db2 terminate; db2stop"
DB2_STATUS = ("ps -ef | grep " + DB2_INSTANCE_OWNER + " | grep db2sysc |"
"grep -v grep | wc -l")
CHANGE_DB_DIR_OWNER = "sudo chown " + DB2_INSTANCE_OWNER + " %(datadir)s"
CHANGE_DB_DIR_GROUP_OWNER = (
"sudo chgrp " + DB2_INSTANCE_OWNER + " %(datadir)s")
CREATE_DB_COMMAND = "db2 create database %(dbname)s"
DELETE_DB_COMMAND = "db2 drop database %(dbname)s"
LIST_DB_COMMAND = (

5
trove/guestagent/datastore/experimental/mongodb/manager.py
View File

@ -71,8 +71,9 @@ class Manager(periodic_task.PeriodicTasks):
if os.path.exists(system.MONGODB_MOUNT_POINT):
device.migrate_data(mount_point)
device.mount(mount_point)
operating_system.update_owner(system.MONGO_USER,
system.MONGO_USER, mount_point)
operating_system.chown(mount_point,
system.MONGO_USER, system.MONGO_USER,
as_root=True)
LOG.debug("Mounted the volume %(path)s as %(mount)s." %
{'path': device_path, "mount": mount_point})

12
trove/guestagent/datastore/experimental/postgresql/service/config.py
View File

@ -69,10 +69,8 @@ class PgSqlConfig(PgSqlProcess):
)
with open('/tmp/pgsql_config', 'w+') as config_file:
config_file.write(configuration)
utils.execute_with_timeout(
'sudo', 'chown', 'postgres', '/tmp/pgsql_config',
timeout=30,
)
operating_system.chown('/tmp/pgsql_config', 'postgres', None,
recursive=False, as_root=True)
operating_system.move('/tmp/pgsql_config', config_location, timeout=30,
as_root=True)
@ -95,10 +93,8 @@ class PgSqlConfig(PgSqlProcess):
config_file.write(out)
config_file.write("host all all 0.0.0.0/0 md5\n")
utils.execute_with_timeout(
'sudo', 'chown', 'postgres', '/tmp/pgsql_hba_config',
timeout=30,
)
operating_system.chown('/tmp/pgsql_hba_config',
'postgres', None, recursive=False, as_root=True)
operating_system.move('/tmp/pgsql_hba_config', PGSQL_HBA_CONFIG.format(
version=self._get_psql_version(),
), timeout=30, as_root=True)

3
trove/guestagent/datastore/experimental/redis/manager.py
View File

@ -97,7 +97,8 @@ class Manager(periodic_task.PeriodicTasks):
device.unmount_device(device_path)
device.format()
device.mount(mount_point)
operating_system.update_owner('redis', 'redis', mount_point)
operating_system.chown(mount_point, 'redis', 'redis',
as_root=True)
LOG.debug('Mounted the volume.')
app.install_if_needed(packages)
LOG.info(_('Writing redis configuration.'))

6
trove/guestagent/datastore/experimental/redis/service.py
View File

@ -162,11 +162,7 @@ class RedisApp(object):
LOG.debug('Installing redis server.')
msg = "Creating %s." % system.REDIS_CONF_DIR
LOG.debug(msg)
utils.execute_with_timeout('mkdir',
'-p',
system.REDIS_CONF_DIR,
run_as_root=True,
root_helper='sudo')
operating_system.create_directory(system.REDIS_CONF_DIR, as_root=True)
pkg_opts = {}
packager.pkg_install(packages, pkg_opts, TIME_OUT)
self.start_redis()

3
trove/guestagent/datastore/mysql/service.py
View File

@ -654,8 +654,7 @@ class MySqlApp(object):
def _create_mysql_confd_dir(self):
conf_dir = "/etc/mysql/conf.d"
LOG.debug("Creating %s." % conf_dir)
command = "sudo mkdir -p %s" % conf_dir
utils.execute_with_timeout(command, shell=True)
operating_system.create_directory(conf_dir, as_root=True)
def _enable_mysql_on_boot(self):
LOG.debug("Enabling MySQL on boot.")

10
trove/guestagent/strategies/restore/mysql_impl.py
View File

@ -216,10 +216,8 @@ class InnoBackupEx(base.RestoreRunner, MySQLRestoreMixin):
def post_restore(self):
self._run_prepare()
utils.execute_with_timeout("chown", "-R", "-f", "mysql",
self.restore_location,
root_helper="sudo",
run_as_root=True)
operating_system.chown(self.restore_location, 'mysql', None,
force=True, as_root=True)
self._delete_old_binlogs()
self.reset_root_password()
app = dbaas.MySqlApp(dbaas.MySqlAppStatus.get())
@ -296,9 +294,7 @@ class InnoBackupExIncremental(InnoBackupEx):
# just use the checksum for the incremental path as it is
# sufficiently unique /var/lib/mysql/<checksum>
incremental_dir = os.path.join(self.restore_location, checksum)
utils.execute("mkdir", "-p", incremental_dir,
root_helper="sudo",
run_as_root=True)
operating_system.create_directory(incremental_dir, as_root=True)
command = self._incremental_restore_cmd(incremental_dir)
else:
# The parent (full backup) use the same command from InnobackupEx

3
trove/guestagent/volume.py
View File

@ -23,6 +23,7 @@ from trove.common.exception import GuestError
from trove.common.exception import ProcessExecutionError
from trove.openstack.common import log as logging
from trove.common.i18n import _
from trove.guestagent.common import operating_system
TMP_MOUNT_POINT = "/mnt/volume"
@ -181,7 +182,7 @@ class VolumeMountPoint(object):
def mount(self):
if not os.path.exists(self.mount_point):
utils.execute("sudo", "mkdir", "-p", self.mount_point)
operating_system.create_directory(self.mount_point, as_root=True)
LOG.debug("Mounting volume. Device path:{0}, mount_point:{1}, "
"volume_type:{2}, mount options:{3}".format(
self.device_path, self.mount_point, self.volume_fstype,

24
trove/tests/unittests/guestagent/test_dbaas.py
View File

@ -1341,12 +1341,10 @@ class MySqlAppInstallTest(MySqlAppTest):
self.assertEqual(remove_expected, remove_root[0].text,
"Remove root queries are not the same")
@patch.object(utils, 'execute_with_timeout')
def test__create_mysql_confd_dir(self, mock_execute):
@patch.object(operating_system, 'create_directory')
def test__create_mysql_confd_dir(self, mkdir_mock):
self.mySqlApp._create_mysql_confd_dir()
self.assertEqual(1, mock_execute.call_count)
mock_execute.assert_called_with('sudo mkdir -p /etc/mysql/conf.d',
shell=True)
mkdir_mock.assert_called_once_with('/etc/mysql/conf.d', as_root=True)
@patch.object(operating_system, 'move')
def test__clear_mysql_config(self, mock_move):
@ -2237,8 +2235,6 @@ class CassandraDBAppTest(testtools.TestCase):
# this test verifies not only that the write_config
# method properly invoked execute, but also that it properly
# attempted to unlink the file (as a result of the exception)
execute_with_timeout = Mock(
side_effect=ProcessExecutionError('some exception'))
mock_unlink = Mock(return_value=0)
@ -2249,11 +2245,12 @@ class CassandraDBAppTest(testtools.TestCase):
configuration = 'this is my configuration'
with patch('trove.guestagent.common.operating_system.move'):
with patch('trove.guestagent.common.operating_system.move',
side_effect=ProcessExecutionError('some exception')):
self.assertRaises(ProcessExecutionError,
self.cassandra.write_config,
config_contents=configuration,
execute_function=execute_with_timeout,
execute_function=Mock(),
mkstemp_function=mock_mkstemp,
unlink_function=mock_unlink)
@ -2263,8 +2260,8 @@ class CassandraDBAppTest(testtools.TestCase):
os.unlink(temp_config_name)
@patch.multiple('trove.guestagent.common.operating_system',
chmod=DEFAULT, move=DEFAULT)
def test_cassandra_write_config(self, chmod, move):
chown=DEFAULT, chmod=DEFAULT, move=DEFAULT)
def test_cassandra_write_config(self, chown, chmod, move):
# ensure that write_config creates a temporary file, and then
# moves the file to the final place. Also validate the
# contents of the file written.
@ -2284,8 +2281,9 @@ class CassandraDBAppTest(testtools.TestCase):
move.assert_called_with(temp_config_name, cass_system.CASSANDRA_CONF,
as_root=True)
mock_execute.assert_called_with("sudo", "chown", "cassandra:cassandra",
cass_system.CASSANDRA_CONF)
chown.assert_called_with(cass_system.CASSANDRA_CONF,
"cassandra", "cassandra", recursive=False,
as_root=True)
chmod.assert_called_with(
cass_system.CASSANDRA_CONF, FileMode.ADD_READ_ALL, as_root=True)

186
trove/tests/unittests/guestagent/test_operating_system.py
View File

@ -334,6 +334,192 @@ class TestOperatingSystem(testtools.TestCase):
"Got unknown keyword args: {'_unknown_kw': 0}"),
'source', 'destination', _unknown_kw=0)
def test_chown(self):
self._assert_execute_call(
[['chown', '-R', 'usr:grp', 'path']],
[{'run_as_root': True, 'root_helper': 'sudo'}],
operating_system.chown, None, 'path', 'usr', 'grp', as_root=True)
self._assert_execute_call(
[['chown', 'usr:grp', 'path']],
[{'run_as_root': True, 'root_helper': 'sudo'}],
operating_system.chown, None,
'path', 'usr', 'grp', recursive=False, as_root=True)
self._assert_execute_call(
[['chown', '-f', '-R', 'usr:grp', 'path']],
[{'run_as_root': True, 'root_helper': 'sudo'}],
operating_system.chown, None,
'path', 'usr', 'grp', force=True, as_root=True)
self._assert_execute_call(
[['chown', '-R', ':grp', 'path']],
[{'run_as_root': True, 'root_helper': 'sudo'}],
operating_system.chown, None, 'path', '', 'grp', as_root=True)
self._assert_execute_call(
[['chown', '-R', 'usr:', 'path']],
[{'run_as_root': True, 'root_helper': 'sudo'}],
operating_system.chown, None, 'path', 'usr', '', as_root=True)
self._assert_execute_call(
[['chown', '-R', ':grp', 'path']],
[{'run_as_root': True, 'root_helper': 'sudo'}],
operating_system.chown, None, 'path', None, 'grp', as_root=True)
self._assert_execute_call(
[['chown', '-R', 'usr:', 'path']],
[{'run_as_root': True, 'root_helper': 'sudo'}],
operating_system.chown, None, 'path', 'usr', None, as_root=True)
self._assert_execute_call(
[['chown', '-R', 'usr:', 'path']],
[{'timeout': 100}],
operating_system.chown, None,
'path', 'usr', None, timeout=100)
self._assert_execute_call(
[['chown', '-R', 'usr:', 'path']],
[{'run_as_root': True, 'root_helper': 'sudo',
'timeout': None}],
operating_system.chown, None,
'path', 'usr', None, timeout=None, as_root=True)
self._assert_execute_call(
None, None,
operating_system.chown,
ExpectedException(exception.UnprocessableEntity,
"Cannot change ownership of a blank file."),
'', 'usr', 'grp')
self._assert_execute_call(
None, None,
operating_system.chown,
ExpectedException(exception.UnprocessableEntity,
"Cannot change ownership of a blank file."),
None, 'usr', 'grp')
self._assert_execute_call(
None, None,
operating_system.chown,
ExpectedException(exception.UnprocessableEntity,
"Please specify owner or group, or both."),
'path', '', '')
self._assert_execute_call(
None, None,
operating_system.chown,
ExpectedException(exception.UnprocessableEntity,
"Please specify owner or group, or both."),
'path', None, None)
self._assert_execute_call(
None, None,
operating_system.chown,
ExpectedException(exception.UnprocessableEntity,
"Cannot change ownership of a blank file."),
None, None, None)
self._assert_execute_call(
None, None,
operating_system.chown,
ExpectedException(exception.UnprocessableEntity,
"Cannot change ownership of a blank file."),
'', '', '')
self._assert_execute_call(
None, None,
operating_system.chown,
ExpectedException(UnknownArgumentError,
"Got unknown keyword args: {'_unknown_kw': 0}"),
'path', 'usr', None, _unknown_kw=0)
def test_create_directory(self):
self._assert_execute_call(
[['mkdir', '-p', 'path']],
[{'run_as_root': True, 'root_helper': 'sudo'}],
operating_system.create_directory, None, 'path', as_root=True)
self._assert_execute_call(
[['mkdir', 'path']],
[{'run_as_root': True, 'root_helper': 'sudo'}],
operating_system.create_directory, None, 'path', force=False,
as_root=True)
self._assert_execute_call(
[['mkdir', '-p', 'path'], ['chown', '-R', 'usr:grp', 'path']],
[{'run_as_root': True, 'root_helper': 'sudo'},
{'run_as_root': True, 'root_helper': 'sudo'}],
operating_system.create_directory, None,
'path', user='usr', group='grp', as_root=True)
self._assert_execute_call(
[['mkdir', '-p', 'path'], ['chown', '-R', ':grp', 'path']],
[{'run_as_root': True, 'root_helper': 'sudo'},
{'run_as_root': True, 'root_helper': 'sudo'}],
operating_system.create_directory, None, 'path', group='grp',
as_root=True)
self._assert_execute_call(
[['mkdir', '-p', 'path'], ['chown', '-R', 'usr:', 'path']],
[{'run_as_root': True, 'root_helper': 'sudo'},
{'run_as_root': True, 'root_helper': 'sudo'}],
operating_system.create_directory, None, 'path', user='usr',
as_root=True)
self._assert_execute_call(
[['mkdir', '-p', 'path'], ['chown', '-R', 'usr:', 'path']],
[{'timeout': 100}, {'timeout': 100}],
operating_system.create_directory, None,
'path', user='usr', timeout=100)
self._assert_execute_call(
[['mkdir', '-p', 'path'], ['chown', '-R', 'usr:', 'path']],
[{'run_as_root': True, 'root_helper': 'sudo', 'timeout': None},
{'run_as_root': True, 'root_helper': 'sudo', 'timeout': None}],
operating_system.create_directory, None,
'path', user='usr', timeout=None, as_root=True)
self._assert_execute_call(
[['mkdir', '-p', 'path'], ['chown', '-R', 'usr:', 'path']],
[{'run_as_root': True, 'root_helper': 'sudo'},
{'run_as_root': True, 'root_helper': 'sudo'}],
operating_system.create_directory, None,
'path', user='usr', group='', as_root=True)
self._assert_execute_call(
[['mkdir', '-p', 'path'], ['chown', '-R', ':grp', 'path']],
[{'run_as_root': True, 'root_helper': 'sudo'},
{'run_as_root': True, 'root_helper': 'sudo'}],
operating_system.create_directory, None,
'path', user='', group='grp', as_root=True)
self._assert_execute_call(
[['mkdir', '-p', 'path']],
[{'run_as_root': True, 'root_helper': 'sudo'}],
operating_system.create_directory, None, 'path', user='', group='',
as_root=True)
self._assert_execute_call(
None, None,
operating_system.create_directory,
ExpectedException(exception.UnprocessableEntity,
"Cannot create a blank directory."),
'', user='usr', group='grp')
self._assert_execute_call(
None, None,
operating_system.create_directory,
ExpectedException(exception.UnprocessableEntity,
"Cannot create a blank directory."), None)
self._assert_execute_call(
None, None,
operating_system.create_directory,
ExpectedException(UnknownArgumentError,
"Got unknown keyword args: {'_unknown_kw': 0}"),
'path', _unknown_kw=0)
def _assert_execute_call(self, exec_args, exec_kwargs,
fun, return_value, *args, **kwargs):
"""

6
trove/tests/unittests/guestagent/test_redis_manager.py
View File

@ -82,7 +82,7 @@ class RedisGuestAgentManagerTest(testtools.TestCase):
redis_service.RedisApp.start_redis = MagicMock(return_value=None)
redis_service.RedisApp.install_if_needed = MagicMock(return_value=None)
redis_service.RedisApp.write_config = MagicMock(return_value=None)
operating_system.update_owner = MagicMock(return_value=None)
operating_system.chown = MagicMock(return_value=None)
redis_service.RedisApp.restart = MagicMock(return_value=None)
mock_status.begin_install = MagicMock(return_value=None)
VolumeDevice.format = MagicMock(return_value=None)
@ -103,8 +103,8 @@ class RedisGuestAgentManagerTest(testtools.TestCase):
VolumeDevice.format.assert_any_call()
redis_service.RedisApp.install_if_needed.assert_any_call(self.packages)
redis_service.RedisApp.write_config.assert_any_call(None)
operating_system.update_owner.assert_any_call(
'redis', 'redis', mount_point)
operating_system.chown.assert_any_call(
mount_point, 'redis', 'redis', as_root=True)
redis_service.RedisApp.restart.assert_any_call()
def test_restart(self):

4
trove/tests/unittests/guestagent/test_volume.py
View File

@ -207,12 +207,12 @@ class VolumeMountPointTest(testtools.TestCase):
os.path.exists = MagicMock(return_value=False)
fake_spawn = _setUp_fake_spawn()
utils.execute = Mock()
utils.execute_with_timeout = Mock()
self.volumeMountPoint.mount()
self.assertEqual(1, os.path.exists.call_count)
self.assertEqual(1, utils.execute.call_count)
self.assertEqual(1, utils.execute_with_timeout.call_count)
self.assertEqual(1, fake_spawn.expect.call_count)
os.path.exists = origin_