openstack
/
ec2-api
Code Issues Proposed changes

Added repair default group feature. 

Added unit-test for nova group with "vpc-..." name

Change-Id: I2007eb532edf4df6d55e27e63c9429e59841b8fe
This commit is contained in:
Alexandre Levine 2015-04-03 17:07:11 +04:00
parent ce9a4a396f
commit e6d37da955
7 changed files with 111 additions and 25 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
ec2api/api/common.py
View File

@ -335,8 +335,11 @@ class UniversalDescriber(object):
selective_describe = ids is not None or names is not None
self.ids = set(ids or [])
self.names = set(names or [])
self.items = self.get_db_items()
# NOTE(Alex): OS items are retrieved here first to let specific
# describer a chance to recreate some default object and refresh
# the db items before their retrieval.
self.os_items = self.get_os_items()
self.items = self.get_db_items()
formatted_items = []
self.items_dict = {i['os_id']: i for i in (self.items or [])}

22
ec2api/api/security_group.py
View File

@ -59,7 +59,7 @@ def get_security_group_engine():
def create_security_group(context, group_name, group_description,
vpc_id=None):
nova = clients.nova(context)
if vpc_id:
if vpc_id and group_name != vpc_id:
security_groups = describe_security_groups(
context,
filter=[{'name': 'vpc-id',
@ -136,14 +136,32 @@ class SecurityGroupDescriber(common.TaggableItemsDescriber):
def get_os_items(self):
if self.all_db_items is None:
self.all_db_items = ec2utils.get_db_items(self.context, 'sg', None)
self.all_db_items = db_api.get_items(self.context, 'sg')
os_groups = security_group_engine.get_os_groups(self.context)
if self.check_and_repair_default_groups(os_groups, self.all_db_items):
self.all_db_items = db_api.get_items(self.context, 'sg')
os_groups = security_group_engine.get_os_groups(self.context)
for os_group in os_groups:
os_group['name'] = _translate_group_name(self.context,
os_group,
self.all_db_items)
return os_groups
def check_and_repair_default_groups(self, os_groups, db_groups):
vpcs = ec2utils.get_db_items(self.context, 'vpc', None)
os_groups_dict = dict((g['name'], g['id']) for g in os_groups)
db_groups_dict = dict((g['os_id'], g['vpc_id']) for g in db_groups)
had_to_repair = False
for vpc in vpcs:
os_group = os_groups_dict.get(vpc['id'])
if os_group:
db_group = db_groups_dict.get(os_group)
if db_group and db_group == vpc['id']:
continue
had_to_repair = True
_create_default_security_group(self.context, vpc)
return had_to_repair
def describe_security_groups(context, group_name=None, group_id=None,
filter=None):

2
ec2api/tests/unit/base.py
View File

@ -155,7 +155,7 @@ class ApiTestCase(test_base.BaseTestCase):
resp_items = resp[resultset_key]
resultset_key = [resultset_key]
for resp_item in resp_items:
if resp_item[id_key] == sample_item_id:
if resp_item.get(id_key) == sample_item_id:
self.assertIn('tagSet', resp_item)
self.assertThat(resp_item['tagSet'],
matchers.ListMatches(ec2_tags))

18
ec2api/tests/unit/fakes.py
View File

@ -141,6 +141,7 @@ ID_EC2_SECURITY_GROUP_1 = random_ec2_id('sg')
ID_EC2_SECURITY_GROUP_2 = random_ec2_id('sg')
ID_OS_SECURITY_GROUP_1 = random_os_id()
ID_OS_SECURITY_GROUP_2 = random_os_id()
ID_OS_SECURITY_GROUP_3 = random_os_id()
NAME_DEFAULT_OS_SECURITY_GROUP = 'default'
NAME_OTHER_OS_SECURITY_GROUP = 'other'
@ -368,7 +369,8 @@ EC2_NETWORK_INTERFACE_1 = {
'sourceDestCheck': True,
'ownerId': ID_OS_PROJECT,
'requesterManaged': False,
'groupSet': [],
'groupSet': [{'groupName': NAME_DEFAULT_OS_SECURITY_GROUP,
'groupId': ID_EC2_SECURITY_GROUP_1}],
'tagSet': [],
}
EC2_NETWORK_INTERFACE_2 = {
@ -427,7 +429,7 @@ OS_PORT_1 = {'id': ID_OS_PORT_1,
'subnet_id': ID_OS_SUBNET_1}],
'device_id': None,
'device_owner': '',
'security_groups': []}
'security_groups': [ID_OS_SECURITY_GROUP_1]}
OS_PORT_2 = {'id': ID_OS_PORT_2,
'network_id': ID_OS_SUBNET_2,
'name': ID_EC2_NETWORK_INTERFACE_2,
@ -871,6 +873,12 @@ OS_SECURITY_GROUP_2 = {
'description': 'Group description',
'tenant_id': ID_OS_PROJECT
}
OS_SECURITY_GROUP_3 = {
'id': ID_OS_SECURITY_GROUP_3,
'name': ID_EC2_VPC_2,
'description': 'Group description',
'tenant_id': ID_OS_PROJECT
}
NOVA_SECURITY_GROUP_RULE_1 = {
'id': random_os_id(),
'from_port': 10,
@ -948,6 +956,12 @@ EC2_SECURITY_GROUP_2 = {
'ownerId': ID_OS_PROJECT,
'groupId': ID_EC2_SECURITY_GROUP_2
}
EC2_SECURITY_GROUP_3 = {
'groupDescription': 'Group description',
'ipPermissions': None,
'groupName': ID_EC2_VPC_2,
'ownerId': ID_OS_PROJECT,
}
EC2_NOVA_SECURITY_GROUP_1 = {
'groupDescription': 'Group description',
'ipPermissions': None,

22
ec2api/tests/unit/test_network_interface.py
View File

@ -28,10 +28,13 @@ from ec2api.tests.unit import tools
class NetworkInterfaceTestCase(base.ApiTestCase):
def test_create_network_interface(self):
self.set_mock_db_items(fakes.DB_SUBNET_1, fakes.DB_VPC_1)
self.set_mock_db_items(fakes.DB_SUBNET_1, fakes.DB_VPC_1,
fakes.DB_SECURITY_GROUP_1)
self.db_api.add_item.return_value = fakes.DB_NETWORK_INTERFACE_1
self.neutron.show_subnet.return_value = {'subnet': fakes.OS_SUBNET_1}
self.neutron.create_port.return_value = {'port': fakes.OS_PORT_1}
self.neutron.list_security_groups.return_value = (
{'security_groups': [copy.deepcopy(fakes.OS_SECURITY_GROUP_1)]})
def check_response(resp, auto_ips=False):
self.assertThat(fakes.EC2_NETWORK_INTERFACE_1,
@ -45,20 +48,23 @@ class NetworkInterfaceTestCase(base.ApiTestCase):
{'network_id': fakes.ID_OS_NETWORK_1,
'fixed_ips':
[{'subnet_id': fakes.ID_OS_SUBNET_1}],
'security_groups': []}})
'security_groups': [fakes.ID_OS_SECURITY_GROUP_1]}})
else:
self.neutron.create_port.assert_called_once_with(
{'port':
{'network_id': fakes.ID_OS_NETWORK_1,
'fixed_ips':
[{'ip_address': fakes.IP_NETWORK_INTERFACE_1}],
'security_groups': []}})
'security_groups': [fakes.ID_OS_SECURITY_GROUP_1]}})
self.neutron.update_port.assert_called_once_with(
fakes.ID_OS_PORT_1,
{'port': {'name':
fakes.ID_EC2_NETWORK_INTERFACE_1}})
self.neutron.reset_mock()
self.db_api.reset_mock()
self.neutron.list_security_groups.return_value = (
{'security_groups': [
copy.deepcopy(fakes.OS_SECURITY_GROUP_1)]})
resp = self.execute(
'CreateNetworkInterface',
@ -103,7 +109,7 @@ class NetworkInterfaceTestCase(base.ApiTestCase):
self.neutron.show_subnet.return_value = {'subnet': fakes.OS_SUBNET_2}
self.neutron.create_port.return_value = {'port': fakes.OS_PORT_2}
self.neutron.list_security_groups.return_value = (
{'security_groups': [fakes.OS_SECURITY_GROUP_1]})
{'security_groups': [copy.deepcopy(fakes.OS_SECURITY_GROUP_1)]})
created_ec2_network_interface = tools.patch_dict(
fakes.EC2_NETWORK_INTERFACE_2,
{'privateIpAddressesSet': [
@ -129,6 +135,9 @@ class NetworkInterfaceTestCase(base.ApiTestCase):
fakes.ID_EC2_NETWORK_INTERFACE_2}})
self.neutron.reset_mock()
self.db_api.reset_mock()
self.neutron.list_security_groups.return_value = (
{'security_groups': [
copy.deepcopy(fakes.OS_SECURITY_GROUP_1)]})
resp = self.execute(
'CreateNetworkInterface',
@ -342,15 +351,14 @@ class NetworkInterfaceTestCase(base.ApiTestCase):
fakes.DB_NETWORK_INTERFACE_1, fakes.DB_NETWORK_INTERFACE_2,
fakes.DB_ADDRESS_1, fakes.DB_ADDRESS_2,
fakes.DB_INSTANCE_1, fakes.DB_INSTANCE_2,
fakes.DB_SECURITY_GROUP_1, fakes.DB_SECURITY_GROUP_2)
fakes.DB_SECURITY_GROUP_1)
self.neutron.list_ports.return_value = (
{'ports': [fakes.OS_PORT_1, fakes.OS_PORT_2]})
self.neutron.list_floatingips.return_value = (
{'floatingips': [fakes.OS_FLOATING_IP_1,
fakes.OS_FLOATING_IP_2]})
self.neutron.list_security_groups.return_value = (
{'security_groups': [fakes.OS_SECURITY_GROUP_1,
fakes.OS_SECURITY_GROUP_2]})
{'security_groups': [copy.deepcopy(fakes.OS_SECURITY_GROUP_1)]})
resp = self.execute('DescribeNetworkInterfaces', {})
self.assertThat(resp['networkInterfaceSet'],

53
ec2api/tests/unit/test_security_group.py
View File

@ -13,6 +13,8 @@
# limitations under the License.
import copy
import mock
from neutronclient.common import exceptions as neutron_exception
from novaclient import exceptions as nova_exception
@ -26,13 +28,24 @@ from ec2api.tests.unit import tools
class SecurityGroupTestCase(base.ApiTestCase):
def setUp(self):
super(SecurityGroupTestCase, self).setUp()
self.addCleanup(self._reset_engine)
def _reset_engine(self):
security_group.security_group_engine = (
security_group.SecurityGroupEngineNeutron())
def test_create_security_group(self):
security_group.security_group_engine = (
security_group.SecurityGroupEngineNeutron())
self.set_mock_db_items(fakes.DB_VPC_1)
self.db_api.add_item.return_value = fakes.DB_SECURITY_GROUP_1
self.set_mock_db_items(fakes.DB_VPC_1,
fakes.DB_SECURITY_GROUP_1)
self.neutron.list_security_groups.return_value = (
{'security_groups': [copy.deepcopy(fakes.OS_SECURITY_GROUP_1)]})
self.db_api.add_item.return_value = fakes.DB_SECURITY_GROUP_2
self.nova.security_groups.create.return_value = (
fakes.NovaSecurityGroup(fakes.OS_SECURITY_GROUP_1))
fakes.NovaSecurityGroup(fakes.OS_SECURITY_GROUP_2))
resp = self.execute(
'CreateSecurityGroup',
@ -47,10 +60,10 @@ class SecurityGroupTestCase(base.ApiTestCase):
{'VpcId': fakes.ID_EC2_VPC_1,
'GroupName': 'groupname',
'GroupDescription': 'Group description'})
self.assertEqual(fakes.ID_EC2_SECURITY_GROUP_1, resp['groupId'])
self.assertEqual(fakes.ID_EC2_SECURITY_GROUP_2, resp['groupId'])
self.db_api.add_item.assert_called_once_with(
mock.ANY, 'sg',
tools.purge_dict(fakes.DB_SECURITY_GROUP_1, ('id',)))
tools.purge_dict(fakes.DB_SECURITY_GROUP_2, ('id',)))
self.nova.security_groups.create.assert_called_once_with(
'groupname', 'Group description')
@ -163,7 +176,7 @@ class SecurityGroupTestCase(base.ApiTestCase):
security_group.security_group_engine = (
security_group.SecurityGroupEngineNova())
self.nova.security_groups.list.return_value = (
[fakes.NovaSecurityGroup(fakes.OS_SECURITY_GROUP_1),
[fakes.NovaSecurityGroup(copy.deepcopy(fakes.OS_SECURITY_GROUP_1)),
fakes.NovaSecurityGroup(fakes.OS_SECURITY_GROUP_2)])
resp = self.execute(
'DeleteSecurityGroup',
@ -227,14 +240,16 @@ class SecurityGroupTestCase(base.ApiTestCase):
self.set_mock_db_items(fakes.DB_SECURITY_GROUP_1,
fakes.DB_SECURITY_GROUP_2)
self.neutron.list_security_groups.return_value = (
{'security_groups': [fakes.OS_SECURITY_GROUP_1,
fakes.OS_SECURITY_GROUP_2]})
{'security_groups': [copy.deepcopy(fakes.OS_SECURITY_GROUP_1),
fakes.OS_SECURITY_GROUP_2,
fakes.OS_SECURITY_GROUP_3]})
resp = self.execute('DescribeSecurityGroups', {})
self.assertThat(resp['securityGroupInfo'],
matchers.ListMatches(
[fakes.EC2_SECURITY_GROUP_1,
fakes.EC2_SECURITY_GROUP_2],
fakes.EC2_SECURITY_GROUP_2,
fakes.EC2_SECURITY_GROUP_3],
orderless_lists=True))
resp = self.execute('DescribeSecurityGroups',
@ -288,6 +303,26 @@ class SecurityGroupTestCase(base.ApiTestCase):
fakes.EC2_NOVA_SECURITY_GROUP_2],
orderless_lists=True))
def test_repair_default_security_group(self):
security_group.security_group_engine = (
security_group.SecurityGroupEngineNeutron())
self.db_api.add_item.return_value = fakes.DB_SECURITY_GROUP_1
self.nova.security_groups.create.return_value = (
fakes.NovaSecurityGroup(fakes.OS_SECURITY_GROUP_1))
self.set_mock_db_items(fakes.DB_VPC_1,
fakes.DB_SECURITY_GROUP_1,
fakes.DB_SECURITY_GROUP_2)
self.neutron.list_security_groups.return_value = (
{'security_groups': [fakes.OS_SECURITY_GROUP_2,
fakes.OS_SECURITY_GROUP_3]})
resp = self.execute('DescribeSecurityGroups', {})
self.db_api.add_item.assert_called_once_with(
mock.ANY, 'sg',
tools.purge_dict(fakes.DB_SECURITY_GROUP_1, ('id',)))
self.nova.security_groups.create.assert_called_once_with(
fakes.ID_EC2_VPC_1, 'Default VPC security group')
def test_authorize_security_group_invalid(self):
security_group.security_group_engine = (
security_group.SecurityGroupEngineNeutron())

