Merge "Use tempest.test.BaseTestCase for murano tempest tests"

This commit is contained in:
Jenkins 2017-06-26 13:05:06 +00:00 committed by Gerrit Code Review
commit 6e3ea6cdfc
4 changed files with 144 additions and 196 deletions

View File

@ -15,7 +15,7 @@
from tempest.common import credentials_factory as common_creds from tempest.common import credentials_factory as common_creds
from tempest import config from tempest import config
from tempest.lib import base from tempest import test
from murano_tempest_tests import clients from murano_tempest_tests import clients
from murano_tempest_tests import utils from murano_tempest_tests import utils
@ -23,22 +23,28 @@ from murano_tempest_tests import utils
CONF = config.CONF CONF = config.CONF
class BaseArtifactsTest(base.BaseTestCase): class BaseArtifactsTest(test.BaseTestCase):
"""Base test class for Murano Glare tests.""" """Base test class for Murano Glare tests."""
@classmethod @classmethod
def setUpClass(cls): def skip_checks(cls):
super(BaseArtifactsTest, cls).setUpClass() super(BaseArtifactsTest, cls).skip_checks()
cls.resource_setup() if not CONF.service_available.murano:
skip_msg = "Murano is disabled"
raise cls.skipException(skip_msg)
@classmethod @classmethod
def tearDownClass(cls): def setup_clients(cls):
cls.resource_cleanup() super(BaseArtifactsTest, cls).setup_clients()
super(BaseArtifactsTest, cls).tearDownClass() if not hasattr(cls, "os_primary"):
creds = cls.get_configured_isolated_creds(type_of_creds='primary')
cls.os_primary = clients.Manager(credentials=creds)
cls.artifacts_client = cls.os_primary.artifacts_client
cls.application_catalog_client = \
cls.os_primary.application_catalog_client
@classmethod @classmethod
def get_client_with_isolated_creds(cls, type_of_creds="admin"): def get_client_with_isolated_creds(cls, type_of_creds="admin"):
creds = cls.get_configured_isolated_creds(type_of_creds=type_of_creds) creds = cls.get_configured_isolated_creds(type_of_creds=type_of_creds)
os = clients.Manager(credentials=creds) os = clients.Manager(credentials=creds)
@ -69,33 +75,6 @@ class BaseArtifactsTest(base.BaseTestCase):
return creds.credentials return creds.credentials
@classmethod
def verify_nonempty(cls, *args):
if not all(args):
msg = "Missing API credentials in configuration."
raise cls.skipException(msg)
@classmethod
def resource_setup(cls):
if not CONF.service_available.murano:
skip_msg = "Murano is disabled"
raise cls.skipException(skip_msg)
if not hasattr(cls, "os_primary"):
creds = cls.get_configured_isolated_creds(type_of_creds='primary')
cls.os_primary = clients.Manager(credentials=creds)
cls.artifacts_client = cls.os_primary.artifacts_client
cls.application_catalog_client = \
cls.os_primary.application_catalog_client
@classmethod
def resource_cleanup(cls):
cls.clear_isolated_creds()
@classmethod
def clear_isolated_creds(cls):
if hasattr(cls, "dynamic_cred"):
cls.credentials.clear_creds()
@classmethod @classmethod
def upload_package(cls, application_name, version=None, require=None): def upload_package(cls, application_name, version=None, require=None):
abs_archive_path, dir_with_archive, archive_name = \ abs_archive_path, dir_with_archive, archive_name = \

View File

