diff options
Diffstat (limited to 'tempest_lib/cli/base.py')
| -rw-r--r-- | tempest_lib/cli/base.py | 410 |
1 files changed, 0 insertions, 410 deletions
diff --git a/tempest_lib/cli/base.py b/tempest_lib/cli/base.py deleted file mode 100644 index d408869..0000000 --- a/tempest_lib/cli/base.py +++ /dev/null @@ -1,410 +0,0 @@ -# Copyright 2013 OpenStack Foundation -# All Rights Reserved. -# -# Licensed under the Apache License, Version 2.0 (the "License"); you may -# not use this file except in compliance with the License. You may obtain -# a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT -# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the -# License for the specific language governing permissions and limitations -# under the License. - -import logging -import os -import shlex -import subprocess - -import six - -from tempest_lib import base -import tempest_lib.cli.output_parser -from tempest_lib import exceptions - - -LOG = logging.getLogger(__name__) - - -def execute(cmd, action, flags='', params='', fail_ok=False, - merge_stderr=False, cli_dir='/usr/bin'): - """Executes specified command for the given action. - - :param cmd: command to be executed - :type cmd: string - :param action: string of the cli command to run - :type action: string - :param flags: any optional cli flags to use - :type flags: string - :param params: string of any optional positional args to use - :type params: string - :param fail_ok: boolean if True an exception is not raised when the - cli return code is non-zero - :type fail_ok: boolean - :param merge_stderr: boolean if True the stderr buffer is merged into - stdout - :type merge_stderr: boolean - :param cli_dir: The path where the cmd can be executed - :type cli_dir: string - """ - cmd = ' '.join([os.path.join(cli_dir, cmd), - flags, action, params]) - LOG.info("running: '%s'" % cmd) - if six.PY2: - cmd = cmd.encode('utf-8') - cmd = shlex.split(cmd) - result = '' - result_err = '' - stdout = subprocess.PIPE - stderr = subprocess.STDOUT if merge_stderr else subprocess.PIPE - proc = subprocess.Popen(cmd, stdout=stdout, stderr=stderr) - result, result_err = proc.communicate() - if not fail_ok and proc.returncode != 0: - raise exceptions.CommandFailed(proc.returncode, - cmd, - result, - result_err) - if six.PY2: - return result - else: - return os.fsdecode(result) - - -class CLIClient(object): - """Class to use OpenStack official python client CLI's with auth - - :param username: The username to authenticate with - :type username: string - :param password: The password to authenticate with - :type password: string - :param tenant_name: The name of the tenant to use with the client calls - :type tenant_name: string - :param uri: The auth uri for the OpenStack Deployment - :type uri: string - :param cli_dir: The path where the python client binaries are installed. - defaults to /usr/bin - :type cli_dir: string - :param insecure: if True, --insecure is passed to python client binaries. - :type insecure: boolean - """ - - def __init__(self, username='', password='', tenant_name='', uri='', - cli_dir='', insecure=False, *args, **kwargs): - """Initialize a new CLIClient object.""" - super(CLIClient, self).__init__() - self.cli_dir = cli_dir if cli_dir else '/usr/bin' - self.username = username - self.tenant_name = tenant_name - self.password = password - self.uri = uri - self.insecure = insecure - - def nova(self, action, flags='', params='', fail_ok=False, - endpoint_type='publicURL', merge_stderr=False): - """Executes nova command for the given action. - - :param action: the cli command to run using nova - :type action: string - :param flags: any optional cli flags to use - :type flags: string - :param params: any optional positional args to use - :type params: string - :param fail_ok: if True an exception is not raised when the - cli return code is non-zero - :type fail_ok: boolean - :param endpoint_type: the type of endpoint for the service - :type endpoint_type: string - :param merge_stderr: if True the stderr buffer is merged into stdout - :type merge_stderr: boolean - """ - flags += ' --endpoint-type %s' % endpoint_type - return self.cmd_with_auth( - 'nova', action, flags, params, fail_ok, merge_stderr) - - def nova_manage(self, action, flags='', params='', fail_ok=False, - merge_stderr=False): - """Executes nova-manage command for the given action. - - :param action: the cli command to run using nova-manage - :type action: string - :param flags: any optional cli flags to use - :type flags: string - :param params: any optional positional args to use - :type params: string - :param fail_ok: if True an exception is not raised when the - cli return code is non-zero - :type fail_ok: boolean - :param merge_stderr: if True the stderr buffer is merged into stdout - :type merge_stderr: boolean - """ - return execute( - 'nova-manage', action, flags, params, fail_ok, merge_stderr, - self.cli_dir) - - def keystone(self, action, flags='', params='', fail_ok=False, - merge_stderr=False): - """Executes keystone command for the given action. - - :param action: the cli command to run using keystone - :type action: string - :param flags: any optional cli flags to use - :type flags: string - :param params: any optional positional args to use - :type params: string - :param fail_ok: if True an exception is not raised when the - cli return code is non-zero - :type fail_ok: boolean - :param merge_stderr: if True the stderr buffer is merged into stdout - :type merge_stderr: boolean - """ - return self.cmd_with_auth( - 'keystone', action, flags, params, fail_ok, merge_stderr) - - def glance(self, action, flags='', params='', fail_ok=False, - endpoint_type='publicURL', merge_stderr=False): - """Executes glance command for the given action. - - :param action: the cli command to run using glance - :type action: string - :param flags: any optional cli flags to use - :type flags: string - :param params: any optional positional args to use - :type params: string - :param fail_ok: if True an exception is not raised when the - cli return code is non-zero - :type fail_ok: boolean - :param endpoint_type: the type of endpoint for the service - :type endpoint_type: string - :param merge_stderr: if True the stderr buffer is merged into stdout - :type merge_stderr: boolean - """ - flags += ' --os-endpoint-type %s' % endpoint_type - return self.cmd_with_auth( - 'glance', action, flags, params, fail_ok, merge_stderr) - - def ceilometer(self, action, flags='', params='', - fail_ok=False, endpoint_type='publicURL', - merge_stderr=False): - """Executes ceilometer command for the given action. - - :param action: the cli command to run using ceilometer - :type action: string - :param flags: any optional cli flags to use - :type flags: string - :param params: any optional positional args to use - :type params: string - :param fail_ok: if True an exception is not raised when the - cli return code is non-zero - :type fail_ok: boolean - :param endpoint_type: the type of endpoint for the service - :type endpoint_type: string - :param merge_stderr: if True the stderr buffer is merged into stdout - :type merge_stderr: boolean - """ - flags += ' --os-endpoint-type %s' % endpoint_type - return self.cmd_with_auth( - 'ceilometer', action, flags, params, fail_ok, merge_stderr) - - def heat(self, action, flags='', params='', - fail_ok=False, endpoint_type='publicURL', merge_stderr=False): - """Executes heat command for the given action. - - :param action: the cli command to run using heat - :type action: string - :param flags: any optional cli flags to use - :type flags: string - :param params: any optional positional args to use - :type params: string - :param fail_ok: if True an exception is not raised when the - cli return code is non-zero - :type fail_ok: boolean - :param endpoint_type: the type of endpoint for the service - :type endpoint_type: string - :param merge_stderr: if True the stderr buffer is merged into stdout - :type merge_stderr: boolean - """ - flags += ' --os-endpoint-type %s' % endpoint_type - return self.cmd_with_auth( - 'heat', action, flags, params, fail_ok, merge_stderr) - - def cinder(self, action, flags='', params='', fail_ok=False, - endpoint_type='publicURL', merge_stderr=False): - """Executes cinder command for the given action. - - :param action: the cli command to run using cinder - :type action: string - :param flags: any optional cli flags to use - :type flags: string - :param params: any optional positional args to use - :type params: string - :param fail_ok: if True an exception is not raised when the - cli return code is non-zero - :type fail_ok: boolean - :param endpoint_type: the type of endpoint for the service - :type endpoint_type: string - :param merge_stderr: if True the stderr buffer is merged into stdout - :type merge_stderr: boolean - """ - flags += ' --endpoint-type %s' % endpoint_type - return self.cmd_with_auth( - 'cinder', action, flags, params, fail_ok, merge_stderr) - - def swift(self, action, flags='', params='', fail_ok=False, - endpoint_type='publicURL', merge_stderr=False): - """Executes swift command for the given action. - - :param action: the cli command to run using swift - :type action: string - :param flags: any optional cli flags to use - :type flags: string - :param params: any optional positional args to use - :type params: string - :param fail_ok: if True an exception is not raised when the - cli return code is non-zero - :type fail_ok: boolean - :param endpoint_type: the type of endpoint for the service - :type endpoint_type: string - :param merge_stderr: if True the stderr buffer is merged into stdout - :type merge_stderr: boolean - """ - flags += ' --os-endpoint-type %s' % endpoint_type - return self.cmd_with_auth( - 'swift', action, flags, params, fail_ok, merge_stderr) - - def neutron(self, action, flags='', params='', fail_ok=False, - endpoint_type='publicURL', merge_stderr=False): - """Executes neutron command for the given action. - - :param action: the cli command to run using neutron - :type action: string - :param flags: any optional cli flags to use - :type flags: string - :param params: any optional positional args to use - :type params: string - :param fail_ok: if True an exception is not raised when the - cli return code is non-zero - :type fail_ok: boolean - :param endpoint_type: the type of endpoint for the service - :type endpoint_type: string - :param merge_stderr: if True the stderr buffer is merged into stdout - :type merge_stderr: boolean - """ - flags += ' --endpoint-type %s' % endpoint_type - return self.cmd_with_auth( - 'neutron', action, flags, params, fail_ok, merge_stderr) - - def sahara(self, action, flags='', params='', - fail_ok=False, endpoint_type='publicURL', merge_stderr=True): - """Executes sahara command for the given action. - - :param action: the cli command to run using sahara - :type action: string - :param flags: any optional cli flags to use - :type flags: string - :param params: any optional positional args to use - :type params: string - :param fail_ok: if True an exception is not raised when the - cli return code is non-zero - :type fail_ok: boolean - :param endpoint_type: the type of endpoint for the service - :type endpoint_type: string - :param merge_stderr: if True the stderr buffer is merged into stdout - :type merge_stderr: boolean - """ - flags += ' --endpoint-type %s' % endpoint_type - return self.cmd_with_auth( - 'sahara', action, flags, params, fail_ok, merge_stderr) - - def openstack(self, action, flags='', params='', fail_ok=False, - merge_stderr=False): - """Executes openstack command for the given action. - - :param action: the cli command to run using openstack - :type action: string - :param flags: any optional cli flags to use - :type flags: string - :param params: any optional positional args to use - :type params: string - :param fail_ok: if True an exception is not raised when the - cli return code is non-zero - :type fail_ok: boolean - :param merge_stderr: if True the stderr buffer is merged into stdout - :type merge_stderr: boolean - """ - return self.cmd_with_auth( - 'openstack', action, flags, params, fail_ok, merge_stderr) - - def cmd_with_auth(self, cmd, action, flags='', params='', - fail_ok=False, merge_stderr=False): - """Executes given command with auth attributes appended. - - :param cmd: command to be executed - :type cmd: string - :param action: command on cli to run - :type action: string - :param flags: optional cli flags to use - :type flags: string - :param params: optional positional args to use - :type params: string - :param fail_ok: if True an exception is not raised when the cli return - code is non-zero - :type fail_ok: boolean - :param merge_stderr: if True the stderr buffer is merged into stdout - :type merge_stderr: boolean - """ - creds = ('--os-username %s --os-tenant-name %s --os-password %s ' - '--os-auth-url %s' % - (self.username, - self.tenant_name, - self.password, - self.uri)) - if self.insecure: - flags = creds + ' --insecure ' + flags - else: - flags = creds + ' ' + flags - return execute(cmd, action, flags, params, fail_ok, merge_stderr, - self.cli_dir) - - -class ClientTestBase(base.BaseTestCase): - """Base test class for testing the OpenStack client CLI interfaces.""" - - def setUp(self): - super(ClientTestBase, self).setUp() - self.clients = self._get_clients() - self.parser = tempest_lib.cli.output_parser - - def _get_clients(self): - """Abstract method to initialize CLIClient object. - - This method must be overloaded in child test classes. It should be - used to initialize the CLIClient object with the appropriate - credentials during the setUp() phase of tests. - """ - raise NotImplementedError - - def assertTableStruct(self, items, field_names): - """Verify that all items has keys listed in field_names. - - :param items: items to assert are field names in the output table - :type items: list - :param field_names: field names from the output table of the cmd - :type field_names: list - """ - for item in items: - for field in field_names: - self.assertIn(field, item) - - def assertFirstLineStartsWith(self, lines, beginning): - """Verify that the first line starts with a string - - :param lines: strings for each line of output - :type lines: list - :param beginning: verify this is at the beginning of the first line - :type beginning: string - """ - self.assertTrue(lines[0].startswith(beginning), - msg=('Beginning of first line has invalid content: %s' - % lines[:3])) |