14
ec2api/tests/unit/test_vpc.py
View File

@ -13,6 +13,8 @@
# limitations under the License.
import copy
import mock
from neutronclient.common import exceptions as neutron_exception
@ -150,17 +152,23 @@ class VpcTestCase(base.ApiTestCase):
self.neutron.reset_mock()
self.db_api.reset_mock()
self.set_mock_db_items(fakes.DB_IGW_1, fakes.DB_VPC_1)
self.neutron.list_security_groups.return_value = (
{'security_groups': [copy.deepcopy(fakes.OS_SECURITY_GROUP_1)]})
self.set_mock_db_items(fakes.DB_SECURITY_GROUP_1,
fakes.DB_IGW_1, fakes.DB_VPC_1, )
do_check()
self.set_mock_db_items(fakes.DB_ROUTE_TABLE_1, fakes.DB_ROUTE_TABLE_2,
self.neutron.list_security_groups.return_value = (
{'security_groups': [copy.deepcopy(fakes.OS_SECURITY_GROUP_1)]})
self.set_mock_db_items(fakes.DB_SECURITY_GROUP_1,
fakes.DB_ROUTE_TABLE_1, fakes.DB_ROUTE_TABLE_2,
fakes.DB_VPC_1)
do_check()
self.set_mock_db_items(fakes.DB_SECURITY_GROUP_1,
fakes.DB_SECURITY_GROUP_2, fakes.DB_VPC_1)
self.neutron.list_security_groups.return_value = (
{'security_groups': [fakes.OS_SECURITY_GROUP_1,
{'security_groups': [copy.deepcopy(fakes.OS_SECURITY_GROUP_1),
fakes.OS_SECURITY_GROUP_2]})
do_check()