--- /dev/null
+# Copyright (c) 2017 Cable Television Laboratories, Inc. ("CableLabs")
+# and others. All rights reserved.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at:
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+
+class SecurityGroup:
+ """
+ SNAPS domain object for SecurityGroups. Should contain attributes that
+ are shared amongst cloud providers
+ """
+ def __init__(self, **kwargs):
+ """
+ Constructor
+ :param name: the security group's name
+ :param id: the security group's id
+ """
+ self.name = kwargs.get('name')
+ self.id = kwargs.get('id')
+ self.project_id = kwargs.get('project_id', kwargs.get('tenant_id'))
+
+ def __eq__(self, other):
+ return (self.name == other.name and self.id == other.id and
+ self.project_id == other.project_id)
+
+
+class SecurityGroupRule:
+ """
+ SNAPS domain object for Security Group Rules. Should contain attributes
+ that are shared amongst cloud providers
+ """
+ def __init__(self, **kwargs):
+ """
+ Constructor
+ :param id: the security group rule's id
+ :param sec_grp_id: the ID of the associated security group
+ :param description: the security group rule's description
+ :param direction: the security group rule's direction
+ :param ethertype: the security group rule's ethertype
+ :param port_range_min: the security group rule's port_range_min
+ :param port_range_max: the security group rule's port_range_max
+ :param protocol: the security group rule's protocol
+ :param remote_group_id: the security group rule's remote_group_id
+ :param remote_ip_prefix: the security group rule's remote_ip_prefix
+ """
+ self.id = kwargs.get('id')
+ self.security_group_id = kwargs.get('security_group_id')
+ self.description = kwargs.get('description')
+ self.direction = kwargs.get('direction')
+ self.ethertype = kwargs.get('ethertype')
+ self.port_range_min = kwargs.get('port_range_min')
+ self.port_range_max = kwargs.get('port_range_max')
+ self.protocol = kwargs.get('protocol')
+ self.remote_group_id = kwargs.get('remote_group_id')
+ self.remote_ip_prefix = kwargs.get('remote_ip_prefix')
+
+ def __eq__(self, other):
+ return (self.id == other.id and
+ self.security_group_id == other.security_group_id and
+ self.description == other.description and
+ self.direction == other.direction and
+ self.ethertype == other.ethertype and
+ self.port_range_min == other.port_range_min and
+ self.port_range_max == other.port_range_max and
+ self.protocol == other.protocol and
+ self.remote_group_id == other.remote_group_id and
+ self.remote_ip_prefix == other.remote_ip_prefix)
--- /dev/null
+# Copyright (c) 2017 Cable Television Laboratories, Inc. ("CableLabs")
+# and others. All rights reserved.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at:
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import unittest
+from snaps.domain.network import SecurityGroup, SecurityGroupRule
+
+
+class SecurityGroupDomainObjectTests(unittest.TestCase):
+ """
+ Tests the construction of the snaps.domain.test.SecurityGroup class
+ """
+
+ def test_construction_proj_id_kwargs(self):
+ sec_grp = SecurityGroup(
+ **{'name': 'name', 'id': 'id',
+ 'project_id': 'foo'})
+ self.assertEqual('name', sec_grp.name)
+ self.assertEqual('id', sec_grp.id)
+ self.assertEqual('foo', sec_grp.project_id)
+
+ def test_construction_tenant_id_kwargs(self):
+ sec_grp = SecurityGroup(
+ **{'name': 'name', 'id': 'id',
+ 'tenant_id': 'foo'})
+ self.assertEqual('name', sec_grp.name)
+ self.assertEqual('id', sec_grp.id)
+ self.assertEqual('foo', sec_grp.project_id)
+
+ def test_construction_named(self):
+ sec_grp = SecurityGroup(tenant_id='foo', id='id', name='name')
+ self.assertEqual('name', sec_grp.name)
+ self.assertEqual('id', sec_grp.id)
+ self.assertEqual('foo', sec_grp.project_id)
+
+
+class SecurityGroupRuleDomainObjectTests(unittest.TestCase):
+ """
+ Tests the construction of the snaps.domain.test.SecurityGroupRule class
+ """
+
+ def test_construction_kwargs(self):
+ sec_grp_rule = SecurityGroupRule(
+ **{'id': 'id', 'security_group_id': 'grp_id', 'description': 'desc',
+ 'direction': 'dir', 'ethertype': 'eType',
+ 'port_range_min': '10.0.0.100', 'port_range_max': '10.0.0.200',
+ 'protocol': 'proto', 'remote_group_id': 'group_id',
+ 'remote_ip_prefix': 'ip_prefix'})
+ self.assertEqual('id', sec_grp_rule.id)
+ self.assertEqual('grp_id', sec_grp_rule.security_group_id)
+ self.assertEqual('desc', sec_grp_rule.description)
+ self.assertEqual('dir', sec_grp_rule.direction)
+ self.assertEqual('eType', sec_grp_rule.ethertype)
+ self.assertEqual('10.0.0.100', sec_grp_rule.port_range_min)
+ self.assertEqual('10.0.0.200', sec_grp_rule.port_range_max)
+ self.assertEqual('proto', sec_grp_rule.protocol)
+ self.assertEqual('group_id', sec_grp_rule.remote_group_id)
+ self.assertEqual('ip_prefix', sec_grp_rule.remote_ip_prefix)
+
+ def test_construction_named(self):
+ sec_grp_rule = SecurityGroupRule(
+ remote_ip_prefix='ip_prefix', remote_group_id='group_id',
+ protocol='proto', port_range_min='10.0.0.100',
+ port_range_max='10.0.0.200', ethertype='eType',
+ direction='dir', description='desc', security_group_id='grp_id',
+ id='id')
+ self.assertEqual('id', sec_grp_rule.id)
+ self.assertEqual('grp_id', sec_grp_rule.security_group_id)
+ self.assertEqual('desc', sec_grp_rule.description)
+ self.assertEqual('dir', sec_grp_rule.direction)
+ self.assertEqual('eType', sec_grp_rule.ethertype)
+ self.assertEqual('10.0.0.100', sec_grp_rule.port_range_min)
+ self.assertEqual('10.0.0.200', sec_grp_rule.port_range_max)
+ self.assertEqual('proto', sec_grp_rule.protocol)
+ self.assertEqual('group_id', sec_grp_rule.remote_group_id)
+ self.assertEqual('ip_prefix', sec_grp_rule.remote_ip_prefix)
def add_security_group(self, security_group):
"""
Adds a security group to this VM. Call will block until VM is active.
- :param security_group: the OpenStack security group object
+ :param security_group: the SNAPS SecurityGroup domain object
:return True if successful else False
"""
self.vm_active(block=True)
try:
nova_utils.add_security_group(self.__nova, self.get_vm_inst(),
- security_group['security_group'][
- 'name'])
+ security_group.name)
return True
except NotFound as e:
logger.warning('Security group not added - ' + str(e))
SecurityGroupRuleSettings object
:return: the newly instantiated SecurityGroupRuleSettings object
"""
- rule_dict = rule['security_group_rule']
- sec_grp_name = None
- if rule_dict['security_group_id']:
- sec_grp = neutron_utils.get_security_group_by_id(
- self.__neutron, rule_dict['security_group_id'])
- if sec_grp:
- sec_grp_name = sec_grp['security_group']['name']
+ sec_grp = neutron_utils.get_security_group_by_id(
+ self.__neutron, rule.security_group_id)
setting = SecurityGroupRuleSettings(
- description=rule_dict['description'],
- direction=rule_dict['direction'], ethertype=rule_dict['ethertype'],
- port_range_min=rule_dict['port_range_min'],
- port_range_max=rule_dict['port_range_max'],
- protocol=rule_dict['protocol'],
- remote_group_id=rule_dict['remote_group_id'],
- remote_ip_prefix=rule_dict['remote_ip_prefix'],
- sec_grp_name=sec_grp_name)
+ description=rule.description,
+ direction=rule.direction,
+ ethertype=rule.ethertype,
+ port_range_min=rule.port_range_min,
+ port_range_max=rule.port_range_max,
+ protocol=rule.protocol,
+ remote_group_id=rule.remote_group_id,
+ remote_ip_prefix=rule.remote_ip_prefix,
+ sec_grp_name=sec_grp.name)
return setting
def clean(self):
sec_grp = neutron_utils.get_security_group(
neutron, self.sec_grp_name)
if sec_grp:
- out['security_group_id'] = sec_grp['security_group']['id']
+ out['security_group_id'] = sec_grp.id
else:
raise Exception(
'Cannot locate security group with name - ' +
:param rule: the rule to evaluate
:return: T/F
"""
- rule_dict = rule['security_group_rule']
-
if self.description is not None:
- if rule_dict['description'] is not None and rule_dict[
- 'description'] != '':
+ if (rule.description is not None and
+ rule.description != ''):
return False
- elif self.description != rule_dict['description']:
- if rule_dict['description'] != '':
+ elif self.description != rule.description:
+ if rule.description != '':
return False
- if self.direction.name != rule_dict['direction']:
+ if self.direction.name != rule.direction:
return False
- if self.ethertype and rule_dict.get('ethertype'):
- if self.ethertype.name != rule_dict['ethertype']:
+ if self.ethertype and rule.ethertype:
+ if self.ethertype.name != rule.ethertype:
return False
- if self.port_range_min and rule_dict.get('port_range_min'):
- if self.port_range_min != rule_dict['port_range_min']:
+ if self.port_range_min and rule.port_range_min:
+ if self.port_range_min != rule.port_range_min:
return False
- if self.port_range_max and rule_dict.get('port_range_max'):
- if self.port_range_max != rule_dict['port_range_max']:
+ if self.port_range_max and rule.port_range_max:
+ if self.port_range_max != rule.port_range_max:
return False
- if self.protocol and rule_dict.get('protocol'):
- if self.protocol.name != rule_dict['protocol']:
+ if self.protocol and rule.protocol:
+ if self.protocol.name != rule.protocol:
return False
- if self.remote_group_id and rule_dict.get('remote_group_id'):
- if self.remote_group_id != rule_dict['remote_group_id']:
+ if self.remote_group_id and rule.remote_group_id:
+ if self.remote_group_id != rule.remote_group_id:
return False
- if self.remote_ip_prefix and rule_dict.get('remote_ip_prefix'):
- if self.remote_ip_prefix != rule_dict['remote_ip_prefix']:
+ if self.remote_ip_prefix and rule.remote_ip_prefix:
+ if self.remote_ip_prefix != rule.remote_ip_prefix:
return False
return True
def __eq__(self, other):
- return self.description == other.description and \
- self.direction == other.direction and \
- self.port_range_min == other.port_range_min and \
- self.port_range_max == other.port_range_max and \
- self.ethertype == other.ethertype and \
- self.protocol == other.protocol and \
- self.sec_grp_name == other.sec_grp_name and \
- self.remote_group_id == other.remote_group_id and \
- self.remote_ip_prefix == other.remote_ip_prefix
+ return (
+ self.description == other.description and
+ self.direction == other.direction and
+ self.port_range_min == other.port_range_min and
+ self.port_range_max == other.port_range_max and
+ self.ethertype == other.ethertype and
+ self.protocol == other.protocol and
+ self.sec_grp_name == other.sec_grp_name and
+ self.remote_group_id == other.remote_group_id and
+ self.remote_ip_prefix == other.remote_ip_prefix)
def __hash__(self):
return hash((self.sec_grp_name, self.description, self.direction,
self.assertIsNotNone(sec_grp)
self.sec_grp_creators.append(sec_grp_creator)
- if 'tenant_id' in sec_grp['security_group']:
- self.assertEqual(self.project_creator.get_project().id,
- sec_grp['security_group']['tenant_id'])
- elif 'project_id' in sec_grp['security_group']:
- self.assertEqual(self.project_creator.get_project().id,
- sec_grp['security_group']['project_id'])
- else:
- self.fail('Cannot locate the project or tenant ID')
+ self.assertEqual(self.project_creator.get_project().id,
+ sec_grp.project_id)
def test_create_project_sec_grp_two_users(self):
"""
self.assertIsNotNone(sec_grp)
self.sec_grp_creators.append(sec_grp_creator)
- if 'tenant_id' in sec_grp['security_group']:
- self.assertEqual(self.project_creator.get_project().id,
- sec_grp['security_group']['tenant_id'])
- elif 'project_id' in sec_grp['security_group']:
- self.assertEqual(self.project_creator.get_project().id,
- sec_grp['security_group']['project_id'])
- else:
- self.fail('Cannot locate the project or tenant ID')
+ self.assertEqual(self.project_creator.get_project().id,
+ sec_grp.project_id)
rules)
self.sec_grp_creator.remove_rule(
- rule_id=rules[0]['security_group_rule']['id'])
+ rule_id=rules[0].id)
rules_after_del = neutron_utils.get_rules_by_security_group(
self.neutron,
self.sec_grp_creator.get_security_group())
from neutronclient.common.exceptions import NotFound
from neutronclient.neutron.client import Client
+
+from snaps.domain.network import SecurityGroup, SecurityGroupRule
from snaps.domain.vm_inst import FloatingIp
from snaps.openstack.utils import keystone_utils
"""
logger.info('Creating security group with name - %s',
sec_grp_settings.name)
- return neutron.create_security_group(
+ os_group = neutron.create_security_group(
sec_grp_settings.dict_for_neutron(keystone))
+ return SecurityGroup(
+ id=os_group['security_group']['id'],
+ name=os_group['security_group']['name'],
+ project_id=os_group['security_group'].get(
+ 'project_id', os_group['security_group'].get('tenant_id')))
def delete_security_group(neutron, sec_grp):
"""
Deletes a security group object from OpenStack
:param neutron: the client
- :param sec_grp: the security group object to delete
+ :param sec_grp: the SNAPS SecurityGroup object to delete
"""
- logger.info('Deleting security group with name - %s',
- sec_grp['security_group']['name'])
- return neutron.delete_security_group(sec_grp['security_group']['id'])
+ logger.info('Deleting security group with name - %s', sec_grp.name)
+ neutron.delete_security_group(sec_grp.id)
def get_security_group(neutron, name):
groups = neutron.list_security_groups(**{'name': name})
for group in groups['security_groups']:
if group['name'] == name:
- return {'security_group': group}
+ return SecurityGroup(
+ id=group['id'], name=group['name'],
+ project_id=group.get('project_id', group.get('tenant_id')))
return None
groups = neutron.list_security_groups(**{'id': sec_grp_id})
for group in groups['security_groups']:
if group['id'] == sec_grp_id:
- return {'security_group': group}
+ return SecurityGroup(
+ id=group['id'], name=group['name'],
+ project_id=group.get('project_id', group.get('tenant_id')))
return None
"""
logger.info('Creating security group to security group - %s',
sec_grp_rule_settings.sec_grp_name)
- return neutron.create_security_group_rule(
+ os_rule = neutron.create_security_group_rule(
sec_grp_rule_settings.dict_for_neutron(neutron))
+ return SecurityGroupRule(**os_rule['security_group_rule'])
def delete_security_group_rule(neutron, sec_grp_rule):
"""
Deletes a security group object from OpenStack
:param neutron: the client
- :param sec_grp_rule: the security group rule object to delete
+ :param sec_grp_rule: the SNAPS SecurityGroupRule object to delete
"""
logger.info('Deleting security group rule with ID - %s',
- sec_grp_rule['security_group_rule']['id'])
- neutron.delete_security_group_rule(
- sec_grp_rule['security_group_rule']['id'])
+ sec_grp_rule.id)
+ neutron.delete_security_group_rule(sec_grp_rule.id)
def get_rules_by_security_group(neutron, sec_grp):
"""
Retrieves all of the rules for a given security group
:param neutron: the client
- :param sec_grp: the security group object
+ :param sec_grp: the SNAPS SecurityGroup object
"""
logger.info('Retrieving security group rules associate with the '
- 'security group - %s', sec_grp['security_group']['name'])
+ 'security group - %s', sec_grp.name)
out = list()
rules = neutron.list_security_group_rules(
- **{'security_group_id': sec_grp['security_group']['id']})
+ **{'security_group_id': sec_grp.id})
for rule in rules['security_group_rules']:
- if rule['security_group_id'] == sec_grp['security_group']['id']:
- out.append({'security_group_rule': rule})
+ if rule['security_group_id'] == sec_grp.id:
+ out.append(SecurityGroupRule(**rule))
return out
"""
Deletes a security group object from OpenStack
:param neutron: the client
- :param sec_grp: the security group object
+ :param sec_grp: the SNAPS SecurityGroup domain object
:param rule_id: the rule's ID
"""
rules = neutron.list_security_group_rules(
- **{'security_group_id': sec_grp['security_group']['id']})
+ **{'security_group_id': sec_grp.id})
for rule in rules['security_group_rules']:
if rule['id'] == rule_id:
- return {'security_group_rule': rule}
+ return SecurityGroupRule(**rule)
return None
"""
try:
os_kp = nova.keypairs.get(keypair_obj)
- return Keypair(name=os_kp.name, id=os_kp.id, public_key=os_kp.public_key)
+ return Keypair(name=os_kp.name, id=os_kp.id,
+ public_key=os_kp.public_key)
except:
return None
Removes a security group from an existing VM
:param nova: the nova client
:param vm: the OpenStack server object (VM) to alter
- :param security_group: the OpenStack security group object to add
+ :param security_group: the SNAPS SecurityGroup domain object to add
"""
- nova.servers.remove_security_group(
- str(vm.id), security_group['security_group']['name'])
+ nova.servers.remove_security_group(str(vm.id), security_group.name)
def add_floating_ip_to_server(nova, vm, floating_ip, ip_addr):
self.keystone,
sec_grp_settings)
- self.assertTrue(sec_grp_settings.name,
- security_group['security_group']['name'])
+ self.assertTrue(sec_grp_settings.name, security_group.name)
sec_grp_get = neutron_utils.get_security_group(self.neutron,
sec_grp_settings.name)
self.assertIsNotNone(sec_grp_get)
self.assertTrue(validation_utils.objects_equivalent(
- security_group['security_group'], sec_grp_get['security_group']))
+ security_group, sec_grp_get))
neutron_utils.delete_security_group(self.neutron, security_group)
sec_grp_get = neutron_utils.get_security_group(self.neutron,
neutron_utils.create_security_group(self.neutron, self.keystone,
sec_grp_settings))
- self.assertTrue(sec_grp_settings.name,
- self.security_groups[0]['security_group']['name'])
- self.assertTrue(sec_grp_settings.description,
- self.security_groups[0]['security_group'][
- 'description'])
+ self.assertTrue(sec_grp_settings.name, self.security_groups[0].name)
+ self.assertEqual(sec_grp_settings.name, self.security_groups[0].name)
sec_grp_get = neutron_utils.get_security_group(self.neutron,
sec_grp_settings.name)
self.assertIsNotNone(sec_grp_get)
- self.assertTrue(validation_utils.objects_equivalent(
- self.security_groups[0]['security_group'],
- sec_grp_get['security_group']))
+ self.assertEqual(self.security_groups[0], sec_grp_get)
def test_create_sec_grp_one_rule(self):
"""
security_group)
self.assertTrue(
- validation_utils.objects_equivalent(self.security_group_rules,
- rules))
+ validation_utils.objects_equivalent(
+ self.security_group_rules, rules))
- self.assertTrue(sec_grp_settings.name,
- security_group['security_group']['name'])
- self.assertTrue(sec_grp_settings.description,
- security_group['security_group']['description'])
+ self.assertTrue(sec_grp_settings.name, security_group.name)
sec_grp_get = neutron_utils.get_security_group(self.neutron,
sec_grp_settings.name)
self.assertIsNotNone(sec_grp_get)
- self.assertTrue(validation_utils.objects_equivalent(
- security_group['security_group'], sec_grp_get['security_group']))
+ self.assertEqual(security_group, sec_grp_get)
def test_get_sec_grp_by_id(self):
"""
description='hello group')))
sec_grp_1b = neutron_utils.get_security_group_by_id(
- self.neutron, self.security_groups[0]['security_group']['id'])
+ self.neutron, self.security_groups[0].id)
sec_grp_2b = neutron_utils.get_security_group_by_id(
- self.neutron, self.security_groups[1]['security_group']['id'])
+ self.neutron, self.security_groups[1].id)
- self.assertEqual(self.security_groups[0]['security_group']['id'],
- sec_grp_1b['security_group']['id'])
- self.assertEqual(self.security_groups[1]['security_group']['id'],
- sec_grp_2b['security_group']['id'])
+ self.assertEqual(self.security_groups[0].id, sec_grp_1b.id)
+ self.assertEqual(self.security_groups[1].id, sec_grp_2b.id)
class NeutronUtilsFloatingIpTests(OSComponentTestCase):
from snaps.domain.test.flavor_tests import FlavorDomainObjectTests
from snaps.domain.test.image_tests import ImageDomainObjectTests
from snaps.domain.test.keypair_tests import KeypairDomainObjectTests
+from snaps.domain.test.network_tests import (
+ SecurityGroupDomainObjectTests, SecurityGroupRuleDomainObjectTests)
from snaps.domain.test.stack_tests import StackDomainObjectTests
from snaps.domain.test.user_tests import UserDomainObjectTests
-from snaps.domain.test.vm_inst_tests import (VmInstDomainObjectTests,
- FloatingIpDomainObjectTests)
+from snaps.domain.test.vm_inst_tests import (
+ VmInstDomainObjectTests, FloatingIpDomainObjectTests)
from snaps.openstack.tests.conf.os_credentials_tests import (
ProxySettingsUnitTests, OSCredsUnitTests)
from snaps.openstack.tests.create_flavor_tests import (
OSCredsUnitTests))
suite.addTest(unittest.TestLoader().loadTestsFromTestCase(
SecurityGroupSettingsUnitTests))
+ suite.addTest(unittest.TestLoader().loadTestsFromTestCase(
+ SecurityGroupDomainObjectTests))
+ suite.addTest(unittest.TestLoader().loadTestsFromTestCase(
+ SecurityGroupRuleDomainObjectTests))
suite.addTest(unittest.TestLoader().loadTestsFromTestCase(
ImageSettingsUnitTests))
suite.addTest(unittest.TestLoader().loadTestsFromTestCase(
Code Review / snaps.git / commitdiff