@ -15,7 +15,7 @@
from tempest.common import credentials_factory as common_creds from tempest.common import credentials_factory as common_creds
from tempest import config from tempest import config
from tempest.lib import base from tempest import test
from murano_tempest_tests import clients from murano_tempest_tests import clients
from murano_tempest_tests import utils from murano_tempest_tests import utils
@ -23,22 +23,28 @@ from murano_tempest_tests import utils
CONF = config.CONF CONF = config.CONF
class BaseApplicationCatalogTest(base.BaseTestCase): class BaseApplicationCatalogTest(test.BaseTestCase):
"""Base test class for Murano Service Broker API tests.""" """Base test class for Murano Service Broker API tests."""
@classmethod @classmethod
def setUpClass(cls): def skip_checks(cls):
super(BaseApplicationCatalogTest, cls).setUpClass() super(BaseApplicationCatalogTest, cls).skip_checks()
cls.resource_setup() if not CONF.service_available.murano:
skip_msg = "Murano is disabled"
raise cls.skipException(skip_msg)
@classmethod @classmethod
def tearDownClass(cls): def setup_clients(cls):
cls.resource_cleanup() super(BaseApplicationCatalogTest, cls).setup_clients()
super(BaseApplicationCatalogTest, cls).tearDownClass() if not hasattr(cls, "os_primary"):
creds = cls.get_configured_isolated_creds(type_of_creds='primary')
cls.os_primary = clients.Manager(credentials=creds)
cls.application_catalog_client = \
cls.os_primary.application_catalog_client
cls.artifacts_client = cls.os_primary.artifacts_client
@classmethod @classmethod
def get_client_with_isolated_creds(cls, type_of_creds="admin"): def get_client_with_isolated_creds(cls, type_of_creds="admin"):
creds = cls.get_configured_isolated_creds(type_of_creds=type_of_creds) creds = cls.get_configured_isolated_creds(type_of_creds=type_of_creds)
os = clients.Manager(credentials=creds) os = clients.Manager(credentials=creds)
@ -53,11 +59,10 @@ class BaseApplicationCatalogTest(base.BaseTestCase):
cls.admin_role = CONF.identity.admin_role cls.admin_role = CONF.identity.admin_role
else: else:
cls.admin_role = 'admin' cls.admin_role = 'admin'
if not hasattr(cls, 'dynamic_cred'): cls.credentials = common_creds.get_credentials_provider(
cls.credentials = common_creds.get_credentials_provider( name=cls.__name__,
name=cls.__name__, force_tenant_isolation=CONF.auth.use_dynamic_credentials,
force_tenant_isolation=CONF.auth.use_dynamic_credentials, identity_version=CONF.identity.auth_version)
identity_version=CONF.identity.auth_version)
if type_of_creds == 'primary': if type_of_creds == 'primary':
creds = cls.credentials.get_primary_creds() creds = cls.credentials.get_primary_creds()
elif type_of_creds == 'admin': elif type_of_creds == 'admin':
@ -70,33 +75,6 @@ class BaseApplicationCatalogTest(base.BaseTestCase):
return creds.credentials return creds.credentials
@classmethod
def verify_nonempty(cls, *args):
if not all(args):
msg = "Missing API credentials in configuration."
raise cls.skipException(msg)
@classmethod
def resource_setup(cls):
if not CONF.service_available.murano:
skip_msg = "Murano is disabled"
raise cls.skipException(skip_msg)
if not hasattr(cls, "os_primary"):
creds = cls.get_configured_isolated_creds(type_of_creds='primary')
cls.os_primary = clients.Manager(credentials=creds)
cls.application_catalog_client = \
cls.os_primary.application_catalog_client
cls.artifacts_client = cls.os_primary.artifacts_client
@classmethod
def resource_cleanup(cls):
cls.clear_isolated_creds()
@classmethod
def clear_isolated_creds(cls):
if hasattr(cls, "dynamic_cred"):
cls.credentials.clear_creds()
@staticmethod @staticmethod
def _get_demo_app(): def _get_demo_app():
return { return {
@ -119,18 +97,14 @@ class BaseApplicationCatalogTest(base.BaseTestCase):
} }
class BaseApplicationCatalogAdminTest(BaseApplicationCatalogTest):
@classmethod
def resource_setup(cls):
cls.os_primary = clients.Manager()
super(BaseApplicationCatalogAdminTest, cls).resource_setup()
class BaseApplicationCatalogIsolatedAdminTest(BaseApplicationCatalogTest): class BaseApplicationCatalogIsolatedAdminTest(BaseApplicationCatalogTest):
@classmethod @classmethod
def resource_setup(cls): def setup_clients(cls):
creds = cls.get_configured_isolated_creds(type_of_creds='admin') super(BaseApplicationCatalogIsolatedAdminTest, cls).setup_clients()
cls.os_primary = clients.Manager(credentials=creds) if not hasattr(cls, "os_admin"):
super(BaseApplicationCatalogIsolatedAdminTest, cls).resource_setup() creds = cls.get_configured_isolated_creds(type_of_creds='admin')
cls.os_admin = clients.Manager(credentials=creds)
cls.application_catalog_client = \
cls.os_admin.application_catalog_client
cls.artifacts_client = cls.os_admin.artifacts_client

