self.__role = keystone_utils.create_role(
self.__keystone, self.project_settings.name + '-role')
- keystone_utils.assoc_user_to_project(self.__keystone, self.__role,
- user, self.__project)
+ keystone_utils.grant_user_role_to_project(self.__keystone, self.__role,
+ user, self.__project)
class ProjectSettings:
self.__keystone = keystone_utils.keystone_client(self.__os_creds)
self.__user = keystone_utils.get_user(self.__keystone,
self.user_settings.name)
- if self.__user:
- logger.info('Found user with name - ' + self.user_settings.name)
- elif not cleanup:
+ if not self.__user and not cleanup:
self.__user = keystone_utils.create_user(self.__keystone,
self.user_settings)
- else:
- logger.info('Did not create user due to cleanup mode')
return self.__user
:param email: the user's email address (optional)
:param enabled: denotes whether or not the user is enabled
(default True)
+ :param roles: dict where key is the role name and value is a list of
+ project names
"""
self.name = kwargs.get('name')
self.password = kwargs.get('password')
self.project_name = kwargs.get('project_name')
self.email = kwargs.get('email')
-
- if kwargs.get('domain_name'):
- self.domain_name = kwargs['domain_name']
- else:
- self.domain_name = 'default'
-
- if kwargs.get('enabled') is not None:
- self.enabled = kwargs['enabled']
- else:
- self.enabled = True
+ self.domain_name = kwargs.get('domain_name', 'default')
+ self.enabled = kwargs.get('enabled', True)
+ self.roles = kwargs.get('roles', dict())
if not self.name or not self.password:
- raise Exception(
+ raise UserSettingsException(
'The attributes name and password are required for '
'UserSettings')
if not isinstance(self.enabled, bool):
- raise Exception('The attribute enabled must be of type boolean')
+ raise UserSettingsException('The attribute enabled must be of type'
+ ' boolean')
+
+
+class UserSettingsException(Exception):
+ """
+ Raised when there is a problem with the values set in the UserSettings
+ class
+ """
created_project = self.project_creator.create()
self.assertIsNotNone(created_project)
- user_creator = OpenStackUser(self.os_creds,
- UserSettings(name=self.guid + '-user',
- password=self.guid))
+ user_creator = OpenStackUser(
+ self.os_creds, UserSettings(
+ name=self.guid + '-user',
+ password=self.guid, roles={'admin': 'admin'}))
self.project_creator.assoc_user(user_creator.create())
self.user_creators.append(user_creator)
created_project = self.project_creator.create()
self.assertIsNotNone(created_project)
- user_creator_1 = OpenStackUser(self.os_creds,
- UserSettings(name=self.guid + '-user1',
- password=self.guid))
+ user_creator_1 = OpenStackUser(
+ self.os_creds, UserSettings(
+ name=self.guid + '-user1', password=self.guid,
+ roles={'admin': 'admin'}))
self.project_creator.assoc_user(user_creator_1.create())
self.user_creators.append(user_creator_1)
- user_creator_2 = OpenStackUser(self.os_creds,
- UserSettings(name=self.guid + '-user2',
- password=self.guid))
+ user_creator_2 = OpenStackUser(
+ self.os_creds, UserSettings(
+ name=self.guid + '-user2', password=self.guid,
+ roles={'admin': 'admin'}))
self.project_creator.assoc_user(user_creator_2.create())
self.user_creators.append(user_creator_2)
guid = str(uuid.uuid4())[:-19]
guid = self.__class__.__name__ + '-' + guid
self.user_settings = UserSettings(name=guid + '-name',
- password=guid + '-password')
+ password=guid + '-password',
+ roles={'admin': 'admin'})
self.keystone = keystone_utils.keystone_client(self.os_creds)
# Delete user
self.user_creator.clean()
self.assertIsNone(self.user_creator.get_user())
+
+ def test_create_admin_user(self):
+ """
+ Tests the creation of an OpenStack user.
+ """
+ self.user_creator = OpenStackUser(self.os_creds, self.user_settings)
+ created_user = self.user_creator.create()
+ self.assertIsNotNone(created_user)
+
+ retrieved_user = keystone_utils.get_user(self.keystone,
+ self.user_settings.name)
+ self.assertIsNotNone(retrieved_user)
+ self.assertEqual(created_user, retrieved_user)
+
+ role = keystone_utils.get_os_role_by_name(self.keystone, 'admin')
+ self.assertIsNotNone(role)
+
+ os_proj = keystone_utils.get_project(
+ keystone=self.keystone, project_name=self.os_creds.project_name)
+ user_roles = keystone_utils.get_os_roles_by_user(
+ self.keystone, retrieved_user, os_proj)
+ self.assertIsNotNone(user_roles)
+ self.assertEqual(1, len(user_roles))
+ self.assertEqual(role.id, user_roles[0].id)
from snaps.openstack.utils import deploy_utils, keystone_utils
-dev_os_env_file = pkg_resources.resource_filename('snaps.openstack.tests.conf', 'os_env.yaml')
-
-# !!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!
-# To run these tests from an IDE, the CWD must be set to the snaps directory of this project
-# !!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!
+dev_os_env_file = pkg_resources.resource_filename(
+ 'snaps.openstack.tests.conf', 'os_env.yaml')
class OSComponentTestCase(unittest.TestCase):
- def __init__(self, method_name='runTest', os_creds=None, ext_net_name=None, image_metadata=None,
- log_level=logging.DEBUG):
+ def __init__(self, method_name='runTest', os_creds=None, ext_net_name=None,
+ image_metadata=None, log_level=logging.DEBUG):
"""
Super for test classes requiring a connection to OpenStack
:param method_name: default 'runTest'
- :param os_creds: the OSCreds object, when null it searches for the file {cwd}/openstack/tests/conf/os_env.yaml
- :param ext_net_name: the name of the external network that is used for creating routers for floating IPs
- :param image_metadata: ability to override images being used in the tests (see examples/image-metadata)
+ :param os_creds: the OSCreds object, when null it searches for the file
+ in the package snaps.openstack.tests.conf.os_env.yaml
+ :param ext_net_name: the name of the external network that is used for
+ creating routers for floating IPs
+ :param image_metadata: ability to override images being used in the
+ tests (see examples/image-metadata)
:param log_level: the logging level of your test run (default DEBUG)
"""
super(OSComponentTestCase, self).__init__(method_name)
if os_creds:
self.os_creds = os_creds
else:
- self.os_creds = openstack_tests.get_credentials(dev_os_env_file=dev_os_env_file)
+ self.os_creds = openstack_tests.get_credentials(
+ dev_os_env_file=dev_os_env_file)
self.ext_net_name = ext_net_name
self.image_metadata = image_metadata
@staticmethod
- def parameterize(testcase_klass, os_creds, ext_net_name, image_metadata=None, log_level=logging.DEBUG):
+ def parameterize(testcase_klass, os_creds, ext_net_name,
+ image_metadata=None, log_level=logging.DEBUG):
""" Create a suite containing all tests taken from the given
subclass, passing them the parameter 'param'.
"""
test_names = test_loader.getTestCaseNames(testcase_klass)
suite = unittest.TestSuite()
for name in test_names:
- suite.addTest(testcase_klass(name, os_creds, ext_net_name, image_metadata, log_level))
+ suite.addTest(testcase_klass(name, os_creds, ext_net_name,
+ image_metadata, log_level))
return suite
class OSIntegrationTestCase(OSComponentTestCase):
- def __init__(self, method_name='runTest', os_creds=None, ext_net_name=None, use_keystone=False,
- flavor_metadata=None, image_metadata=None, log_level=logging.DEBUG):
+ def __init__(self, method_name='runTest', os_creds=None, ext_net_name=None,
+ use_keystone=False, flavor_metadata=None, image_metadata=None,
+ log_level=logging.DEBUG):
"""
Super for integration tests requiring a connection to OpenStack
:param method_name: default 'runTest'
- :param os_creds: the OSCreds object, when null it searches for the file {cwd}/openstack/tests/conf/os_env.yaml
- :param ext_net_name: the name of the external network that is used for creating routers for floating IPs
- :param use_keystone: when true, these tests will create a new user/project under which to run the test
- :param image_metadata: dict() containing the URLs for the disk, kernel, and ramdisk images when multi-part
- images are required.
- image_metadata={'disk_url': 'http://download.cirros-cloud.net/0.3.4/cirros-0.3.4-x86_64-disk.img',
- 'kernel_url': 'http://download.cirros-cloud.net/0.3.4/cirros-0.3.4-x86_64-kernel',
- 'ramdisk_url': 'http://download.cirros-cloud.net/0.3.4/cirros-0.3.4-x86_64-initramfs'})
- :param flavor_metadata: dict() to be sent directly into the Nova client generally used for page sizes
+ :param os_creds: the OSCreds object, when null it searches for the file
+ in the package snaps.openstack.tests.conf.os_env.yaml
+ :param ext_net_name: the name of the external network that is used for
+ creating routers for floating IPs
+ :param use_keystone: when true, these tests will create a new
+ user/project under which to run the test
+ :param image_metadata: dict() containing the URLs for the disk, kernel,
+ and ramdisk images when multi-part images are
+ required. See below for a simple example
+ image_metadata={'disk_url': '{URI}/cirros-0.3.4-x86_64-disk.img',
+ 'kernel_url': '{URI}/cirros-0.3.4-x86_64-kernel',
+ 'ramdisk_url': '{URI}/cirros-0.3.4-x86_64-initramfs'})
+ :param flavor_metadata: dict() to be sent directly into the Nova client
+ generally used for page sizes
:param log_level: the logging level of your test run (default DEBUG)
"""
- super(OSIntegrationTestCase, self).__init__(method_name=method_name, os_creds=os_creds,
- ext_net_name=ext_net_name, image_metadata=image_metadata,
- log_level=log_level)
+ super(OSIntegrationTestCase, self).__init__(
+ method_name=method_name, os_creds=os_creds,
+ ext_net_name=ext_net_name, image_metadata=image_metadata,
+ log_level=log_level)
self.use_keystone = use_keystone
self.keystone = None
self.flavor_metadata = flavor_metadata
@staticmethod
- def parameterize(testcase_klass, os_creds, ext_net_name, use_keystone=False, flavor_metadata=None,
+ def parameterize(testcase_klass, os_creds, ext_net_name,
+ use_keystone=False, flavor_metadata=None,
image_metadata=None, log_level=logging.DEBUG):
"""
Create a suite containing all tests taken from the given
test_names = test_loader.getTestCaseNames(testcase_klass)
suite = unittest.TestSuite()
for name in test_names:
- suite.addTest(testcase_klass(name, os_creds, ext_net_name, use_keystone, flavor_metadata, image_metadata,
- log_level))
+ suite.addTest(testcase_klass(name, os_creds, ext_net_name,
+ use_keystone, flavor_metadata,
+ image_metadata, log_level))
return suite
"""
- Super for test classes that should be run within their own project/tenant as they can run for quite some time
+ Super for test classes that should be run within their own project/tenant
+ as they can run for quite some time
"""
def __start__(self):
"""
- Creates a project and user to be leveraged by subclass test methods. If implementing class uses this method,
- it must call __clean__() else you will be left with unwanted users and tenants
+ Creates a project and user to be leveraged by subclass test methods. If
+ implementing class uses this method, it must call __clean__() else you
+ will be left with unwanted users and tenants
"""
self.project_creator = None
self.user_creator = None
self.keystone = keystone_utils.keystone_client(self.os_creds)
guid = self.__class__.__name__ + '-' + str(uuid.uuid4())[:-19]
project_name = guid + '-proj'
- self.project_creator = deploy_utils.create_project(self.admin_os_creds, ProjectSettings(name=project_name))
+ self.project_creator = deploy_utils.create_project(
+ self.admin_os_creds, ProjectSettings(name=project_name))
self.user_creator = deploy_utils.create_user(
- self.admin_os_creds, UserSettings(name=guid + '-user', password=guid, project_name=project_name))
- self.os_creds = self.user_creator.get_os_creds(self.project_creator.project_settings.name)
+ self.admin_os_creds, UserSettings(
+ name=guid + '-user', password=guid,
+ project_name=project_name, roles={
+ 'admin': self.project_creator.project_settings.name}))
+
+ self.os_creds = self.user_creator.get_os_creds(
+ self.project_creator.project_settings.name)
# add user to project
self.project_creator.assoc_user(self.user_creator.get_user())
def __clean__(self):
"""
Cleans up test user and project.
- Must be called at the end of child classes tearDown() if __start__() is called during setUp() else these
- objects will persist after the test is run
+ Must be called at the end of child classes tearDown() if __start__() is
+ called during setUp() else these objects will persist after the test is
+ run
"""
if self.role:
keystone_utils.delete_role(self.keystone, self.role)
email=user_settings.email, tenant_id=project_id,
enabled=user_settings.enabled)
else:
- # TODO - need to support groups
os_user = keystone.users.create(
name=user_settings.name, password=user_settings.password,
email=user_settings.email, project=project,
domain=user_settings.domain_name, enabled=user_settings.enabled)
+ for role_name, role_project in user_settings.roles.items():
+ os_role = get_os_role_by_name(keystone, role_name)
+ os_project = get_project(keystone=keystone, project_name=role_project)
+
+ if os_role and os_project:
+ existing_roles = get_os_roles_by_user(keystone, os_user,
+ os_project)
+ found = False
+ for role in existing_roles:
+ if role.id == os_role.id:
+ found = True
+
+ if not found:
+ grant_user_role_to_project(
+ keystone=keystone, user=os_user, role=os_role,
+ project=os_project)
+
if os_user:
return User(name=os_user.name, user_id=os_user.id)
keystone.users.delete(user.id)
+def get_os_role_by_name(keystone, name):
+ """
+ Returns an OpenStack role object of a given name or None if not exists
+ :param keystone: the keystone client
+ :param name: the role name
+ :return: the OpenStack role object
+ """
+ roles = keystone.roles.list()
+ for role in roles:
+ if role.name == name:
+ return role
+
+
+def get_os_roles_by_user(keystone, user, project):
+ """
+ Returns a list of OpenStack role object associated with a user
+ :param keystone: the keystone client
+ :param user: the OpenStack user object
+ :param project: the OpenStack project object (only required for v2)
+ :return: a list of OpenStack role objects
+ """
+ if keystone.version == V2_VERSION:
+ os_user = get_os_user(keystone, user)
+ roles = keystone.roles.roles_for_user(os_user, project)
+ return roles
+ else:
+ return keystone.roles.list(user=user, project=project)
+
+
+def get_os_role_by_id(keystone, role_id):
+ """
+ Returns an OpenStack role object of a given name or None if not exists
+ :param keystone: the keystone client
+ :param role_id: the role ID
+ :return: the OpenStack role object
+ """
+ return keystone.roles.get(role_id)
+
+
def create_role(keystone, name):
"""
Creates an OpenStack role
keystone.roles.delete(role)
-def assoc_user_to_project(keystone, role, user, project):
+def grant_user_role_to_project(keystone, role, user, project):
"""
- Adds a user to a project
+ Grants user and role to a project
:param keystone: the Keystone client
:param role: the role used to join a project/user
:param user: the user to add to the project (SNAPS-OO User Domain object
Instantiates the CreateImage object that is responsible for downloading
and creating an OS image file within OpenStack
"""
- guid = uuid.uuid4()
- self.username = self.__class__.__name__ + '-' + str(guid)
+ self.guid = self.__class__.__name__ + '-' + str(uuid.uuid4())
+ self.username = self.guid + '-username'
self.user = None
- self.project_name = self.__class__.__name__ + '-' + str(guid)
+ self.project_name = self.guid + '-projName'
self.project = None
+ self.role = None
self.keystone = keystone_utils.keystone_client(self.os_creds)
def tearDown(self):
if self.user:
keystone_utils.delete_user(self.keystone, self.user)
+ if self.role:
+ keystone_utils.delete_role(self.keystone, self.role)
+
def test_create_user_minimal(self):
"""
Tests the keystone_utils.create_user() function
self.assertNotEqual(endpoint_public, endpoint_internal)
self.assertNotEqual(endpoint_public, endpoint_admin)
self.assertEqual(endpoint_admin, endpoint_internal)
+
+ def test_grant_user_role_to_project(self):
+ """
+ Tests the keystone_utils function grant_user_role_to_project()
+ :return:
+ """
+ user_settings = UserSettings(name=self.username,
+ password=str(uuid.uuid4()))
+ self.user = keystone_utils.create_user(self.keystone, user_settings)
+ self.assertEqual(self.username, self.user.name)
+
+ project_settings = ProjectSettings(name=self.project_name)
+ self.project = keystone_utils.create_project(self.keystone,
+ project_settings)
+ self.assertEqual(self.project_name, self.project.name)
+
+ role_name = self.guid + '-role'
+ self.role = keystone_utils.create_role(self.keystone, role_name)
+ self.assertEqual(role_name, self.role.name)
+
+ keystone_utils.grant_user_role_to_project(
+ self.keystone, self.role, self.user, self.project)
+
+ user_roles = keystone_utils.get_os_roles_by_user(
+ self.keystone, self.user, self.project)
+ self.assertIsNotNone(user_roles)
+ self.assertEqual(1, len(user_roles))
+ self.assertEqual(self.role.id, user_roles[0].id)
Code Review / snaps.git / commitdiff