/usr/lib/python2.7/dist-packages/octaviaclient/tests/functional/base.py is in python-octaviaclient 1.4.0-2.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 | # 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 os
import re
import shlex
import subprocess
from osc_lib.tests import utils
from tempest.lib.cli import output_parser
from tempest.lib import exceptions
COMMON_DIR = os.path.dirname(os.path.abspath(__file__))
FUNCTIONAL_DIR = os.path.normpath(os.path.join(COMMON_DIR, '..'))
ROOT_DIR = os.path.normpath(os.path.join(FUNCTIONAL_DIR, '..'))
EXAMPLE_DIR = os.path.join(ROOT_DIR, 'examples')
def execute(cmd, fail_ok=False, merge_stderr=False):
"""Executes specified command for the given action."""
cmdlist = shlex.split(cmd)
result = ''
result_err = ''
stdout = subprocess.PIPE
stderr = subprocess.STDOUT if merge_stderr else subprocess.PIPE
proc = subprocess.Popen(cmdlist, stdout=stdout, stderr=stderr)
result, result_err = proc.communicate()
result = result.decode('utf-8')
if not fail_ok and proc.returncode != 0:
raise exceptions.CommandFailed(proc.returncode, cmd, result,
result_err)
return result
class TestCase(utils.TestCase):
delimiter_line = re.compile('^\+\-[\+\-]+\-\+$')
@classmethod
def openstack(cls, cmd, fail_ok=False):
"""Executes openstackclient command for the given action."""
return execute('openstack ' + cmd, fail_ok=fail_ok)
@classmethod
def get_openstack_configuration_value(cls, configuration):
opts = cls.get_opts([configuration])
return cls.openstack('configuration show ' + opts)
@classmethod
def get_openstack_extention_names(cls):
opts = cls.get_opts(['Name'])
return cls.openstack('extension list ' + opts)
@classmethod
def get_opts(cls, fields, output_format='value'):
return ' -f {0} {1}'.format(output_format,
' '.join(['-c ' + it for it in fields]))
@classmethod
def assertOutput(cls, expected, actual):
if expected != actual:
raise Exception(expected + ' != ' + actual)
@classmethod
def assertInOutput(cls, expected, actual):
if expected not in actual:
raise Exception(expected + ' not in ' + actual)
@classmethod
def assertsOutputNotNone(cls, observed):
if observed is None:
raise Exception('No output observed')
def assert_table_structure(self, items, field_names):
"""Verify that all items have keys listed in field_names."""
for item in items:
for field in field_names:
self.assertIn(field, item)
def assert_show_fields(self, show_output, field_names):
"""Verify that all items have keys listed in field_names."""
# field_names = ['name', 'description']
# show_output = [{'name': 'fc2b98d8faed4126b9e371eda045ade2'},
# {'description': 'description-821397086'}]
# this next line creates a flattened list of all 'keys' (like 'name',
# and 'description' out of the output
all_headers = [item for sublist in show_output for item in sublist]
for field_name in field_names:
self.assertIn(field_name, all_headers)
def parse_show_as_object(self, raw_output):
"""Return a dict with values parsed from cli output."""
items = self.parse_show(raw_output)
o = {}
for item in items:
o.update(item)
return o
def parse_show(self, raw_output):
"""Return list of dicts with item values parsed from cli output."""
items = []
table_ = output_parser.table(raw_output)
for row in table_['values']:
item = {}
item[row[0]] = row[1]
items.append(item)
return items
def parse_listing(self, raw_output):
"""Return list of dicts with basic item parsed from cli output."""
return output_parser.listing(raw_output)
|