View File

@ -18,55 +18,20 @@ import time
from tempest.common import credentials_factory as common_creds from tempest.common import credentials_factory as common_creds
from tempest import config from tempest import config
from tempest.lib import base
from tempest.lib import exceptions from tempest.lib import exceptions
from tempest import test
from murano_tempest_tests import clients from murano_tempest_tests import clients
CONF = config.CONF CONF = config.CONF
class BaseServiceBrokerTest(base.BaseTestCase): class BaseServiceBrokerTest(test.BaseTestCase):
"""Base test class for Murano Service Broker API tests.""" """Base test class for Murano Service Broker API tests."""
@classmethod @classmethod
def setUpClass(cls): def skip_checks(cls):
super(BaseServiceBrokerTest, cls).setUpClass() super(BaseServiceBrokerTest, cls).skip_checks()
cls.resource_setup()
@classmethod
def tearDownClass(cls):
cls.resource_cleanup()
super(BaseServiceBrokerTest, cls).tearDownClass()
@classmethod
def get_client_with_isolated_creds(cls, name=None,
type_of_creds="admin"):
cls.credentials = common_creds.get_credentials_provider(
name=cls.__name__,
force_tenant_isolation=CONF.auth.use_dynamic_credentials,
identity_version=CONF.identity.auth_version)
if "admin" in type_of_creds:
creds = cls.credentials.get_admin_creds()
elif "alt" in type_of_creds:
creds = cls.credentials.get_alt_creds()
else:
creds = cls.credentials.get_credentials(type_of_creds)
cls.credentials.type_of_creds = type_of_creds
os = clients.Manager(credentials=creds)
client = os.service_broker_client
return client
@classmethod
def verify_nonempty(cls, *args):
if not all(args):
msg = "Missing API credentials in configuration."
raise cls.skipException(msg)
@classmethod
def resource_setup(cls):
if not CONF.service_broker.run_service_broker_tests: if not CONF.service_broker.run_service_broker_tests:
skip_msg = "Service Broker API tests are disabled" skip_msg = "Service Broker API tests are disabled"
raise cls.skipException(skip_msg) raise cls.skipException(skip_msg)
@ -76,6 +41,10 @@ class BaseServiceBrokerTest(base.BaseTestCase):
if not CONF.service_available.murano: if not CONF.service_available.murano:
skip_msg = "Murano is disabled" skip_msg = "Murano is disabled"
raise cls.skipException(skip_msg) raise cls.skipException(skip_msg)
@classmethod
def setup_clients(cls):
super(BaseServiceBrokerTest, cls).setup_clients()
if not hasattr(cls, "os_primary"): if not hasattr(cls, "os_primary"):
cls.username = CONF.identity.username cls.username = CONF.identity.username
cls.password = CONF.identity.password cls.password = CONF.identity.password
@ -86,18 +55,43 @@ class BaseServiceBrokerTest(base.BaseTestCase):
cls.application_catalog_client = \ cls.application_catalog_client = \
cls.os_primary.application_catalog_client cls.os_primary.application_catalog_client
def setUp(self): @classmethod
super(BaseServiceBrokerTest, self).setUp() def get_client_with_isolated_creds(cls, type_of_creds="admin"):
self.addCleanup(self.clear_isolated_creds) creds = cls.get_configured_isolated_creds(type_of_creds=type_of_creds)
os = clients.Manager(credentials=creds)
client = os.application_catalog_client
return client
@classmethod @classmethod
def resource_cleanup(cls): def get_configured_isolated_creds(cls, type_of_creds='admin'):
cls.clear_isolated_creds() identity_version = CONF.identity.auth_version
if identity_version == 'v3':
cls.admin_role = CONF.identity.admin_role
else:
cls.admin_role = 'admin'
cls.credentials = common_creds.get_credentials_provider(
name=cls.__name__,
force_tenant_isolation=CONF.auth.use_dynamic_credentials,
identity_version=CONF.identity.auth_version)
if type_of_creds == 'primary':
creds = cls.credentials.get_primary_creds()
elif type_of_creds == 'admin':
creds = cls.credentials.get_admin_creds()
elif type_of_creds == 'alt':
creds = cls.credentials.get_alt_creds()
else:
creds = cls.credentials.get_credentials(type_of_creds)
cls.credentials.type_of_creds = type_of_creds
return creds.credentials
@classmethod @classmethod
def clear_isolated_creds(cls): def verify_nonempty(cls, *args):
if hasattr(cls, "dynamic_cred"): if not all(args):
cls.credentials.clear_creds() msg = "Missing API credentials in configuration."
raise cls.skipException(msg)
def wait_for_result(self, instance_id, timeout): def wait_for_result(self, instance_id, timeout):
start_time = time.time() start_time = time.time()
@ -129,6 +123,14 @@ class BaseServiceBrokerTest(base.BaseTestCase):
class BaseServiceBrokerAdminTest(BaseServiceBrokerTest): class BaseServiceBrokerAdminTest(BaseServiceBrokerTest):
@classmethod @classmethod
def resource_setup(cls): def setup_clients(cls):
cls.os_primary = clients.Manager() super(BaseServiceBrokerTest, cls).setup_clients()
super(BaseServiceBrokerAdminTest, cls).resource_setup() if not hasattr(cls, "os_admin"):
cls.username = CONF.identity.username
cls.password = CONF.identity.password
cls.tenant_name = CONF.identity.tenant_name
cls.verify_nonempty(cls.username, cls.password, cls.tenant_name)
cls.os_admin = clients.Manager()
cls.service_broker_client = cls.os_admin.service_broker_client
cls.application_catalog_client = \
cls.os_admin.application_catalog_client

