| | | setting object |
+---------------------------------------+---------------+-----------------------------------------------------------+
+create_security_group_tests.py - CreateMultipleSecurityGroupTests
+-----------------------------------------------------------------
+
++---------------------------------------+---------------+-----------------------------------------------------------+
+| Test Name | API Versions | Description |
++=======================================+===============+===========================================================+
+| test_sec_grp_same_name_diff_proj | Keysone 2 & 3 | Ensures the OpenStackSecurityGroup class does not |
+| | Neutron 2 | initialize security groups with the same name from other |
+| | | project/tenants |
++---------------------------------------+---------------+-----------------------------------------------------------+
+
create_image_tests.py - CreateImageSuccessTests
-----------------------------------------------
sec_grp_ids = list()
for sec_grp_name in self.security_groups:
sec_grp = neutron_utils.get_security_group(
- neutron, sec_grp_name=sec_grp_name)
+ neutron, sec_grp_name=sec_grp_name, project_id=project_id)
if sec_grp:
sec_grp_ids.append(sec_grp.id)
out['security_groups'] = sec_grp_ids
'Rule settings must correspond with the name of this '
'security group')
- def dict_for_neutron(self, keystone):
+ def dict_for_neutron(self, keystone, project_id):
"""
Returns a dictionary object representing this object.
This is meant to be converted into JSON designed for use by the Neutron
TODO - expand automated testing to exercise all parameters
:param keystone: the Keystone client
+ :param project_id: the default project ID
:return: the dictionary object
"""
out = dict()
raise SecurityGroupConfigError(
'Could not find project ID for project named - ' +
self.project_name)
+ else:
+ out['tenant_id'] = project_id
return {'security_group': out}
raise SecurityGroupRuleConfigError(
'direction and sec_grp_name are required')
- def dict_for_neutron(self, neutron):
+ def dict_for_neutron(self, neutron, project_id):
"""
Returns a dictionary object representing this object.
This is meant to be converted into JSON designed for use by the Neutron
API
-
:param neutron: the neutron client for performing lookups
+ :param project_id: the ID of the project associated with the group
:return: the dictionary object
"""
out = dict()
out['protocol'] = self.protocol.value
if self.sec_grp_name:
sec_grp = neutron_utils.get_security_group(
- neutron, sec_grp_name=self.sec_grp_name)
+ neutron, sec_grp_name=self.sec_grp_name, project_id=project_id)
if sec_grp:
out['security_group_id'] = sec_grp.id
else:
super(self.__class__, self).initialize()
self.__security_group = neutron_utils.get_security_group(
- self._neutron, sec_grp_settings=self.sec_grp_settings)
+ self._neutron, sec_grp_settings=self.sec_grp_settings,
+ project_id=self.project_id)
if self.__security_group:
# Populate rules
existing_rules = neutron_utils.get_rules_by_security_group(
keystone = keystone_utils.keystone_client(self._os_creds)
self.__security_group = neutron_utils.create_security_group(
- self._neutron, keystone,
- self.sec_grp_settings)
+ self._neutron, keystone, self.sec_grp_settings,
+ project_id=self.project_id)
# Get the rules added for free
auto_rules = neutron_utils.get_rules_by_security_group(
for sec_grp_rule_setting in self.sec_grp_settings.rule_settings:
try:
custom_rule = neutron_utils.create_security_group_rule(
- self._neutron, sec_grp_rule_setting)
+ self._neutron, sec_grp_rule_setting, self.project_id)
self.__rules[sec_grp_rule_setting] = custom_rule
except Conflict as e:
logger.warn('Unable to create rule due to conflict - %s',
e)
# Refresh security group object to reflect the new rules added
- self.__security_group = neutron_utils.get_security_group(
- self._neutron, sec_grp_settings=self.sec_grp_settings)
+ self.__security_group = neutron_utils.get_security_group_by_id(
+ self._neutron, self.__security_group.id)
return self.__security_group
:param rule_setting: the rule configuration
"""
rule_setting.sec_grp_name = self.sec_grp_settings.name
- new_rule = neutron_utils.create_security_group_rule(self._neutron,
- rule_setting)
+ new_rule = neutron_utils.create_security_group_rule(
+ self._neutron, rule_setting, self.project_id)
self.__rules[rule_setting] = new_rule
self.sec_grp_settings.rule_settings.append(rule_setting)
Constructor
:param os_creds: the OpenStack credentials object
"""
- # super(self.__class__, self, os_creds)
self._os_creds = os_creds
+ keystone = keystone_utils.keystone_client(os_creds)
+ self.project_id = keystone_utils.get_project(
+ keystone=keystone, project_name=os_creds.project_name).id
def initialize(self):
raise NotImplementedError('Do not override abstract method')
from snaps.openstack import create_security_group
from snaps.openstack.create_security_group import (
SecurityGroupSettings, SecurityGroupRuleSettings, Direction, Ethertype,
- Protocol)
+ Protocol, OpenStackSecurityGroup)
from snaps.openstack.tests import validation_utils
from snaps.openstack.tests.os_source_file_test import OSIntegrationTestCase
from snaps.openstack.utils import neutron_utils
return False
return True
+
+
+class CreateMultipleSecurityGroupTests(OSIntegrationTestCase):
+ """
+ Test for the CreateSecurityGroup class and how it interacts with security
+ groups within other projects with the same name
+ """
+
+ def setUp(self):
+ """
+ Instantiates the CreateSecurityGroup object that is responsible for
+ downloading and creating an OS image file within OpenStack
+ """
+ super(self.__class__, self).__start__()
+
+ guid = self.__class__.__name__ + '-' + str(uuid.uuid4())
+ self.sec_grp_name = guid + 'name'
+ self.neutron = neutron_utils.neutron_client(self.os_creds)
+
+ # Initialize for cleanup
+ self.admin_sec_grp_config = SecurityGroupConfig(
+ name=self.sec_grp_name, description='hello group')
+ self.sec_grp_creator_admin = OpenStackSecurityGroup(
+ self.admin_os_creds, self.admin_sec_grp_config)
+ self.sec_grp_creator_admin.create()
+ self.sec_grp_creator_proj = None
+
+ def tearDown(self):
+ """
+ Cleans the image and downloaded image file
+ """
+ if self.sec_grp_creator_admin:
+ self.sec_grp_creator_admin.clean()
+ if self.sec_grp_creator_proj:
+ self.sec_grp_creator_proj.clean()
+
+ super(self.__class__, self).__clean__()
+
+ def test_sec_grp_same_name_diff_proj(self):
+ """
+ Tests the creation of an OpenStack Security Group with the same name
+ within a different project/tenant.
+ """
+ # Create Image
+ sec_grp_config = SecurityGroupConfig(
+ name=self.sec_grp_name, description='hello group')
+ self.sec_grp_creator_proj = OpenStackSecurityGroup(
+ self.os_creds, sec_grp_config)
+ self.sec_grp_creator_proj.create()
+
+ self.assertNotEqual(
+ self.sec_grp_creator_admin.get_security_group().id,
+ self.sec_grp_creator_proj.get_security_group().id)
+
+ admin_sec_grp_creator = OpenStackSecurityGroup(
+ self.admin_os_creds, self.admin_sec_grp_config)
+ admin_sec_grp_creator.create()
+ self.assertEqual(self.sec_grp_creator_admin.get_security_group().id,
+ admin_sec_grp_creator.get_security_group().id)
+
+ proj_sec_grp_creator = OpenStackSecurityGroup(
+ self.os_creds, sec_grp_config)
+ proj_sec_grp_creator.create()
+ self.assertEqual(self.sec_grp_creator_proj.get_security_group().id,
+ proj_sec_grp_creator.get_security_group().id)
self.image_metadata = image_metadata
+ keystone = keystone_utils.keystone_client(self.os_creds)
+ self.project_id = keystone_utils.get_project(
+ keystone=keystone, project_name=self.os_creds.project_name)
+
@staticmethod
def parameterize(testcase_klass, os_creds, ext_net_name,
image_metadata=None, log_level=logging.DEBUG):
return out
-def create_security_group(neutron, keystone, sec_grp_settings):
+def create_security_group(neutron, keystone, sec_grp_settings, project_id):
"""
Creates a security group object in OpenStack
:param neutron: the Neutron client
:param keystone: the Keystone client
:param sec_grp_settings: the security group settings
+ :param project_id: the default project to associated the security group
:return: a SNAPS-OO SecurityGroup domain object
"""
logger.info('Creating security group with name - %s',
sec_grp_settings.name)
os_group = neutron.create_security_group(
- sec_grp_settings.dict_for_neutron(keystone))
+ sec_grp_settings.dict_for_neutron(keystone, project_id))
return __map_os_security_group(neutron, os_group['security_group'])
return None
-def create_security_group_rule(neutron, sec_grp_rule_settings):
+def create_security_group_rule(neutron, sec_grp_rule_settings, proj_id):
"""
Creates a security group rule in OpenStack
:param neutron: the client
:param sec_grp_rule_settings: the security group rule settings
+ :param proj_id: the default project to apply to the rule settings
:return: a SNAPS-OO SecurityGroupRule domain object
"""
logger.info('Creating security group to security group - %s',
sec_grp_rule_settings.sec_grp_name)
os_rule = neutron.create_security_group_rule(
- sec_grp_rule_settings.dict_for_neutron(neutron))
+ sec_grp_rule_settings.dict_for_neutron(neutron, proj_id))
return SecurityGroupRule(**os_rule['security_group_rule'])
Tests the neutron_utils.create_security_group() function
"""
sec_grp_settings = SecurityGroupConfig(name=self.sec_grp_name)
- security_group = neutron_utils.create_security_group(self.neutron,
- self.keystone,
- sec_grp_settings)
+ security_group = neutron_utils.create_security_group(
+ self.neutron, self.keystone, sec_grp_settings, self.project_id)
self.assertTrue(sec_grp_settings.name, security_group.name)
with self.assertRaises(Exception):
sec_grp_settings = SecurityGroupConfig()
self.security_groups.append(
- neutron_utils.create_security_group(self.neutron,
- self.keystone,
- sec_grp_settings))
+ neutron_utils.create_security_group(
+ self.neutron, self.keystone, sec_grp_settings,
+ self.project_id))
def test_create_sec_grp_no_rules(self):
"""
sec_grp_settings = SecurityGroupConfig(
name=self.sec_grp_name, description='hello group')
self.security_groups.append(
- neutron_utils.create_security_group(self.neutron, self.keystone,
- sec_grp_settings))
+ neutron_utils.create_security_group(
+ self.neutron, self.keystone, sec_grp_settings,
+ self.project_id))
self.assertTrue(sec_grp_settings.name, self.security_groups[0].name)
self.assertEqual(sec_grp_settings.name, self.security_groups[0].name)
rule_settings=[sec_grp_rule_settings])
self.security_groups.append(
- neutron_utils.create_security_group(self.neutron, self.keystone,
- sec_grp_settings))
+ neutron_utils.create_security_group(
+ self.neutron, self.keystone, sec_grp_settings,
+ self.project_id))
free_rules = neutron_utils.get_rules_by_security_group(
self.neutron, self.security_groups[0])
for free_rule in free_rules:
self.security_group_rules.append(free_rule)
+ keystone = keystone_utils.keystone_client(self.os_creds)
+ project_id = keystone_utils.get_project(
+ keystone, self.os_creds.project_name)
self.security_group_rules.append(
neutron_utils.create_security_group_rule(
- self.neutron, sec_grp_settings.rule_settings[0]))
+ self.neutron, sec_grp_settings.rule_settings[0], project_id))
# Refresh object so it is populated with the newly added rule
security_group = neutron_utils.get_security_group(
self.security_groups.append(neutron_utils.create_security_group(
self.neutron, self.keystone,
SecurityGroupConfig(
- name=self.sec_grp_name + '-1', description='hello group')))
+ name=self.sec_grp_name + '-1', description='hello group'),
+ self.project_id))
self.security_groups.append(neutron_utils.create_security_group(
self.neutron, self.keystone,
SecurityGroupConfig(
- name=self.sec_grp_name + '-2', description='hello group')))
+ name=self.sec_grp_name + '-2', description='hello group'),
+ self.project_id))
sec_grp_1b = neutron_utils.get_security_group_by_id(
self.neutron, self.security_groups[0].id)
RouterSettingsUnitTests)
from snaps.openstack.tests.create_security_group_tests import (
CreateSecurityGroupTests, SecurityGroupRuleSettingsUnitTests,
- SecurityGroupSettingsUnitTests)
+ SecurityGroupSettingsUnitTests, CreateMultipleSecurityGroupTests)
from snaps.openstack.tests.create_stack_tests import (
StackSettingsUnitTests, CreateStackSuccessTests, CreateStackNegativeTests,
CreateStackFlavorTests, CreateStackFloatingIpTests,
use_keystone=use_keystone,
flavor_metadata=flavor_metadata, image_metadata=image_metadata,
log_level=log_level))
+ suite.addTest(OSIntegrationTestCase.parameterize(
+ CreateMultipleSecurityGroupTests, os_creds=os_creds,
+ ext_net_name=ext_net_name, use_keystone=use_keystone,
+ flavor_metadata=flavor_metadata, image_metadata=image_metadata,
+ log_level=log_level))
suite.addTest(OSIntegrationTestCase.parameterize(
CreateImageSuccessTests, os_creds=os_creds, ext_net_name=ext_net_name,
use_keystone=use_keystone,
Code Review / snaps.git / commitdiff