else:
return None
- @classmethod
- def _get_fake_keystone_client(cls):
- kclient = mock.Mock()
- kclient.service_catalog = mock.Mock()
- kclient.service_catalog.url_for = mock.Mock(
- side_effect=cls._fake_url_for)
- return kclient
-
def _get_main_kwargs(self, key=None):
kwargs = {'odlusername': self._odl_username,
'odlpassword': self._odl_password,
def _test_run(self, status=testcase_base.TestcaseBase.EX_OK,
exception=None, odlip="127.0.0.3", odlwebport="8080"):
- with mock.patch('functest.utils.openstack_utils.get_keystone_client',
- return_value=self._get_fake_keystone_client()):
+ with mock.patch('functest.utils.openstack_utils.get_endpoint',
+ side_effect=[self._fake_url_for('identity'),
+ self._fake_url_for('network')]):
if exception:
self.test.main = mock.Mock(side_effect=exception)
else:
odlwebport=self._odl_webport)
def test_run_missing_sdn_controller_ip(self):
- with mock.patch('functest.utils.openstack_utils.get_keystone_client',
- return_value=self._get_fake_keystone_client()):
- ft_constants.CI_INSTALLER_TYPE = None
- ft_constants.SDN_CONTROLLER_IP = None
- self.assertEqual(self.test.run(),
- testcase_base.TestcaseBase.EX_RUN_ERROR)
+ ft_constants.CI_INSTALLER_TYPE = None
+ ft_constants.SDN_CONTROLLER_IP = None
+ self.assertEqual(self.test.run(),
+ testcase_base.TestcaseBase.EX_RUN_ERROR)
def test_run_without_installer_type(self):
ft_constants.SDN_CONTROLLER_IP = self._sdn_controller_ip
odlip=self._neutron_ip, odlwebport='8282')
def test_run_apex_missing_sdn_controller_ip(self):
- with mock.patch('functest.utils.openstack_utils.get_keystone_client',
- return_value=self._get_fake_keystone_client()):
- ft_constants.CI_INSTALLER_TYPE = "apex"
- ft_constants.SDN_CONTROLLER_IP = None
- self.assertEqual(self.test.run(),
- testcase_base.TestcaseBase.EX_RUN_ERROR)
+ ft_constants.CI_INSTALLER_TYPE = "apex"
+ ft_constants.SDN_CONTROLLER_IP = None
+ self.assertEqual(self.test.run(),
+ testcase_base.TestcaseBase.EX_RUN_ERROR)
def test_run_apex(self):
ft_constants.SDN_CONTROLLER_IP = self._sdn_controller_ip
odlip=self._sdn_controller_ip, odlwebport='8181')
def test_run_joid_missing_sdn_controller(self):
- with mock.patch('functest.utils.openstack_utils.get_keystone_client',
- return_value=self._get_fake_keystone_client()):
- ft_constants.CI_INSTALLER_TYPE = "joid"
- ft_constants.SDN_CONTROLLER = None
- self.assertEqual(self.test.run(),
- testcase_base.TestcaseBase.EX_RUN_ERROR)
+ ft_constants.CI_INSTALLER_TYPE = "joid"
+ ft_constants.SDN_CONTROLLER = None
+ self.assertEqual(self.test.run(),
+ testcase_base.TestcaseBase.EX_RUN_ERROR)
def test_run_joid(self):
ft_constants.SDN_CONTROLLER = self._sdn_controller_ip
import sys
import time
+from keystoneauth1 import loading
+from keystoneauth1 import session
from cinderclient import client as cinderclient
-import functest.utils.functest_logger as ft_logger
-import functest.utils.functest_utils as ft_utils
from glanceclient import client as glanceclient
-from keystoneclient.v2_0 import client as keystoneclient
-from neutronclient.v2_0 import client as neutronclient
from novaclient import client as novaclient
+from keystoneclient import client as keystoneclient
+from neutronclient.neutron import client as neutronclient
+
+import functest.utils.functest_logger as ft_logger
+import functest.utils.functest_utils as ft_utils
logger = ft_logger.Logger("openstack_utils").getLogger()
+DEFAULT_API_VERSION = '2'
+
# *********************************************
# CREDENTIALS
return all(map(lambda v: v in os.environ and os.environ[v], env_vars))
-def get_credentials(service):
- """Returns a creds dictionary filled with the following keys:
- * username
- * password/api_key (depending on the service)
- * tenant_name/project_id (depending on the service)
- * auth_url
- :param service: a string indicating the name of the service
- requesting the credentials.
+def get_credentials():
+ """Returns a creds dictionary filled with parsed from env
"""
creds = {}
if os.getenv(envvar) is None:
raise MissingEnvVar(envvar)
- # Unfortunately, each of the OpenStack client will request slightly
- # different entries in their credentials dict.
- if service.lower() in ("nova", "cinder"):
- password = "api_key"
- tenant = "project_id"
- else:
- password = "password"
-
# The most common way to pass these info to the script is to do it through
# environment variables.
creds.update({
"username": os.environ.get("OS_USERNAME"),
- password: os.environ.get("OS_PASSWORD"),
+ "password": os.environ.get("OS_PASSWORD"),
"auth_url": os.environ.get("OS_AUTH_URL"),
tenant: os.environ.get(tenant_env)
})
def get_credentials_for_rally():
- creds = get_credentials("keystone")
+ creds = get_credentials()
keystone_api_version = os.getenv('OS_IDENTITY_API_VERSION')
if (keystone_api_version is None or
keystone_api_version == '2'):
return rally_conf
+def get_session_auth():
+ loader = loading.get_plugin_loader('password')
+ creds = get_credentials()
+ auth = loader.load_from_options(**creds)
+ return auth
+
+
+def get_endpoint(service_type, endpoint_type='publicURL'):
+ auth = get_session_auth()
+ return get_session().get_endpoint(auth=auth,
+ service_type=service_type,
+ endpoint_type=endpoint_type)
+
+
+def get_session():
+ auth = get_session_auth()
+ return session.Session(auth=auth)
+
+
# *********************************************
# CLIENTS
# *********************************************
+def get_keystone_client_version():
+ api_version = os.getenv('OS_IDENTITY_API_VERSION')
+ if api_version is not None:
+ logger.info("OS_IDENTITY_API_VERSION is set in env as '%s'",
+ api_version)
+ return api_version
+ return DEFAULT_API_VERSION
+
+
def get_keystone_client():
- creds_keystone = get_credentials("keystone")
- return keystoneclient.Client(**creds_keystone)
+ sess = get_session()
+ return keystoneclient.Client(get_keystone_client_version(), session=sess)
+
+
+def get_nova_client_version():
+ api_version = os.getenv('OS_COMPUTE_API_VERSION')
+ if api_version is not None:
+ logger.info("OS_COMPUTE_API_VERSION is set in env as '%s'",
+ api_version)
+ return api_version
+ return DEFAULT_API_VERSION
def get_nova_client():
- creds_nova = get_credentials("nova")
- return novaclient.Client('2', **creds_nova)
+ sess = get_session()
+ return novaclient.Client(get_nova_client_version(), session=sess)
+
+
+def get_cinder_client_version():
+ api_version = os.getenv('OS_VOLUME_API_VERSION')
+ if api_version is not None:
+ logger.info("OS_VOLUME_API_VERSION is set in env as '%s'",
+ api_version)
+ return api_version
+ return DEFAULT_API_VERSION
def get_cinder_client():
- creds_cinder = get_credentials("cinder")
- creds_cinder.update({
- "service_type": "volume"
- })
- return cinderclient.Client('2', **creds_cinder)
+ sess = get_session()
+ return cinderclient.Client(get_cinder_client_version(), session=sess)
+
+
+def get_neutron_client_version():
+ api_version = os.getenv('OS_NETWORK_API_VERSION')
+ if api_version is not None:
+ logger.info("OS_NETWORK_API_VERSION is set in env as '%s'",
+ api_version)
+ return api_version
+ return DEFAULT_API_VERSION
def get_neutron_client():
- creds_neutron = get_credentials("neutron")
- return neutronclient.Client(**creds_neutron)
+ sess = get_session()
+ return neutronclient.Client(get_neutron_client_version(), session=sess)
+
+
+def get_glance_client_version():
+ api_version = os.getenv('OS_IMAGE_API_VERSION')
+ if api_version is not None:
+ logger.info("OS_IMAGE_API_VERSION is set in env as '%s'", api_version)
+ return api_version
+ return DEFAULT_API_VERSION
def get_glance_client():
- keystone_client = get_keystone_client()
- glance_endpoint_type = 'publicURL'
- os_endpoint_type = os.getenv('OS_ENDPOINT_TYPE')
- if os_endpoint_type is not None:
- glance_endpoint_type = os_endpoint_type
- glance_endpoint = keystone_client.service_catalog.url_for(
- service_type='image', endpoint_type=glance_endpoint_type)
- return glanceclient.Client(1, glance_endpoint,
- token=keystone_client.auth_token)
+ sess = get_session()
+ return glanceclient.Client(get_glance_client_version(), session=sess)
# *********************************************
def create_glance_image(glance_client, image_name, file_path, disk="qcow2",
- container="bare", public=True):
+ container="bare", public="public"):
if not os.path.isfile(file_path):
logger.error("Error: file %s does not exist." % file_path)
return None
try:
image_id = get_image_id(glance_client, image_name)
if image_id != '':
- if logger:
- logger.info("Image %s already exists." % image_name)
+ logger.info("Image %s already exists." % image_name)
else:
- if logger:
- logger.info("Creating image '%s' from '%s'..." % (image_name,
- file_path))
- try:
- properties = ft_utils.get_functest_config(
- 'general.image_properties')
- except ValueError:
- # image properties are not configured
- # therefore don't add any properties
- properties = {}
- with open(file_path) as fimage:
- image = glance_client.images.create(name=image_name,
- is_public=public,
- disk_format=disk,
- container_format=container,
- properties=properties,
- data=fimage)
+ logger.info("Creating image '%s' from '%s'..." % (image_name,
+ file_path))
+
+ image = glance_client.images.create(name=image_name,
+ visibility=public,
+ disk_format=disk,
+ container_format=container)
image_id = image.id
+ with open(file_path) as image_data:
+ glance_client.images.upload(image_id, image_data)
return image_id
except Exception, e:
logger.error("Error [create_glance_image(glance_client, '%s', '%s', "
- "'%s')]: %s" % (image_name, file_path, str(public), e))
+ "'%s')]: %s" % (image_name, file_path, public, e))
return None
Code Review / functest-xtesting.git / commitdiff