View File

@ -21,8 +21,8 @@ from tempest.clients import Manager as services_manager
from tempest.common import credentials_factory as common_creds from tempest.common import credentials_factory as common_creds
from tempest.common import waiters from tempest.common import waiters
from tempest import config from tempest import config
from tempest.lib import base
from tempest.lib import exceptions from tempest.lib import exceptions
from tempest import test
from murano_tempest_tests import clients from murano_tempest_tests import clients
from murano_tempest_tests import utils from murano_tempest_tests import utils
@ -30,22 +30,42 @@ from murano_tempest_tests import utils
CONF = config.CONF CONF = config.CONF
class BaseApplicationCatalogScenarioTest(base.BaseTestCase): class BaseApplicationCatalogScenarioTest(test.BaseTestCase):
"""Base test class for Murano Application Catalog Scenario tests.""" """Base test class for Murano Application Catalog Scenario tests."""
@classmethod @classmethod
def setUpClass(cls): def skip_checks(cls):
super(BaseApplicationCatalogScenarioTest, cls).setUpClass() super(BaseApplicationCatalogScenarioTest, cls).skip_checks()
cls.resource_setup() if not CONF.service_available.murano:
skip_msg = "Murano is disabled"
raise cls.skipException(skip_msg)
@classmethod @classmethod
def tearDownClass(cls): def setup_clients(cls):
cls.resource_cleanup() super(BaseApplicationCatalogScenarioTest, cls).setup_clients()
super(BaseApplicationCatalogScenarioTest, cls).tearDownClass() if not hasattr(cls, "os_primary"):
creds = cls.get_configured_isolated_creds(type_of_creds='primary')
cls.os_primary = clients.Manager(credentials=creds)
cls.services_manager = services_manager(creds)
cls.application_catalog_client = \
cls.os_primary.application_catalog_client
cls.artifacts_client = cls.os_primary.artifacts_client
cls.servers_client = cls.services_manager.servers_client
cls.orchestration_client = cls.services_manager.orchestration_client
cls.snapshots_client = cls.services_manager.snapshots_v2_client
cls.volumes_client = cls.services_manager.volumes_v2_client
cls.backups_client = cls.services_manager.backups_v2_client
cls.images_client = cls.services_manager.image_client_v2
@classmethod
def resource_setup(cls):
super(BaseApplicationCatalogScenarioTest, cls).resource_setup()
cls.linux_image = CONF.application_catalog.linux_image
cls.cirros_image = cls.get_required_image_name()
@classmethod @classmethod
def get_client_with_isolated_creds(cls, type_of_creds="admin"): def get_client_with_isolated_creds(cls, type_of_creds="admin"):
creds = cls.get_configured_isolated_creds(type_of_creds=type_of_creds) creds = cls.get_configured_isolated_creds(type_of_creds=type_of_creds)
os = clients.Manager(credentials=creds) os = clients.Manager(credentials=creds)
@ -60,11 +80,10 @@ class BaseApplicationCatalogScenarioTest(base.BaseTestCase):
cls.admin_role = CONF.identity.admin_role cls.admin_role = CONF.identity.admin_role
else: else:
cls.admin_role = 'admin' cls.admin_role = 'admin'
if not hasattr(cls, 'dynamic_cred'): cls.credentials = common_creds.get_credentials_provider(
cls.credentials = common_creds.get_credentials_provider( name=cls.__name__,
name=cls.__name__, force_tenant_isolation=CONF.auth.use_dynamic_credentials,
force_tenant_isolation=CONF.auth.use_dynamic_credentials, identity_version=CONF.identity.auth_version)
identity_version=CONF.identity.auth_version)
if type_of_creds == 'primary': if type_of_creds == 'primary':
creds = cls.credentials.get_primary_creds() creds = cls.credentials.get_primary_creds()
elif type_of_creds == 'admin': elif type_of_creds == 'admin':
@ -77,48 +96,11 @@ class BaseApplicationCatalogScenarioTest(base.BaseTestCase):
return creds.credentials return creds.credentials
@classmethod
def verify_nonempty(cls, *args):
if not all(args):
msg = "Missing API credentials in configuration."
raise cls.skipException(msg)
@classmethod
def resource_setup(cls):
if not CONF.service_available.murano:
skip_msg = "Murano is disabled"
raise cls.skipException(skip_msg)
if not hasattr(cls, "os_primary"):
creds = cls.get_configured_isolated_creds(type_of_creds='primary')
cls.os_primary = clients.Manager(credentials=creds)
cls.services_manager = services_manager(creds)
cls.linux_image = CONF.application_catalog.linux_image
cls.application_catalog_client = \
cls.os_primary.application_catalog_client
cls.artifacts_client = cls.os_primary.artifacts_client
cls.servers_client = cls.services_manager.servers_client
cls.orchestration_client = cls.services_manager.orchestration_client
cls.snapshots_client = cls.services_manager.snapshots_v2_client
cls.volumes_client = cls.services_manager.volumes_v2_client
cls.backups_client = cls.services_manager.backups_v2_client
cls.images_client = cls.services_manager.image_client_v2
cls.cirros_image = cls.get_required_image_name()
@classmethod @classmethod
def get_required_image_name(cls): def get_required_image_name(cls):
image = cls.images_client.show_image(CONF.compute.image_ref) image = cls.images_client.show_image(CONF.compute.image_ref)
return image['name'] return image['name']
@classmethod
def resource_cleanup(cls):
cls.clear_isolated_creds()
@classmethod
def clear_isolated_creds(cls):
if hasattr(cls, "dynamic_cred"):
cls.credentials.clear_creds()
def environment_delete(self, environment_id, timeout=180): def environment_delete(self, environment_id, timeout=180):
self.application_catalog_client.delete_environment(environment_id) self.application_catalog_client.delete_environment(environment_id)
@ -399,9 +381,20 @@ class BaseApplicationCatalogScenarioIsolatedAdminTest(
BaseApplicationCatalogScenarioTest): BaseApplicationCatalogScenarioTest):
@classmethod @classmethod
def resource_setup(cls): def setup_clients(cls):
creds = cls.get_configured_isolated_creds(type_of_creds='admin') super(BaseApplicationCatalogScenarioIsolatedAdminTest,
cls.os_primary = clients.Manager(credentials=creds) cls).setup_clients()
cls.services_manager = services_manager(creds) if not hasattr(cls, "os_admin"):
super(BaseApplicationCatalogScenarioIsolatedAdminTest, cls).\ creds = cls.get_configured_isolated_creds(type_of_creds='admin')
resource_setup() cls.os_admin = clients.Manager(credentials=creds)
cls.services_manager = services_manager(creds)
cls.application_catalog_client = \
cls.os_admin.application_catalog_client
cls.artifacts_client = cls.os_admin.artifacts_client
cls.servers_client = cls.services_manager.servers_client
cls.orchestration_client = cls.services_manager.orchestration_client
cls.snapshots_client = cls.services_manager.snapshots_v2_client
cls.volumes_client = cls.services_manager.volumes_v2_client
cls.backups_client = cls.services_manager.backups_v2_client
cls.images_client = cls.services_manager.image_client_v2