198a486413
Previously fucntional tests for network segment feature are skipped as neutron 'segment' API extension was disabled in the gate. We now enable neutron 'segment' API extension, so we can safely drop the check for the segment extension from the test code. Also setup code in test_network_segment is moved from setUpClass to setUp. There is no good reason to do them in setUpClass and having them in setUp simplifies the test code. no user once this commit is applied. Change-Id: I183310b94d9b6d7f4311a3859b59dc22d36440db
131 lines
4.5 KiB
Python
131 lines
4.5 KiB
Python
# 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
|
|
import testtools
|
|
|
|
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
|
|
|
|
|
|
def is_service_enabled(service):
|
|
"""Ask client cloud if service is available"""
|
|
try:
|
|
ret = execute('openstack service show -f value -c enabled ' + service)
|
|
except exceptions.CommandFailed:
|
|
# We get here for multiple reasons, all of them mean that a working
|
|
# service is not available
|
|
return False
|
|
|
|
return "True" in ret
|
|
|
|
|
|
class TestCase(testtools.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_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)
|