from functest.opnfv_tests.openstack.tempest import conf_utils
from functest.utils.constants import CONST
import functest.utils.functest_utils as ft_utils
+import functest.utils.openstack_utils as os_utils
# logging configuration """
LOGGER = logging.getLogger(__name__)
-class RefstackClient(testcase.OSGCTestCase):
+class RefstackClient(testcase.TestCase):
"""RefstackClient testcase implementation class."""
def __init__(self, **kwargs):
self.conf_path = pkg_resources.resource_filename(
'functest',
'opnfv_tests/openstack/refstack_client/refstack_tempest.conf')
+ self.tempestconf = None
self.functest_test = pkg_resources.resource_filename(
'functest', 'opnfv_tests')
self.defcore_list = 'openstack/refstack_client/defcore.txt'
os.makedirs(conf_utils.REFSTACK_RESULTS_DIR)
try:
- tempestconf = TempestConf()
- tempestconf.generate_tempestconf()
+ self.tempestconf = TempestConf()
+ self.tempestconf.generate_tempestconf()
self.run_defcore_default()
self.parse_refstack_result()
res = testcase.TestCase.EX_OK
except Exception:
LOGGER.exception("Error with run")
res = testcase.TestCase.EX_RUN_ERROR
+ finally:
+ self.tempestconf.clean()
self.stop_time = time.time()
return res
return res
+ def create_snapshot(self):
+ """
+ Run the Tempest cleanup utility to initialize OS state.
+ For details, see https://docs.openstack.org/tempest/latest/cleanup.html
+
+ :return: TestCase.EX_OK
+ """
+ LOGGER.info("Initializing the saved state of the OpenStack deployment")
+
+ # Make sure that the verifier is configured
+ conf_utils.configure_verifier(self.tempestconf.DEPLOYMENT_DIR)
+
+ os_utils.init_tempest_cleanup(
+ self.tempestconf.DEPLOYMENT_DIR, 'tempest.conf',
+ os.path.join(conf_utils.REFSTACK_RESULTS_DIR,
+ "tempest-cleanup-init.log")
+ )
+
+ return super(RefstackClient, self).create_snapshot()
+
+ def clean(self):
+ """
+ Run the Tempest cleanup utility to delete and destroy OS resources.
+ For details, see https://docs.openstack.org/tempest/latest/cleanup.html
+ """
+ LOGGER.info("Initializing the saved state of the OpenStack deployment")
+
+ os_utils.init_tempest_cleanup(
+ self.tempestconf.DEPLOYMENT_DIR, 'tempest.conf',
+ os.path.join(conf_utils.REFSTACK_RESULTS_DIR,
+ "tempest-cleanup.log")
+ )
+
+ return super(RefstackClient, self).clean()
+
class RefstackClientParser(object): # pylint: disable=too-few-public-methods
"""Command line argument parser helper."""
from functest.opnfv_tests.openstack.tempest import conf_utils
from functest.utils import openstack_utils
from functest.utils.constants import CONST
+from functest.opnfv_tests.openstack.tempest.tempest \
+ import TempestResourcesManager
""" logging configuration """
logger = logging.getLogger(__name__)
class TempestConf(object):
- def __init__(self):
+ def __init__(self, **kwargs):
self.VERIFIER_ID = conf_utils.get_verifier_id()
self.VERIFIER_REPO_DIR = conf_utils.get_verifier_repo_dir(
self.VERIFIER_ID)
self.confpath = pkg_resources.resource_filename(
'functest',
'opnfv_tests/openstack/refstack_client/refstack_tempest.conf')
+ self.resources = TempestResourcesManager(**kwargs)
def generate_tempestconf(self):
try:
openstack_utils.source_credentials(
CONST.__getattribute__('openstack_creds'))
- img_flavor_dict = conf_utils.create_tempest_resources(
- use_custom_images=True, use_custom_flavors=True)
+ resources = self.resources.create(create_project=True,
+ use_custom_images=True,
+ use_custom_flavors=True)
conf_utils.configure_tempest_defcore(
- self.DEPLOYMENT_DIR, img_flavor_dict)
+ self.DEPLOYMENT_DIR,
+ image_id=resources.get("image_id"),
+ flavor_id=resources.get("flavor_id"),
+ image_id_alt=resources.get("image_id_alt"),
+ flavor_id_alt=resources.get("flavor_id_alt"),
+ tenant_id=resources.get("project_id"))
except Exception as e:
logger.error("error with generating refstack client "
"reference tempest conf file: %s", e)
except Exception as e:
logger.error('Error with run: %s', e)
+ def clean(self):
+ self.resources.cleanup()
+
def main():
logging.basicConfig()
logger = logging.getLogger(__name__)
-def create_tempest_resources(use_custom_images=False,
- use_custom_flavors=False):
-
- logger.debug("Creating private network for Tempest suite")
- network_dic = os_utils.create_shared_network_full(
- CONST.__getattribute__('tempest_private_net_name'),
- CONST.__getattribute__('tempest_private_subnet_name'),
- CONST.__getattribute__('tempest_router_name'),
- CONST.__getattribute__('tempest_private_subnet_cidr'))
- if network_dic is None:
- raise Exception('Failed to create private network')
-
- image_id = ""
- image_id_alt = ""
- flavor_id = ""
- flavor_id_alt = ""
-
- if (CONST.__getattribute__('tempest_use_custom_images') or
- use_custom_images):
- # adding alternative image should be trivial should we need it
- logger.debug("Creating image for Tempest suite")
- _, image_id = os_utils.get_or_create_image(
- CONST.__getattribute__('openstack_image_name'),
- GLANCE_IMAGE_PATH,
- CONST.__getattribute__('openstack_image_disk_format'))
- if image_id is None:
- raise Exception('Failed to create image')
-
- if use_custom_images:
- logger.debug("Creating 2nd image for Tempest suite")
- _, image_id_alt = os_utils.get_or_create_image(
- CONST.__getattribute__('openstack_image_name_alt'),
- GLANCE_IMAGE_PATH,
- CONST.__getattribute__('openstack_image_disk_format'))
- if image_id_alt is None:
- raise Exception('Failed to create image')
-
- if (CONST.__getattribute__('tempest_use_custom_flavors') or
- use_custom_flavors):
- # adding alternative flavor should be trivial should we need it
- logger.debug("Creating flavor for Tempest suite")
- _, flavor_id = os_utils.get_or_create_flavor(
- CONST.__getattribute__('openstack_flavor_name'),
- CONST.__getattribute__('openstack_flavor_ram'),
- CONST.__getattribute__('openstack_flavor_disk'),
- CONST.__getattribute__('openstack_flavor_vcpus'))
- if flavor_id is None:
- raise Exception('Failed to create flavor')
-
- if use_custom_flavors:
- logger.debug("Creating 2nd flavor for tempest_defcore")
- _, flavor_id_alt = os_utils.get_or_create_flavor(
- CONST.__getattribute__('openstack_flavor_name_alt'),
- CONST.__getattribute__('openstack_flavor_ram'),
- CONST.__getattribute__('openstack_flavor_disk'),
- CONST.__getattribute__('openstack_flavor_vcpus'))
- if flavor_id_alt is None:
- raise Exception('Failed to create flavor')
-
- img_flavor_dict = {}
- img_flavor_dict['image_id'] = image_id
- img_flavor_dict['image_id_alt'] = image_id_alt
- img_flavor_dict['flavor_id'] = flavor_id
- img_flavor_dict['flavor_id_alt'] = flavor_id_alt
-
- return img_flavor_dict
-
-
-def create_tenant_user():
- keystone_client = os_utils.get_keystone_client()
-
- logger.debug("Creating tenant and user for Tempest suite")
- tenant_id = os_utils.create_tenant(
- keystone_client,
- CONST.__getattribute__('tempest_identity_tenant_name'),
- CONST.__getattribute__('tempest_identity_tenant_description'))
- if not tenant_id:
- logger.error("Failed to create %s tenant"
- % CONST.__getattribute__('tempest_identity_tenant_name'))
-
- user_id = os_utils.create_user(
- keystone_client,
- CONST.__getattribute__('tempest_identity_user_name'),
- CONST.__getattribute__('tempest_identity_user_password'),
- None, tenant_id)
- if not user_id:
- logger.error("Failed to create %s user" %
- CONST.__getattribute__('tempest_identity_user_name'))
-
- return tenant_id
-
-
def get_verifier_id():
"""
- Returns verifer id for current Tempest
+ Returns verifier id for current Tempest
"""
cmd = ("rally verify list-verifiers | awk '/" +
CONST.__getattribute__('tempest_deployment_name') +
def get_verifier_repo_dir(verifier_id):
"""
- Returns installed verfier repo directory for Tempest
+ Returns installed verifier repo directory for Tempest
"""
if not verifier_id:
verifier_id = get_verifier_id()
os.path.join(TEMPEST_RESULTS_DIR, 'tempest.conf'))
-def configure_tempest(deployment_dir, IMAGE_ID=None, FLAVOR_ID=None,
- MODE=None):
+def configure_tempest(deployment_dir, image_id=None, flavor_id=None,
+ mode=None):
"""
Calls rally verify and updates the generated tempest.conf with
given parameters
"""
conf_file = configure_verifier(deployment_dir)
- configure_tempest_update_params(conf_file,
- IMAGE_ID, FLAVOR_ID)
+ configure_tempest_update_params(conf_file, image_id, flavor_id)
-def configure_tempest_defcore(deployment_dir, img_flavor_dict):
+def configure_tempest_defcore(deployment_dir, image_id, flavor_id,
+ image_id_alt, flavor_id_alt, tenant_id):
"""
Add/update needed parameters into tempest.conf file
"""
conf_file = configure_verifier(deployment_dir)
- configure_tempest_update_params(conf_file,
- img_flavor_dict.get("image_id"),
- img_flavor_dict.get("flavor_id"))
+ configure_tempest_update_params(conf_file, image_id, flavor_id)
logger.debug("Updating selected tempest.conf parameters for defcore...")
config = ConfigParser.RawConfigParser()
config.set('DEFAULT', 'log_file', '{}/tempest.log'.format(deployment_dir))
config.set('oslo_concurrency', 'lock_path',
'{}/lock_files'.format(deployment_dir))
- generate_test_accounts_file()
+ generate_test_accounts_file(tenant_id=tenant_id)
config.set('auth', 'test_accounts_file', TEST_ACCOUNTS_FILE)
config.set('scenario', 'img_dir', '{}'.format(deployment_dir))
config.set('scenario', 'img_file', 'tempest-image')
- config.set('compute', 'image_ref', img_flavor_dict.get("image_id"))
- config.set('compute', 'image_ref_alt',
- img_flavor_dict['image_id_alt'])
- config.set('compute', 'flavor_ref', img_flavor_dict.get("flavor_id"))
- config.set('compute', 'flavor_ref_alt',
- img_flavor_dict['flavor_id_alt'])
+ config.set('compute', 'image_ref', image_id)
+ config.set('compute', 'image_ref_alt', image_id_alt)
+ config.set('compute', 'flavor_ref', flavor_id)
+ config.set('compute', 'flavor_ref_alt', flavor_id_alt)
with open(conf_file, 'wb') as config_file:
config.write(config_file)
shutil.copyfile(conf_file, confpath)
-def generate_test_accounts_file():
+def generate_test_accounts_file(tenant_id):
"""
Add needed tenant and user params into test_accounts.yaml
"""
logger.debug("Add needed params into test_accounts.yaml...")
- tenant_id = create_tenant_user()
accounts_list = [
{
'tenant_name':
def configure_tempest_update_params(tempest_conf_file,
- IMAGE_ID=None, FLAVOR_ID=None):
+ image_id=None, flavor_id=None):
"""
Add/update needed parameters into tempest.conf file
"""
config.set('compute', 'volume_device_name',
CONST.__getattribute__('tempest_volume_device_name'))
if CONST.__getattribute__('tempest_use_custom_images'):
- if IMAGE_ID is not None:
- config.set('compute', 'image_ref', IMAGE_ID)
+ if image_id is not None:
+ config.set('compute', 'image_ref', image_id)
if IMAGE_ID_ALT is not None:
config.set('compute', 'image_ref_alt', IMAGE_ID_ALT)
if CONST.__getattribute__('tempest_use_custom_flavors'):
- if FLAVOR_ID is not None:
- config.set('compute', 'flavor_ref', FLAVOR_ID)
+ if flavor_id is not None:
+ config.set('compute', 'flavor_ref', flavor_id)
if FLAVOR_ID_ALT is not None:
config.set('compute', 'flavor_ref_alt', FLAVOR_ID_ALT)
config.set('identity', 'region', 'RegionOne')
from functest.opnfv_tests.openstack.tempest import conf_utils
from functest.utils.constants import CONST
import functest.utils.functest_utils as ft_utils
+import functest.utils.openstack_utils as os_utils
+
+from snaps.openstack import create_flavor
+from snaps.openstack.create_flavor import FlavorSettings, OpenStackFlavor
+from snaps.openstack.create_project import ProjectSettings
+from snaps.openstack.create_network import NetworkSettings, SubnetSettings
+from snaps.openstack.create_user import UserSettings
+from snaps.openstack.tests import openstack_tests
+from snaps.openstack.utils import deploy_utils
+
""" logging configuration """
logger = logging.getLogger(__name__)
-class TempestCommon(testcase.OSGCTestCase):
+class TempestCommon(testcase.TestCase):
def __init__(self, **kwargs):
super(TempestCommon, self).__init__(**kwargs)
+ self.resources = TempestResourcesManager(**kwargs)
self.MODE = ""
self.OPTION = ""
self.VERIFIER_ID = conf_utils.get_verifier_id()
try:
if not os.path.exists(conf_utils.TEMPEST_RESULTS_DIR):
os.makedirs(conf_utils.TEMPEST_RESULTS_DIR)
- image_and_flavor = conf_utils.create_tempest_resources()
+ resources = self.resources.create()
conf_utils.configure_tempest(
self.DEPLOYMENT_DIR,
- IMAGE_ID=image_and_flavor.get("image_id"),
- FLAVOR_ID=image_and_flavor.get("flavor_id"),
- MODE=self.MODE)
+ image_id=resources.get("image_id"),
+ flavor_id=resources.get("flavor_id"),
+ mode=self.MODE)
self.generate_test_list(self.VERIFIER_REPO_DIR)
self.apply_tempest_blacklist()
self.run_verifier_tests()
except Exception as e:
logger.error('Error with run: %s' % e)
res = testcase.TestCase.EX_RUN_ERROR
+ finally:
+ self.resources.cleanup()
self.stop_time = time.time()
return res
+ def create_snapshot(self):
+ """
+ Run the Tempest cleanup utility to initialize OS state.
+
+ :return: TestCase.EX_OK
+ """
+ logger.info("Initializing the saved state of the OpenStack deployment")
+
+ # Make sure that the verifier is configured
+ conf_utils.configure_verifier(self.DEPLOYMENT_DIR)
+
+ os_utils.init_tempest_cleanup(
+ self.DEPLOYMENT_DIR, 'tempest.conf',
+ os.path.join(conf_utils.TEMPEST_RESULTS_DIR,
+ "tempest-cleanup-init.log")
+ )
+
+ return super(TempestCommon, self).create_snapshot()
+
+ def clean(self):
+ """
+ Run the Tempest cleanup utility to delete and destroy OS resources
+ created by Tempest.
+ """
+ logger.info("Initializing the saved state of the OpenStack deployment")
+
+ os_utils.init_tempest_cleanup(
+ self.DEPLOYMENT_DIR, 'tempest.conf',
+ os.path.join(conf_utils.REFSTACK_RESULTS_DIR,
+ "tempest-cleanup.log")
+ )
+
+ return super(TempestCommon, self).clean()
+
class TempestSmokeSerial(TempestCommon):
TempestCommon.__init__(self, **kwargs)
self.MODE = "defcore"
self.OPTION = "--concurrency 1"
+
+
+class TempestResourcesManager(object):
+
+ def __init__(self, **kwargs):
+ self.os_creds = None
+ if 'os_creds' in kwargs:
+ self.os_creds = kwargs['os_creds']
+ else:
+ self.os_creds = openstack_tests.get_credentials(
+ os_env_file=CONST.__getattribute__('openstack_creds'))
+
+ self.creators = list()
+
+ if hasattr(CONST, 'snaps_images_cirros'):
+ self.cirros_image_config = CONST.__getattribute__(
+ 'snaps_images_cirros')
+ else:
+ self.cirros_image_config = None
+
+ def create(self, use_custom_images=False, use_custom_flavors=False,
+ create_project=False):
+ if create_project:
+ logger.debug("Creating project (tenant) for Tempest suite")
+ project_name = CONST.__getattribute__(
+ 'tempest_identity_tenant_name')
+ project_creator = deploy_utils.create_project(
+ self.os_creds, ProjectSettings(
+ name=project_name,
+ description=CONST.__getattribute__(
+ 'tempest_identity_tenant_description')))
+ if (project_creator is None or
+ project_creator.get_project() is None):
+ raise Exception("Failed to create tenant")
+ project_id = project_creator.get_project().id
+ self.creators.append(project_creator)
+
+ logger.debug("Creating user for Tempest suite")
+ user_creator = deploy_utils.create_user(
+ self.os_creds, UserSettings(
+ name=CONST.__getattribute__('tempest_identity_user_name'),
+ password=CONST.__getattribute__(
+ 'tempest_identity_user_password'),
+ project_name=project_name))
+ if user_creator is None or user_creator.get_user() is None:
+ raise Exception("Failed to create user")
+ user_id = user_creator.get_user().id
+ self.creators.append(user_creator)
+ else:
+ project_name = None
+ project_id = None
+ user_id = None
+
+ logger.debug("Creating private network for Tempest suite")
+ network_creator = deploy_utils.create_network(
+ self.os_creds, NetworkSettings(
+ name=CONST.__getattribute__('tempest_private_net_name'),
+ project_name=project_name,
+ subnet_settings=[SubnetSettings(
+ name=CONST.__getattribute__('tempest_private_subnet_name'),
+ cidr=CONST.__getattribute__('tempest_private_subnet_cidr'))
+ ]))
+ if network_creator is None or network_creator.get_network() is None:
+ raise Exception("Failed to create private network")
+ self.creators.append(network_creator)
+
+ image_id = None
+ image_id_alt = None
+ flavor_id = None
+ flavor_id_alt = None
+
+ if (CONST.__getattribute__('tempest_use_custom_images') or
+ use_custom_images):
+ logger.debug("Creating image for Tempest suite")
+ image_base_name = CONST.__getattribute__('openstack_image_name')
+ os_image_settings = openstack_tests.cirros_image_settings(
+ image_base_name, public=True,
+ image_metadata=self.cirros_image_config)
+ logger.debug("Creating image for Tempest suite")
+ image_creator = deploy_utils.create_image(
+ self.os_creds, os_image_settings)
+ if image_creator is None:
+ raise Exception('Failed to create image')
+ self.creators.append(image_creator)
+ image_id = image_creator.get_image().id
+
+ if use_custom_images:
+ logger.debug("Creating 2nd image for Tempest suite")
+ image_base_name_alt = CONST.__getattribute__(
+ 'openstack_image_name_alt')
+ os_image_settings_alt = openstack_tests.cirros_image_settings(
+ image_base_name_alt, public=True,
+ image_metadata=self.cirros_image_config)
+ logger.debug("Creating 2nd image for Tempest suite")
+ image_creator_alt = deploy_utils.create_image(
+ self.os_creds, os_image_settings_alt)
+ if image_creator_alt is None:
+ raise Exception('Failed to create image')
+ self.creators.append(image_creator_alt)
+ image_id_alt = image_creator_alt.get_image().id
+
+ if (CONST.__getattribute__('tempest_use_custom_flavors') or
+ use_custom_flavors):
+ logger.info("Creating flavor for Tempest suite")
+ scenario = ft_utils.get_scenario()
+ flavor_metadata = None
+ if 'ovs' in scenario or 'fdio' in scenario:
+ flavor_metadata = create_flavor.MEM_PAGE_SIZE_LARGE
+ flavor_creator = OpenStackFlavor(
+ self.os_creds, FlavorSettings(
+ name=CONST.__getattribute__('openstack_flavor_name'),
+ ram=CONST.__getattribute__('openstack_flavor_ram'),
+ disk=CONST.__getattribute__('openstack_flavor_disk'),
+ vcpus=CONST.__getattribute__('openstack_flavor_vcpus'),
+ metadata=flavor_metadata))
+ flavor = flavor_creator.create()
+ if flavor is None:
+ raise Exception('Failed to create flavor')
+ self.creators.append(flavor_creator)
+ flavor_id = flavor.id
+
+ if use_custom_flavors:
+ logger.info("Creating 2nd flavor for Tempest suite")
+ scenario = ft_utils.get_scenario()
+ flavor_metadata_alt = None
+ if 'ovs' in scenario or 'fdio' in scenario:
+ flavor_metadata_alt = create_flavor.MEM_PAGE_SIZE_LARGE
+ flavor_creator_alt = OpenStackFlavor(
+ self.os_creds, FlavorSettings(
+ name=CONST.__getattribute__('openstack_flavor_name_alt'),
+ ram=CONST.__getattribute__('openstack_flavor_ram'),
+ disk=CONST.__getattribute__('openstack_flavor_disk'),
+ vcpus=CONST.__getattribute__('openstack_flavor_vcpus'),
+ metadata=flavor_metadata_alt))
+ flavor_alt = flavor_creator_alt.create()
+ if flavor_alt is None:
+ raise Exception('Failed to create flavor')
+ self.creators.append(flavor_creator_alt)
+ flavor_id_alt = flavor_alt.id
+
+ print("RESOURCES CREATE: image_id: %s, image_id_alt: %s, "
+ "flavor_id: %s, flavor_id_alt: %s" % (
+ image_id, image_id_alt, flavor_id, flavor_id_alt,))
+
+ result = {
+ 'image_id': image_id,
+ 'image_id_alt': image_id_alt,
+ 'flavor_id': flavor_id,
+ 'flavor_id_alt': flavor_id_alt
+ }
+
+ if create_project:
+ result['project_id'] = project_id
+ result['tenant_id'] = project_id # for compatibility
+ result['user_id'] = user_id
+
+ return result
+
+ def cleanup(self):
+ """
+ Cleanup all OpenStack objects. Should be called on completion.
+ """
+ for creator in reversed(self.creators):
+ try:
+ creator.clean()
+ except Exception as e:
+ logger.error('Unexpected error cleaning - %s', e)
import unittest
from functest.core import testcase
-from functest.opnfv_tests.openstack.refstack_client import refstack_client
+from functest.opnfv_tests.openstack.refstack_client.refstack_client import \
+ RefstackClient, RefstackClientParser
from functest.utils.constants import CONST
+from snaps.openstack.os_credentials import OSCreds
+
class OSRefstackClientTesting(unittest.TestCase):
'functest', 'opnfv_tests/openstack/refstack_client/defcore.txt')
def setUp(self):
- self.defaultargs = {'config': self._config,
- 'testlist': self._testlist}
+ self.default_args = {'config': self._config,
+ 'testlist': self._testlist}
CONST.__setattr__('OS_AUTH_URL', 'https://ip:5000/v3')
CONST.__setattr__('OS_INSECURE', 'true')
- self.refstackclient = refstack_client.RefstackClient()
+ self.os_creds = OSCreds(
+ username='user', password='pass',
+ auth_url='http://foo.com:5000/v3', project_name='bar')
+
+ def _create_client(self):
+ with mock.patch('snaps.openstack.tests.openstack_tests.'
+ 'get_credentials', return_value=self.os_creds):
+ return RefstackClient()
def test_run_defcore_insecure(self):
insecure = '-k'
config = 'tempest.conf'
testlist = 'testlist'
+ client = self._create_client()
with mock.patch('functest.opnfv_tests.openstack.refstack_client.'
'refstack_client.ft_utils.execute_command') as m:
cmd = ("refstack-client test {0} -c {1} -v --test-list {2}"
.format(insecure, config, testlist))
- self.refstackclient.run_defcore(config, testlist)
+ client.run_defcore(config, testlist)
m.assert_any_call(cmd)
def test_run_defcore(self):
CONST.__setattr__('OS_AUTH_URL', 'http://ip:5000/v3')
- refstackclient = refstack_client.RefstackClient()
insecure = ''
config = 'tempest.conf'
testlist = 'testlist'
+ client = self._create_client()
with mock.patch('functest.opnfv_tests.openstack.refstack_client.'
'refstack_client.ft_utils.execute_command') as m:
cmd = ("refstack-client test {0} -c {1} -v --test-list {2}"
.format(insecure, config, testlist))
- refstackclient.run_defcore(config, testlist)
+ client.run_defcore(config, testlist)
m.assert_any_call(cmd)
@mock.patch('functest.opnfv_tests.openstack.refstack_client.'
mock_logger_info):
self.case_name = 'refstack_defcore'
self.result = 0
- self.refstackclient.parse_refstack_result()
+ self._create_client().parse_refstack_result()
mock_logger_info.assert_called_once_with(
"Testcase %s success_rate is %s%%",
self.case_name, self.result)
"success": ['tempest.api.compute [18.464988s]'],
"errors": ['tempest.api.volume [0.230334s]'],
"skipped": ['tempest.api.network [1.265828s]']}
+ client = self._create_client()
with mock.patch('__builtin__.open',
mock.mock_open(read_data=log_file)):
- self.refstackclient.parse_refstack_result()
- self.assertEqual(self.refstackclient.details, self.details)
+ client.parse_refstack_result()
+ self.assertEqual(client.details, self.details)
def _get_main_kwargs(self, key=None):
kwargs = {'config': self._config,
def _test_main(self, status, *args):
kwargs = self._get_main_kwargs()
- self.assertEqual(self.refstackclient.main(**kwargs), status)
+ client = self._create_client()
+ self.assertEqual(client.main(**kwargs), status)
if len(args) > 0:
args[0].assert_called_once_with(
- refstack_client.RefstackClient.result_dir)
+ RefstackClient.result_dir)
if len(args) > 1:
args
def _test_main_missing_keyword(self, key):
kwargs = self._get_main_kwargs(key)
- self.assertEqual(self.refstackclient.main(**kwargs),
+ client = self._create_client()
+ self.assertEqual(client.main(**kwargs),
testcase.TestCase.EX_RUN_ERROR)
def test_main_missing_conf(self):
self._test_main_missing_keyword('testlist')
def _test_argparser(self, arg, value):
- self.defaultargs[arg] = value
- parser = refstack_client.RefstackClientParser()
+ self.default_args[arg] = value
+ parser = RefstackClientParser()
self.assertEqual(parser.parse_args(["--{}={}".format(arg, value)]),
- self.defaultargs)
+ self.default_args)
def test_argparser_conf(self):
self._test_argparser('config', self._config)
self._test_argparser('testlist', self._testlist)
def test_argparser_multiple_args(self):
- self.defaultargs['config'] = self._config
- self.defaultargs['testlist'] = self._testlist
- parser = refstack_client.RefstackClientParser()
+ self.default_args['config'] = self._config
+ self.default_args['testlist'] = self._testlist
+ parser = RefstackClientParser()
self.assertEqual(parser.parse_args(
["--config={}".format(self._config),
"--testlist={}".format(self._testlist)
- ]), self.defaultargs)
+ ]), self.default_args)
if __name__ == "__main__":
import mock
-from functest.opnfv_tests.openstack.tempest import conf_utils
+from functest.opnfv_tests.openstack.tempest import tempest, conf_utils
from functest.utils.constants import CONST
+from snaps.openstack.os_credentials import OSCreds
class OSTempestConfUtilsTesting(unittest.TestCase):
- def test_create_tempest_resources_missing_network_dic(self):
- with mock.patch('functest.opnfv_tests.openstack.tempest.conf_utils.'
- 'os_utils.create_shared_network_full',
- return_value=None), \
- self.assertRaises(Exception) as context:
- conf_utils.create_tempest_resources()
- msg = 'Failed to create private network'
- self.assertTrue(msg in context)
-
- def test_create_tempest_resources_missing_image(self):
- with mock.patch('functest.opnfv_tests.openstack.tempest.conf_utils.'
- 'os_utils.create_shared_network_full',
- return_value=mock.Mock()), \
- mock.patch('functest.opnfv_tests.openstack.tempest.conf_utils.'
- 'os_utils.get_or_create_image',
- return_value=(mock.Mock(), None)), \
- self.assertRaises(Exception) as context:
-
- CONST.__setattr__('tempest_use_custom_images', True)
- conf_utils.create_tempest_resources()
- msg = 'Failed to create image'
- self.assertTrue(msg in context)
-
- CONST.__setattr__('tempest_use_custom_images', False)
- conf_utils.create_tempest_resources(use_custom_images=True)
- msg = 'Failed to create image'
- self.assertTrue(msg in context)
-
- def test_create_tempest_resources_missing_flavor(self):
- with mock.patch('functest.opnfv_tests.openstack.tempest.conf_utils.'
- 'os_utils.create_shared_network_full',
- return_value=mock.Mock()), \
- mock.patch('functest.opnfv_tests.openstack.tempest.conf_utils.'
- 'os_utils.get_or_create_image',
- return_value=(mock.Mock(), 'image_id')), \
- mock.patch('functest.opnfv_tests.openstack.tempest.conf_utils.'
- 'os_utils.get_or_create_flavor',
- return_value=(mock.Mock(), None)), \
- self.assertRaises(Exception) as context:
- CONST.__setattr__('tempest_use_custom_images', True)
- CONST.__setattr__('tempest_use_custom_flavors', True)
- conf_utils.create_tempest_resources()
- msg = 'Failed to create flavor'
- self.assertTrue(msg in context)
-
- CONST.__setattr__('tempest_use_custom_images', True)
- CONST.__setattr__('tempest_use_custom_flavors', False)
- conf_utils.create_tempest_resources(use_custom_flavors=False)
- msg = 'Failed to create flavor'
- self.assertTrue(msg in context)
-
- @mock.patch('functest.opnfv_tests.openstack.tempest.conf_utils.'
- 'logger.error')
- def create_tenant_user_and_tenant_ok(self, mock_logger_error):
- with mock.patch('functest.opnfv_tests.openstack.tempest.conf_utils.'
- 'os_utils.get_keystone_client',
- return_value=mock.Mock()), \
- mock.patch('functest.opnfv_tests.openstack.tempest.conf_utils.'
- 'os_utils.create_tenant',
- return_value='test_tenant_id'):
- conf_utils.create_tenant_user()
- mock_logger_error.assert_not_called()
-
- @mock.patch('functest.opnfv_tests.openstack.tempest.conf_utils.'
- 'logger.error')
- def create_tenant_user_and_user_ok(self, mock_logger_error):
- with mock.patch('functest.opnfv_tests.openstack.tempest.conf_utils.'
- 'os_utils.get_keystone_client',
- return_value=mock.Mock()), \
- mock.patch('functest.opnfv_tests.openstack.tempest.conf_utils.'
- 'os_utils.create_user',
- return_value='test_user_id'):
- conf_utils.create_tenant_user()
- mock_logger_error.assert_not_called()
-
- @mock.patch('functest.opnfv_tests.openstack.tempest.conf_utils.'
- 'logger.error')
- def create_tenant_user_and_tenant_failed(self, mock_logger_error):
- with mock.patch('functest.opnfv_tests.openstack.tempest.conf_utils.'
- 'os_utils.get_keystone_client',
- return_value=mock.Mock()), \
- mock.patch('functest.opnfv_tests.openstack.tempest.conf_utils.'
- 'os_utils.create_tenant',
- return_value=None):
- conf_utils.create_tenant_user()
- msg = ("Failed to create %s tenant"
- % CONST.__getattribute__('tempest_identity_tenant_name'))
- mock_logger_error.assert_any_call(msg)
-
- @mock.patch('functest.opnfv_tests.openstack.tempest.conf_utils.'
- 'logger.error')
- def create_tenant_user_and_user_failed(self, mock_logger_error):
- with mock.patch('functest.opnfv_tests.openstack.tempest.conf_utils.'
- 'os_utils.get_keystone_client',
- return_value=mock.Mock()), \
- mock.patch('functest.opnfv_tests.openstack.tempest.conf_utils.'
- 'os_utils.create_user',
- return_value=None):
- conf_utils.create_tenant_user()
- msg = ("Failed to create %s user"
- % CONST.__getattribute__('tempest_identity_user_name'))
- mock_logger_error.assert_any_call(msg)
+ def setUp(self):
+ self.os_creds = OSCreds(
+ username='user', password='pass',
+ auth_url='http://foo.com:5000/v3', project_name='bar')
+
+ @mock.patch('snaps.openstack.utils.deploy_utils.create_project',
+ return_value=mock.Mock())
+ @mock.patch('snaps.openstack.utils.deploy_utils.create_user',
+ return_value=mock.Mock())
+ @mock.patch('snaps.openstack.utils.deploy_utils.create_network',
+ return_value=None)
+ @mock.patch('snaps.openstack.utils.deploy_utils.create_image',
+ return_value=mock.Mock())
+ def test_create_tempest_resources_missing_network_dic(self, *mock_args):
+ tempest_resources = tempest.TempestResourcesManager(os_creds={})
+ with self.assertRaises(Exception) as context:
+ tempest_resources.create()
+ msg = 'Failed to create private network'
+ self.assertTrue(msg in context.exception)
+
+ @mock.patch('snaps.openstack.utils.deploy_utils.create_project',
+ return_value=mock.Mock())
+ @mock.patch('snaps.openstack.utils.deploy_utils.create_user',
+ return_value=mock.Mock())
+ @mock.patch('snaps.openstack.utils.deploy_utils.create_network',
+ return_value=mock.Mock())
+ @mock.patch('snaps.openstack.utils.deploy_utils.create_image',
+ return_value=None)
+ def test_create_tempest_resources_missing_image(self, *mock_args):
+ tempest_resources = tempest.TempestResourcesManager(os_creds={})
+
+ CONST.__setattr__('tempest_use_custom_imagess', True)
+ with self.assertRaises(Exception) as context:
+ tempest_resources.create()
+ msg = 'Failed to create image'
+ self.assertTrue(msg in context.exception, msg=str(context.exception))
+
+ CONST.__setattr__('tempest_use_custom_imagess', False)
+ with self.assertRaises(Exception) as context:
+ tempest_resources.create(use_custom_images=True)
+ msg = 'Failed to create image'
+ self.assertTrue(msg in context.exception, msg=str(context.exception))
+
+ @mock.patch('snaps.openstack.utils.deploy_utils.create_project',
+ return_value=mock.Mock())
+ @mock.patch('snaps.openstack.utils.deploy_utils.create_user',
+ return_value=mock.Mock())
+ @mock.patch('snaps.openstack.utils.deploy_utils.create_network',
+ return_value=mock.Mock())
+ @mock.patch('snaps.openstack.utils.deploy_utils.create_image',
+ return_value=mock.Mock())
+ @mock.patch('snaps.openstack.create_flavor.OpenStackFlavor.create',
+ return_value=None)
+ def test_create_tempest_resources_missing_flavor(self, *mock_args):
+ tempest_resources = tempest.TempestResourcesManager(
+ os_creds=self.os_creds)
+
+ CONST.__setattr__('tempest_use_custom_images', True)
+ CONST.__setattr__('tempest_use_custom_flavors', True)
+ with self.assertRaises(Exception) as context:
+ tempest_resources.create()
+ msg = 'Failed to create flavor'
+ self.assertTrue(msg in context.exception, msg=str(context.exception))
+
+ CONST.__setattr__('tempest_use_custom_images', True)
+ CONST.__setattr__('tempest_use_custom_flavors', False)
+ with self.assertRaises(Exception) as context:
+ tempest_resources.create(use_custom_flavors=True)
+ msg = 'Failed to create flavor'
+ self.assertTrue(msg in context.exception, msg=str(context.exception))
def test_get_verifier_id_missing_verifier(self):
CONST.__setattr__('tempest_deployment_name', 'test_deploy_name')
self.assertTrue(m1.called)
def test_configure_tempest_defcore_default(self):
- img_flavor_dict = {'image_id': 'test_image_id',
- 'flavor_id': 'test_flavor_id',
- 'image_id_alt': 'test_image_alt_id',
- 'flavor_id_alt': 'test_flavor_alt_id'}
with mock.patch('functest.opnfv_tests.openstack.tempest.'
'conf_utils.configure_verifier',
return_value='test_conf_file'), \
'conf_utils.generate_test_accounts_file'), \
mock.patch('functest.opnfv_tests.openstack.tempest.'
'conf_utils.shutil.copyfile'):
- conf_utils.configure_tempest_defcore('test_dep_dir',
- img_flavor_dict)
+ conf_utils.configure_tempest_defcore(
+ 'test_dep_dir', 'test_image_id', 'test_flavor_id',
+ 'test_image_alt_id', 'test_flavor_alt_id', 'test_tenant_id')
mset.assert_any_call('compute', 'image_ref', 'test_image_id')
mset.assert_any_call('compute', 'image_ref_alt',
'test_image_alt_id')
self.assertTrue(mwrite.called)
def test_generate_test_accounts_file_default(self):
- with mock.patch('functest.opnfv_tests.openstack.tempest.conf_utils.'
- 'create_tenant_user',
- return_value='test_tenant_id') as mock_create, \
- mock.patch("__builtin__.open", mock.mock_open()), \
+ with mock.patch("__builtin__.open", mock.mock_open()), \
mock.patch('functest.opnfv_tests.openstack.tempest.conf_utils.'
'yaml.dump') as mock_dump:
- conf_utils.generate_test_accounts_file()
- self.assertTrue(mock_create.called)
+ conf_utils.generate_test_accounts_file('test_tenant_id')
self.assertTrue(mock_dump.called)
def _test_missing_param(self, params, image_id, flavor_id):
CONST.__setattr__('OS_ENDPOINT_TYPE', None)
conf_utils.\
configure_tempest_update_params('test_conf_file',
- IMAGE_ID=image_id,
- FLAVOR_ID=flavor_id)
+ image_id=image_id,
+ flavor_id=flavor_id)
mset.assert_any_call(params[0], params[1], params[2])
self.assertTrue(mread.called)
self.assertTrue(mwrite.called)
from functest.opnfv_tests.openstack.tempest import conf_utils
from functest.utils.constants import CONST
+from snaps.openstack.os_credentials import OSCreds
+
class OSTempestTesting(unittest.TestCase):
def setUp(self):
+ os_creds = OSCreds(
+ username='user', password='pass',
+ auth_url='http://foo.com:5000/v3', project_name='bar')
+
with mock.patch('functest.opnfv_tests.openstack.tempest.tempest.'
'conf_utils.get_verifier_id',
return_value='test_deploy_id'), \
return_value='test_verifier_repo_dir'), \
mock.patch('functest.opnfv_tests.openstack.tempest.tempest.'
'conf_utils.get_verifier_deployment_dir',
- return_value='test_verifier_deploy_dir'):
+ return_value='test_verifier_deploy_dir'), \
+ mock.patch('snaps.openstack.tests.openstack_tests.get_credentials',
+ return_value=os_creds):
self.tempestcommon = tempest.TempestCommon()
self.tempestsmoke_serial = tempest.TempestSmokeSerial()
self.tempestsmoke_parallel = tempest.TempestSmokeParallel()
'os.path.exists', return_value=False)
@mock.patch('functest.opnfv_tests.openstack.tempest.tempest.os.makedirs')
@mock.patch('functest.opnfv_tests.openstack.tempest.tempest.'
- 'conf_utils.create_tempest_resources', side_effect=Exception)
- def test_run_create_tempest_resources_ko(self, *args):
+ 'TempestResourcesManager.create', side_effect=Exception)
+ def test_run_tempest_create_resources_ko(self, *args):
self.assertEqual(self.tempestcommon.run(),
testcase.TestCase.EX_RUN_ERROR)
'os.path.exists', return_value=False)
@mock.patch('functest.opnfv_tests.openstack.tempest.tempest.os.makedirs')
@mock.patch('functest.opnfv_tests.openstack.tempest.tempest.'
- 'conf_utils.create_tempest_resources', return_value={})
+ 'TempestResourcesManager.create', return_value={})
@mock.patch('functest.opnfv_tests.openstack.tempest.tempest.'
'conf_utils.configure_tempest', side_effect=Exception)
def test_run_configure_tempest_ko(self, *args):
'os.path.exists', return_value=False)
@mock.patch('functest.opnfv_tests.openstack.tempest.tempest.os.makedirs')
@mock.patch('functest.opnfv_tests.openstack.tempest.tempest.'
- 'conf_utils.create_tempest_resources', return_value={})
+ 'TempestResourcesManager.create', return_value={})
@mock.patch('functest.opnfv_tests.openstack.tempest.tempest.'
'conf_utils.configure_tempest')
def _test_run(self, status, *args):
def execute_command_raise(cmd, info=False, error_msg="",
- verbose=True, output_file=None):
- ret = execute_command(cmd, info, error_msg, verbose, output_file)
+ verbose=True, output_file=None, env=None):
+ ret = execute_command(cmd, info, error_msg, verbose, output_file, env)
if ret != 0:
raise Exception(error_msg)
def execute_command(cmd, info=False, error_msg="",
- verbose=True, output_file=None):
+ verbose=True, output_file=None, env=None):
if not error_msg:
error_msg = ("The command '%s' failed." % cmd)
msg_exec = ("Executing command: '%s'" % cmd)
logger.info(msg_exec)
else:
logger.debug(msg_exec)
- p = subprocess.Popen(cmd, shell=True, stdout=subprocess.PIPE,
+ p = subprocess.Popen(cmd, env=env, shell=True, stdout=subprocess.PIPE,
stderr=subprocess.STDOUT)
if output_file:
f = open(output_file, "w")
except Exception as e:
logger.error("Error [get_resource]: %s" % e)
return None
+
+
+# *********************************************
+# TEMPEST
+# *********************************************
+def init_tempest_cleanup(tempest_config_dir=None,
+ tempest_config_filename='tempest.conf',
+ output_file=None):
+ """
+ Initialize the Tempest Cleanup utility.
+ See https://docs.openstack.org/tempest/latest/cleanup.html for docs.
+
+ :param tempest_config_dir: The directory where the Tempest config file is
+ located. If not specified, we let Tempest pick both the directory
+ and the filename (i.e. second parameter is ignored)
+ :param tempest_config_filename: The filename of the Tempest config file
+ :param output_file: Optional file where to save output
+ """
+ # The Tempest cleanup utility currently offers no cmd argument to specify
+ # the config file, therefore it has to be configured with env variables
+ env = None
+ if tempest_config_dir:
+ env = os.environ.copy()
+ env['TEMPEST_CONFIG_DIR'] = tempest_config_dir
+ env['TEMPEST_CONFIG'] = tempest_config_filename
+
+ # If this command fails, an exception must be raised to stop the script
+ # otherwise the later cleanup would destroy also other resources
+ cmd_line = "tempest cleanup --init-saved-state"
+ ft_utils.execute_command_raise(cmd_line, env=env, output_file=output_file,
+ error_msg="Tempest cleanup init failed")
+
+
+def perform_tempest_cleanup(tempest_config_dir=None,
+ tempest_config_filename='tempest.conf',
+ output_file=None):
+ """
+ Perform cleanup using the Tempest Cleanup utility.
+ See https://docs.openstack.org/tempest/latest/cleanup.html for docs.
+
+ :param tempest_config_dir: The directory where the Tempest config file is
+ located. If not specified, we let Tempest pick both the directory
+ and the filename (i.e. second parameter is ignored)
+ :param tempest_config_filename: The filename of the Tempest config file
+ :param output_file: Optional file where to save output
+ """
+ # The Tempest cleanup utility currently offers no cmd argument to specify
+ # the config file, therefore it has to be configured with env variables
+ env = None
+ if tempest_config_dir:
+ env = os.environ.copy()
+ env['TEMPEST_CONFIG_DIR'] = tempest_config_dir
+ env['TEMPEST_CONFIG'] = tempest_config_filename
+
+ # If this command fails, an exception must be raised to stop the script
+ # otherwise the later cleanup would destroy also other resources
+ cmd_line = "tempest cleanup"
+ ft_utils.execute_command(cmd_line, env=env, output_file=output_file,
+ error_msg="Tempest cleanup failed")
Code Review / functest-xtesting.git / commitdiff