Merge "[AWS] Added Neutron router plugin unit test cases"
This commit is contained in:
commit
4fdcdd95a8
|
@ -12,11 +12,11 @@ under the License.
|
|||
"""
|
||||
|
||||
import neutron_lib
|
||||
from neutron_lib import constants as const
|
||||
|
||||
from distutils.version import LooseVersion
|
||||
from neutron.common.aws_utils import AwsException
|
||||
from neutron.common.aws_utils import AwsUtils
|
||||
from neutron.common import constants as n_const
|
||||
from neutron.db import common_db_mixin
|
||||
from neutron.db import extraroute_db
|
||||
from neutron.db import l3_db
|
||||
|
@ -83,6 +83,7 @@ class AwsRouterPlugin(
|
|||
# FLOATING IP FEATURES
|
||||
|
||||
def create_floatingip(self, context, floatingip):
|
||||
public_ip_allocated = None
|
||||
try:
|
||||
response = self.aws_utils.allocate_elastic_ip()
|
||||
public_ip_allocated = response['PublicIp']
|
||||
|
@ -107,7 +108,7 @@ class AwsRouterPlugin(
|
|||
try:
|
||||
res = super(AwsRouterPlugin, self).create_floatingip(
|
||||
context, floatingip,
|
||||
initial_status=n_const.FLOATINGIP_STATUS_DOWN)
|
||||
initial_status=const.FLOATINGIP_STATUS_DOWN)
|
||||
except Exception as e:
|
||||
LOG.error("Error when adding floating ip in openstack. "
|
||||
"Deleting Elastic IP: %s" % public_ip_allocated)
|
||||
|
|
|
@ -0,0 +1,291 @@
|
|||
"""
|
||||
Copyright (c) 2017 Platform9 Systems Inc.
|
||||
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 expressed or implied. See the
|
||||
License for the specific language governing permissions and limitations
|
||||
under the License.
|
||||
"""
|
||||
|
||||
import mock
|
||||
|
||||
from moto import mock_ec2
|
||||
from neutron.common.aws_utils import AwsException
|
||||
from neutron.common.aws_utils import AwsUtils
|
||||
from neutron.common.aws_utils import cfg
|
||||
from neutron.common import exceptions
|
||||
from neutron.services.l3_router.aws_router_plugin import AwsRouterPlugin
|
||||
from neutron.tests import base
|
||||
from neutron.tests.unit.extensions import test_securitygroup as test_sg
|
||||
from neutron_lib import constants as const
|
||||
|
||||
L3_NAT_DBONLY_MIXIN = 'neutron.db.l3_db.L3_NAT_dbonly_mixin'
|
||||
L3_NAT_WITH_DVR_DB_MIXIN = 'neutron.db.l3_dvr_db.L3_NAT_with_dvr_db_mixin'
|
||||
AWS_ROUTER = 'neutron.services.l3_router.aws_router_plugin.AwsRouterPlugin'
|
||||
|
||||
|
||||
class AWSRouterPluginTests(test_sg.SecurityGroupsTestCase, base.BaseTestCase):
|
||||
@mock_ec2
|
||||
def setUp(self):
|
||||
super(AWSRouterPluginTests, self).setUp()
|
||||
cfg.CONF.AWS.secret_key = 'aws_access_key'
|
||||
cfg.CONF.AWS.access_key = 'aws_secret_key'
|
||||
cfg.CONF.AWS.region_name = 'us-east-1'
|
||||
self._driver = AwsRouterPlugin()
|
||||
self.context = self._create_fake_context()
|
||||
|
||||
def _create_fake_context(self):
|
||||
context = mock.Mock()
|
||||
context.current = {}
|
||||
context.current['id'] = "fake_id_1234"
|
||||
context.current['cidr'] = "192.168.1.0/24"
|
||||
context.current['network_id'] = "fake_network_id_1234"
|
||||
return context
|
||||
|
||||
def _get_fake_tags(self):
|
||||
tags = []
|
||||
tags.append({'Value': 'fake_name', 'Key': 'Name'})
|
||||
tags.append({'Value': 'fake_id', 'Key': 'openstack_network_id'})
|
||||
tags.append({'Value': 'fake_tenant_id', 'Key': 'openstack_tenant_id'})
|
||||
return tags
|
||||
|
||||
def _create_router(self, mock_create):
|
||||
mock_create.return_value = {'id': 'fake_id'}
|
||||
router = {'router': {'name': 'fake_name'}}
|
||||
response = self._driver.create_router(self.context, router)
|
||||
mock_create.assert_called_once_with(self.context, router)
|
||||
return response
|
||||
|
||||
@mock_ec2
|
||||
@mock.patch(L3_NAT_WITH_DVR_DB_MIXIN + '.create_floatingip')
|
||||
@mock.patch(AWS_ROUTER + '._associate_floatingip_to_port')
|
||||
def test_create_floatingip_with_port(self, mock_assoc, mock_create):
|
||||
floatingip = {'floatingip': {'port_id': 'fake_port_id'}}
|
||||
mock_assoc.return_value = None
|
||||
mock_create.return_value = None
|
||||
self.assertIsNone(self._driver.create_floatingip(self.context,
|
||||
floatingip))
|
||||
self.assertTrue(mock_assoc.called)
|
||||
mock_create.assert_called_once_with(
|
||||
self.context, floatingip,
|
||||
initial_status=const.FLOATINGIP_STATUS_DOWN)
|
||||
|
||||
@mock_ec2
|
||||
@mock.patch(L3_NAT_WITH_DVR_DB_MIXIN + '.create_floatingip')
|
||||
def test_create_floatingip_without_port(self, mock_create):
|
||||
floatingip = {'floatingip': {}}
|
||||
mock_create.return_value = None
|
||||
self.assertIsNone(self._driver.create_floatingip(self.context,
|
||||
floatingip))
|
||||
mock_create.assert_called_once_with(
|
||||
self.context, floatingip,
|
||||
initial_status=const.FLOATINGIP_STATUS_DOWN)
|
||||
|
||||
@mock_ec2
|
||||
@mock.patch('neutron.db.db_base_plugin_v2.NeutronDbPluginV2.get_port')
|
||||
def test_create_floatingip_with_failure_in_associating(self, mock_get):
|
||||
floatingip = {'floatingip': {'port_id': 'fake_port_id'}}
|
||||
port = {'fixed_ips': []}
|
||||
mock_get.return_value = port
|
||||
self.assertRaises(AwsException, self._driver.create_floatingip,
|
||||
self.context, floatingip)
|
||||
|
||||
@mock_ec2
|
||||
@mock.patch(L3_NAT_WITH_DVR_DB_MIXIN + '.create_floatingip')
|
||||
@mock.patch(AWS_ROUTER + '._associate_floatingip_to_port')
|
||||
def test_create_floatingip_with_failure_in_creating(
|
||||
self, mock_assoc, mock_create):
|
||||
floatingip = {'floatingip': {'port_id': 'fake_port_id'}}
|
||||
mock_create.side_effect = exceptions.PhysicalNetworkNameError()
|
||||
mock_assoc.return_value = None
|
||||
self.assertRaises(exceptions.PhysicalNetworkNameError,
|
||||
self._driver.create_floatingip, self.context,
|
||||
floatingip)
|
||||
|
||||
@mock_ec2
|
||||
@mock.patch(L3_NAT_WITH_DVR_DB_MIXIN + '.update_floatingip')
|
||||
@mock.patch(L3_NAT_DBONLY_MIXIN + '.get_floatingip')
|
||||
def test_update_floatingip_without_port(self, mock_get, mock_update):
|
||||
fake_id = 'fake_id'
|
||||
floatingip = {'floatingip': {}}
|
||||
mock_get.return_value = {}
|
||||
mock_update.return_value = None
|
||||
self.assertIsNone(self._driver.update_floatingip(self.context, fake_id,
|
||||
floatingip))
|
||||
self.assertTrue(mock_get.called)
|
||||
mock_get.assert_called_once_with(self.context, fake_id)
|
||||
mock_update.assert_called_once_with(self.context, fake_id, floatingip)
|
||||
|
||||
@mock_ec2
|
||||
@mock.patch(L3_NAT_WITH_DVR_DB_MIXIN + '.delete_floatingip')
|
||||
@mock.patch(L3_NAT_DBONLY_MIXIN + '.get_floatingip')
|
||||
def test_delete_floatingip_success(self, mock_get, mock_delete):
|
||||
fake_id = 'fake_id'
|
||||
mock_get.return_value = {'floating_ip_address': '192.169.10.1'}
|
||||
mock_delete.return_value = None
|
||||
self.assertIsNone(self._driver.delete_floatingip(self.context,
|
||||
fake_id))
|
||||
self.assertTrue(mock_get.called)
|
||||
mock_get.assert_called_once_with(self.context, fake_id)
|
||||
mock_delete.assert_called_once_with(self.context, fake_id)
|
||||
|
||||
@mock_ec2
|
||||
@mock.patch(L3_NAT_WITH_DVR_DB_MIXIN + '.delete_floatingip')
|
||||
@mock.patch(L3_NAT_DBONLY_MIXIN + '.get_floatingip')
|
||||
def test_delete_floatingip_failure(self, mock_get, mock_delete):
|
||||
fake_id = 'fake_id'
|
||||
mock_get.return_value = {'floating_ip_address': '192.169.10.1'}
|
||||
mock_delete.side_effect = exceptions.PhysicalNetworkNameError()
|
||||
self.assertRaises(exceptions.PhysicalNetworkNameError,
|
||||
self._driver.delete_floatingip,
|
||||
self.context, fake_id)
|
||||
self.assertTrue(mock_get.called)
|
||||
mock_get.assert_called_once_with(self.context, fake_id)
|
||||
|
||||
@mock_ec2
|
||||
@mock.patch(L3_NAT_WITH_DVR_DB_MIXIN + '.delete_floatingip')
|
||||
@mock.patch(L3_NAT_DBONLY_MIXIN + '.get_floatingip')
|
||||
def test_delete_floatingip_aws_failure(self, mock_get, mock_delete):
|
||||
fake_id = 'fake_id'
|
||||
mock_get.return_value = {'floating_ip_address': None}
|
||||
mock_delete.side_effect = {}
|
||||
self.assertRaises(AwsException, self._driver.delete_floatingip,
|
||||
self.context, fake_id)
|
||||
self.assertTrue(mock_get.called)
|
||||
mock_get.assert_called_once_with(self.context, fake_id)
|
||||
|
||||
@mock_ec2
|
||||
@mock.patch(L3_NAT_WITH_DVR_DB_MIXIN + '.update_floatingip')
|
||||
@mock.patch(L3_NAT_DBONLY_MIXIN + '.get_floatingip')
|
||||
def test_update_floatingip_with_port(self, mock_get, mock_update):
|
||||
fake_id = 'fake_id'
|
||||
floatingip = {'floatingip': {'port_id': None}}
|
||||
mock_get.return_value = {'floating_ip_address': '192.169.10.1'}
|
||||
mock_update.return_value = None
|
||||
self.assertIsNone(self._driver.update_floatingip(self.context, fake_id,
|
||||
floatingip))
|
||||
self.assertTrue(mock_get.called)
|
||||
mock_get.assert_called_once_with(self.context, fake_id)
|
||||
mock_update.assert_called_once_with(self.context, fake_id, floatingip)
|
||||
|
||||
@mock_ec2
|
||||
@mock.patch(L3_NAT_WITH_DVR_DB_MIXIN + '.update_floatingip')
|
||||
@mock.patch(L3_NAT_DBONLY_MIXIN + '.get_floatingip')
|
||||
def test_update_floatingip_failure_without_port(self, mock_get,
|
||||
mock_update):
|
||||
fake_id = 'fake_id'
|
||||
floatingip = {'floatingip': {}}
|
||||
mock_get.return_value = {'floating_ip_address': '192.169.10.1'}
|
||||
mock_update.side_effect = exceptions.PhysicalNetworkNameError()
|
||||
self.assertRaises(exceptions.PhysicalNetworkNameError,
|
||||
self._driver.update_floatingip, self.context,
|
||||
fake_id, floatingip)
|
||||
self.assertTrue(mock_get.called)
|
||||
mock_get.assert_called_once_with(self.context, fake_id)
|
||||
|
||||
@mock_ec2
|
||||
@mock.patch(L3_NAT_WITH_DVR_DB_MIXIN + '.update_floatingip')
|
||||
@mock.patch(L3_NAT_DBONLY_MIXIN + '.get_floatingip')
|
||||
def test_update_floatingip_failure_with_port(self, mock_get, mock_update):
|
||||
fake_id = 'fake_id'
|
||||
floatingip = {'floatingip': {'port': 'fake_port_id'}}
|
||||
mock_get.return_value = {'floating_ip_address': '192.169.10.1'}
|
||||
mock_update.side_effect = exceptions.PhysicalNetworkNameError()
|
||||
self.assertRaises(exceptions.PhysicalNetworkNameError,
|
||||
self._driver.update_floatingip, self.context,
|
||||
fake_id, floatingip)
|
||||
self.assertTrue(mock_get.called)
|
||||
mock_get.assert_called_once_with(self.context, fake_id)
|
||||
|
||||
@mock_ec2
|
||||
@mock.patch(L3_NAT_DBONLY_MIXIN + '.create_router')
|
||||
def test_create_router_success(self, mock_create):
|
||||
response = self._create_router(mock_create)
|
||||
self.assertIsInstance(response, dict)
|
||||
|
||||
@mock_ec2
|
||||
@mock.patch(L3_NAT_DBONLY_MIXIN + '.create_router')
|
||||
def test_create_router_failure(self, mock_create):
|
||||
mock_create.side_effect = exceptions.PhysicalNetworkNameError()
|
||||
router = {'router': {'name': 'fake_name'}}
|
||||
self.assertRaises(exceptions.PhysicalNetworkNameError,
|
||||
self._driver.create_router, self.context, router)
|
||||
|
||||
@mock_ec2
|
||||
@mock.patch(L3_NAT_DBONLY_MIXIN + '.delete_router')
|
||||
@mock.patch(L3_NAT_DBONLY_MIXIN + '.create_router')
|
||||
def test_delete_router_success(self, mock_create, mock_delete):
|
||||
mock_delete.return_value = None
|
||||
response = self._create_router(mock_create)
|
||||
self.assertIsNone(self._driver.delete_router(self.context,
|
||||
response['id']))
|
||||
self.assertTrue(mock_delete.called)
|
||||
mock_delete.assert_called_once_with(self.context, response['id'])
|
||||
|
||||
@mock_ec2
|
||||
@mock.patch(L3_NAT_DBONLY_MIXIN + '.delete_router')
|
||||
@mock.patch(L3_NAT_DBONLY_MIXIN + '.create_router')
|
||||
def test_delete_router_failure(self, mock_create, mock_delete):
|
||||
mock_delete.side_effect = exceptions.PhysicalNetworkNameError()
|
||||
response = self._create_router(mock_create)
|
||||
self.assertRaises(
|
||||
exceptions.PhysicalNetworkNameError, self._driver.delete_router,
|
||||
self.context, response['id'])
|
||||
|
||||
@mock_ec2
|
||||
@mock.patch(L3_NAT_DBONLY_MIXIN + '.update_router')
|
||||
@mock.patch(L3_NAT_DBONLY_MIXIN + '.create_router')
|
||||
def test_update_router_success(self, mock_create, mock_update):
|
||||
mock_update.return_value = {'id': "fake_id"}
|
||||
response = self._create_router(mock_create)
|
||||
router = {'router': {'name': 'fake_name'}}
|
||||
response = self._driver.update_router(
|
||||
self.context, response['id'], router)
|
||||
self.assertIsInstance(response, dict)
|
||||
mock_update.assert_called_once_with(
|
||||
self.context, response['id'], router)
|
||||
|
||||
@mock_ec2
|
||||
@mock.patch(L3_NAT_DBONLY_MIXIN + '.add_router_interface')
|
||||
@mock.patch(
|
||||
'neutron.common.aws_utils.AwsUtils.get_vpc_from_neutron_network_id')
|
||||
@mock.patch('neutron.db.db_base_plugin_v2.NeutronDbPluginV2.get_subnet')
|
||||
@mock.patch(L3_NAT_DBONLY_MIXIN + '.create_router')
|
||||
def test_add_router_interface(self, mock_create, mock_get, mock_vpc,
|
||||
mock_add):
|
||||
aws_obj = AwsUtils()
|
||||
vpc_id = aws_obj.create_vpc_and_tags(self.context.current['cidr'],
|
||||
self._get_fake_tags())
|
||||
interface_info = {'subnet_id': 'fake_subnet_id'}
|
||||
response = self._create_router(mock_create)
|
||||
router_id = response['id']
|
||||
mock_get.return_value = {'network_id': 'fake_network_id'}
|
||||
# We need to mock 'get_vpc_from_neutron_network_id' from aws_utils,
|
||||
# because we need a valid vpc_id when attaching internet gateway.
|
||||
mock_vpc.return_value = vpc_id
|
||||
mock_add.return_value = {'id': 'fake_id',
|
||||
'subnet_id': 'fake_subnet_id'}
|
||||
response = self._driver.add_router_interface(
|
||||
self.context, router_id, interface_info)
|
||||
self.assertIsInstance(response, dict)
|
||||
mock_add.assert_called_once_with(self.context, router_id,
|
||||
interface_info)
|
||||
|
||||
@mock_ec2
|
||||
@mock.patch(L3_NAT_DBONLY_MIXIN + '.remove_router_interface')
|
||||
@mock.patch(L3_NAT_DBONLY_MIXIN + '.create_router')
|
||||
def test_remove_router_interface(self, mock_create, mock_remove):
|
||||
response = self._create_router(mock_create)
|
||||
router_id = response['id']
|
||||
interface_info = {'port_id': 'fake_port_id'}
|
||||
mock_remove.return_value = {'id': 'fake_id',
|
||||
'subnet_id': 'fake_subnet_id'}
|
||||
response = self._driver.remove_router_interface(
|
||||
self.context, router_id, interface_info)
|
||||
self.assertIsInstance(response, dict)
|
||||
mock_remove.assert_called_once_with(self.context, router_id,
|
||||
interface_info)
|
Loading…
Reference in New Issue