OpenStack Networking (Neutron)
You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
 
 
 
 

456 lines
21 KiB

# Copyright (C) 2018 OpenStack Foundation
# 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 mock
from neutron_lib.callbacks import registry
from neutron_lib import context
from neutron_lib import exceptions as lib_exc
from neutron_lib.exceptions import l3 as lib_l3_exc
from neutron_lib.objects import exceptions as obj_exc
from neutron_lib.plugins import constants as lib_plugin_conts
from neutron_lib.plugins import directory
from oslo_config import cfg
from neutron.api.rpc.callbacks.consumer import registry as cons_registry
from neutron.api.rpc.callbacks import events as rpc_events
from neutron.api.rpc.callbacks.producer import registry as prod_registry
from neutron.api.rpc.callbacks import resource_manager
from neutron.api.rpc.handlers import resources_rpc
from neutron.db import db_base_plugin_v2
from neutron.db import l3_db
from neutron import manager
from neutron.objects import base as obj_base
from neutron.objects import port_forwarding
from neutron.objects import router
from neutron.services.portforwarding.common import exceptions as pf_exc
from neutron.services.portforwarding import pf_plugin
from neutron.tests.unit import testlib_api
DB_PLUGIN_KLASS = 'neutron.db.db_base_plugin_v2.NeutronDbPluginV2'
class TestPortForwardingPlugin(testlib_api.SqlTestCase):
def setUp(self):
super(TestPortForwardingPlugin, self).setUp()
with mock.patch.object(
resource_manager.ResourceCallbacksManager, '_singleton',
new_callable=mock.PropertyMock(return_value=False)):
self.cons_mgr = resource_manager.ConsumerResourceCallbacksManager()
self.prod_mgr = resource_manager.ProducerResourceCallbacksManager()
for mgr in (self.cons_mgr, self.prod_mgr):
mgr.clear()
mock.patch.object(
cons_registry, '_get_manager', return_value=self.cons_mgr).start()
mock.patch.object(
prod_registry, '_get_manager', return_value=self.prod_mgr).start()
self.setup_coreplugin(load_plugins=False)
mock.patch('neutron.objects.db.api.create_object').start()
mock.patch('neutron.objects.db.api.update_object').start()
mock.patch('neutron.objects.db.api.delete_object').start()
mock.patch('neutron.objects.db.api.get_object').start()
# We don't use real models as per mocks above. We also need to mock-out
# methods that work with real data types
mock.patch(
'neutron.objects.base.NeutronDbObject.modify_fields_from_db'
).start()
cfg.CONF.set_override("core_plugin", DB_PLUGIN_KLASS)
cfg.CONF.set_override("service_plugins", ["router", "port_forwarding"])
manager.init()
self.pf_plugin = directory.get_plugin(lib_plugin_conts.PORTFORWARDING)
self.ctxt = context.Context('admin', 'fake_tenant')
mock.patch.object(self.ctxt.session, 'refresh').start()
mock.patch.object(self.ctxt.session, 'expunge').start()
@mock.patch.object(port_forwarding.PortForwarding, 'get_object')
def test_get_floatingip_port_forwarding(self, get_object_mock):
self.pf_plugin.get_floatingip_port_forwarding(
self.ctxt, 'pf_id', 'test-fip-id', fields=None)
get_object_mock.assert_called_once_with(self.ctxt, id='pf_id')
@mock.patch.object(port_forwarding.PortForwarding, 'get_object',
return_value=None)
def test_negative_get_floatingip_port_forwarding(self, get_object_mock):
self.assertRaises(
pf_exc.PortForwardingNotFound,
self.pf_plugin.get_floatingip_port_forwarding,
self.ctxt, 'pf_id', 'test-fip-id', fields=None)
@mock.patch.object(port_forwarding.PortForwarding, 'get_objects')
def test_get_floatingip_port_forwardings(self, get_objects_mock):
self.pf_plugin.get_floatingip_port_forwardings(self.ctxt)
get_objects_mock.assert_called_once_with(
self.ctxt, _pager=mock.ANY, floatingip_id=None)
@mock.patch.object(resources_rpc.ResourcesPushRpcApi, 'push')
@mock.patch.object(port_forwarding.PortForwarding, 'get_object')
@mock.patch.object(port_forwarding.PortForwarding, 'get_objects')
@mock.patch.object(router.FloatingIP, 'get_object')
def test_delete_floatingip_port_forwarding(
self, fip_get_object_mock, pf_get_objects_mock,
pf_get_object_mock, push_api_mock):
# After delete, not empty resource list
pf_get_objects_mock.return_value = [mock.Mock(id='pf_id'),
mock.Mock(id='pf_id2')]
pf_obj = mock.Mock(id='pf_id', floatingip_id='fip_id')
pf_get_object_mock.return_value = pf_obj
self.pf_plugin.delete_floatingip_port_forwarding(
self.ctxt, 'pf_id', 'fip_id')
pf_get_objects_mock.assert_called_once_with(
self.ctxt, floatingip_id='fip_id')
pf_obj.delete.assert_called()
push_api_mock.assert_called_once_with(
self.ctxt, mock.ANY, rpc_events.DELETED)
# After delete, empty resource list
pf_get_objects_mock.reset_mock()
pf_get_object_mock.reset_mock()
push_api_mock.reset_mock()
pf_obj = mock.Mock(id='need_to_delete_pf_id', floatingip_id='fip_id')
fip_obj = mock.Mock(id='fip_id')
fip_get_object_mock.return_value = fip_obj
pf_get_object_mock.return_value = pf_obj
pf_get_objects_mock.return_value = [
mock.Mock(id='need_to_delete_pf_id')]
self.pf_plugin.delete_floatingip_port_forwarding(
self.ctxt, 'need_to_delete_pf_id', 'fip_id')
pf_get_objects_mock.assert_called_once_with(
self.ctxt, floatingip_id='fip_id')
pf_get_object_mock.assert_called_once_with(
self.ctxt, id='need_to_delete_pf_id')
fip_obj.update_fields.assert_called_once_with({'router_id': None})
fip_obj.update.assert_called()
push_api_mock.assert_called_once_with(
self.ctxt, mock.ANY, rpc_events.DELETED)
@mock.patch.object(port_forwarding.PortForwarding, 'get_object')
def test_negative_delete_floatingip_port_forwarding(
self, pf_get_object_mock):
pf_get_object_mock.return_value = None
self.assertRaises(
pf_exc.PortForwardingNotFound,
self.pf_plugin.delete_floatingip_port_forwarding,
self.ctxt, 'pf_id', floatingip_id='fip_id')
@mock.patch.object(port_forwarding.PortForwarding, 'get_objects')
@mock.patch.object(db_base_plugin_v2.NeutronDbPluginV2, 'get_port')
@mock.patch.object(registry, 'notify')
@mock.patch.object(resources_rpc.ResourcesPushRpcApi, 'push')
@mock.patch.object(port_forwarding.PortForwarding, 'get_object')
def test_update_floatingip_port_forwarding(
self, mock_pf_get_object, mock_rpc_push, mock_registry_notify,
mock_get_port, mock_pf_get_objects):
pf_input = {
'port_forwarding':
{'port_forwarding': {
'internal_ip_address': '1.1.1.1',
'floatingip_id': 'fip_id'}},
'floatingip_id': 'fip_id'}
pf_obj = mock.Mock()
pf_obj.internal_ip_address = "10.0.0.1"
mock_pf_get_object.return_value = pf_obj
port_dict = {'id': 'ID', 'fixed_ips': [{"subnet_id": "test-subnet-id",
"ip_address": "10.0.0.1"}]}
mock_get_port.return_value = port_dict
mock_pf_get_objects.return_value = []
self.pf_plugin.update_floatingip_port_forwarding(
self.ctxt, 'pf_id', **pf_input)
mock_pf_get_object.assert_called_once_with(self.ctxt, id='pf_id')
self.assertTrue(pf_obj.update_fields)
self.assertTrue(pf_obj.update)
mock_rpc_push.assert_called_once_with(
self.ctxt, mock.ANY, rpc_events.UPDATED)
@mock.patch.object(port_forwarding.PortForwarding, 'get_objects')
@mock.patch.object(db_base_plugin_v2.NeutronDbPluginV2, 'get_port')
def test_check_port_forwarding_update(self, mock_get_port,
mock_pf_get_objects):
port_dict = {'fixed_ips': [{"ip_address": "10.0.0.1"}]}
mock_get_port.return_value = port_dict
other_pf_obj = mock.Mock(id='cmp_obj_id')
pf_obj = mock.Mock(id='pf_obj_id', internal_port_id='int_port_id',
internal_ip_address='10.0.0.1')
mock_pf_get_objects.return_value = [pf_obj, other_pf_obj]
self.pf_plugin._check_port_forwarding_update(self.ctxt, pf_obj)
mock_get_port.assert_called_once_with(self.ctxt, 'int_port_id')
mock_pf_get_objects.assert_called_once_with(self.ctxt,
floatingip_id=pf_obj.floatingip_id, protocol=pf_obj.protocol)
@mock.patch.object(port_forwarding.PortForwarding, 'get_objects')
@mock.patch.object(db_base_plugin_v2.NeutronDbPluginV2, 'get_port')
def test_check_port_forwarding_update_invalid_address(
self, mock_get_port, mock_pf_get_objects):
port_dict = {'fixed_ips': [{"ip_address": "10.0.0.1"}]}
mock_get_port.return_value = port_dict
pf_obj = mock.Mock(id='pf_obj_id', internal_port_id='int_port_id',
internal_ip_address="99.99.99.99")
self.assertRaisesRegex(
lib_exc.BadRequest,
"not correspond to an address on the internal port",
self.pf_plugin._check_port_forwarding_update,
self.ctxt, pf_obj)
mock_get_port.assert_called_once_with(self.ctxt, 'int_port_id')
mock_pf_get_objects.assert_not_called()
@mock.patch.object(port_forwarding.PortForwarding, 'get_objects')
@mock.patch.object(db_base_plugin_v2.NeutronDbPluginV2, 'get_port')
def test_check_port_forwarding_update_invalid_external(
self, mock_get_port, mock_pf_get_objects):
port_dict = {'fixed_ips': [{"ip_address": "10.0.0.1"}]}
mock_get_port.return_value = port_dict
pf_obj_dict = {'id': 'pf_obj_one',
'floating_ip_address': 'same_fip_addr',
'external_port': 'same_ext_port'}
other_pf_obj = mock.Mock(**pf_obj_dict)
pf_obj_dict.update(id='pf_obj_two', internal_ip_address='10.0.0.1')
pf_obj = mock.Mock(**pf_obj_dict)
mock_pf_get_objects.return_value = [pf_obj, other_pf_obj]
self.assertRaisesRegex(
lib_exc.BadRequest,
"already exist.*same_fip_addr",
self.pf_plugin._check_port_forwarding_update,
self.ctxt, pf_obj)
mock_get_port.assert_called_once_with(self.ctxt, mock.ANY)
mock_pf_get_objects.assert_called_with(
self.ctxt, floatingip_id=mock.ANY, protocol=mock.ANY)
@mock.patch.object(port_forwarding.PortForwarding, 'get_objects')
@mock.patch.object(db_base_plugin_v2.NeutronDbPluginV2, 'get_port')
def test_check_port_forwarding_update_invalid_internal(
self, mock_get_port, mock_pf_get_objects):
same_internal_ip = "10.0.0.10"
port_dict = {'fixed_ips': [{"ip_address": same_internal_ip}]}
mock_get_port.return_value = port_dict
pf_obj_dict = {'id': 'pf_obj_one',
'internal_port_id': 'same_int_port_id',
'internal_ip_address': same_internal_ip,
'internal_port': 'same_int_port'}
other_pf_obj = mock.Mock(**pf_obj_dict)
pf_obj_dict.update(id='pf_obj_two')
pf_obj = mock.Mock(**pf_obj_dict)
mock_pf_get_objects.return_value = [pf_obj, other_pf_obj]
self.assertRaisesRegex(
lib_exc.BadRequest,
"already exist.*{}".format(same_internal_ip),
self.pf_plugin._check_port_forwarding_update,
self.ctxt, pf_obj)
mock_get_port.assert_called_once_with(self.ctxt, 'same_int_port_id')
mock_pf_get_objects.assert_called_with(
self.ctxt, floatingip_id=mock.ANY, protocol=mock.ANY)
@mock.patch.object(port_forwarding.PortForwarding, 'get_object')
def test_negative_update_floatingip_port_forwarding(
self, mock_pf_get_object):
pf_input = {
'port_forwarding':
{'port_forwarding': {
'internal_ip_address': '1.1.1.1',
'floatingip_id': 'fip_id'}},
'floatingip_id': 'fip_id'}
mock_pf_get_object.return_value = None
self.assertRaises(
pf_exc.PortForwardingNotFound,
self.pf_plugin.update_floatingip_port_forwarding,
self.ctxt, 'pf_id', **pf_input)
@mock.patch.object(pf_plugin.PortForwardingPlugin,
'_check_port_has_binding_floating_ip')
@mock.patch.object(obj_base.NeutronDbObject, 'update_objects')
@mock.patch.object(resources_rpc.ResourcesPushRpcApi, 'push')
@mock.patch.object(pf_plugin.PortForwardingPlugin, '_check_router_match')
@mock.patch.object(pf_plugin.PortForwardingPlugin,
'_find_a_router_for_fip_port_forwarding',
return_value='target_router_id')
@mock.patch.object(router.FloatingIP, 'get_object')
@mock.patch('neutron.objects.port_forwarding.PortForwarding')
def test_create_floatingip_port_forwarding(
self, mock_port_forwarding, mock_fip_get_object, mock_find_router,
mock_check_router_match, mock_push_api, mock_update_objects,
mock_check_bind_fip):
# Update fip
pf_input = {
'port_forwarding':
{'port_forwarding': {
'internal_ip_address': '1.1.1.1',
'floatingip_id': 'fip_id'}},
'floatingip_id': 'fip_id'}
pf_obj = mock.Mock()
fip_obj = mock.Mock()
mock_port_forwarding.return_value = pf_obj
mock_fip_get_object.return_value = fip_obj
fip_obj.router_id = ''
fip_obj.fixed_port_id = ''
self.pf_plugin.create_floatingip_port_forwarding(
self.ctxt, **pf_input)
mock_port_forwarding.assert_called_once_with(
self.ctxt, **pf_input['port_forwarding']['port_forwarding'])
self.assertTrue(mock_update_objects.called)
self.assertTrue(pf_obj.create.called)
mock_push_api.assert_called_once_with(
self.ctxt, mock.ANY, rpc_events.CREATED)
# Not update fip
pf_obj.reset_mock()
fip_obj.reset_mock()
mock_port_forwarding.reset_mock()
mock_update_objects.reset_mock()
mock_push_api.reset_mock()
mock_port_forwarding.return_value = pf_obj
fip_obj.router_id = 'router_id'
fip_obj.fixed_port_id = ''
self.pf_plugin.create_floatingip_port_forwarding(
self.ctxt, **pf_input)
mock_port_forwarding.assert_called_once_with(
self.ctxt, **pf_input['port_forwarding']['port_forwarding'])
self.assertTrue(pf_obj.create.called)
self.assertFalse(mock_update_objects.called)
mock_push_api.assert_called_once_with(
self.ctxt, mock.ANY, rpc_events.CREATED)
@mock.patch.object(pf_plugin.PortForwardingPlugin,
'_check_port_has_binding_floating_ip')
@mock.patch.object(pf_plugin.PortForwardingPlugin,
'_find_existing_port_forwarding')
@mock.patch.object(pf_plugin.PortForwardingPlugin,
'_check_router_match')
@mock.patch.object(pf_plugin.PortForwardingPlugin,
'_find_a_router_for_fip_port_forwarding',
return_value='target_router_id')
@mock.patch.object(router.FloatingIP, 'get_object')
@mock.patch('neutron.objects.port_forwarding.PortForwarding')
def test_negative_create_floatingip_port_forwarding(
self, mock_port_forwarding, mock_fip_get_object,
mock_find_router,
mock_check_router_match, mock_try_find_exist,
mock_check_bind_fip):
pf_input = {
'port_forwarding': {
'internal_ip_address': '1.1.1.1',
'floatingip_id': 'fip_id'}}
pf_obj = mock.Mock()
fip_obj = mock.Mock()
mock_port_forwarding.return_value = pf_obj
mock_fip_get_object.return_value = fip_obj
fip_obj.fixed_port_id = ''
pf_obj.create.side_effect = obj_exc.NeutronDbObjectDuplicateEntry(
mock.Mock(), mock.Mock())
mock_try_find_exist.return_value = ('pf_obj', 'conflict_param')
self.assertRaises(
lib_exc.BadRequest,
self.pf_plugin.create_floatingip_port_forwarding,
self.ctxt, 'fip_id', pf_input)
@mock.patch.object(pf_plugin.PortForwardingPlugin,
'_get_internal_ip_subnet')
@mock.patch.object(l3_db.L3_NAT_dbonly_mixin, 'get_router_for_floatingip')
@mock.patch.object(db_base_plugin_v2.NeutronDbPluginV2, 'get_port')
@mock.patch.object(db_base_plugin_v2.NeutronDbPluginV2, 'get_subnet')
def test_negative_find_a_router_for_fip_port_forwarding(
self, mock_get_subnet, mock_get_port, mock_get_router,
mock_get_ip_subnet):
fip_obj = mock.Mock()
pf_dict = {'internal_port_id': 'internal_neutron_port',
'internal_ip_address': '10.0.0.1'}
port_dict = {'id': 'ID', 'fixed_ips': [{"subnet_id": "test-subnet-id",
"ip_address": "10.0.0.1"}]}
mock_get_port.return_value = port_dict
mock_get_ip_subnet.return_value = None
self.assertRaises(
lib_exc.BadRequest,
self.pf_plugin._find_a_router_for_fip_port_forwarding,
self.ctxt, pf_dict, fip_obj)
self.assertTrue(not mock_get_subnet.called)
mock_get_ip_subnet.return_value = 'internal_subnet_id'
mock_get_router.side_effect = (
lib_l3_exc.ExternalGatewayForFloatingIPNotFound(
external_network_id=mock.Mock(),
subnet_id=mock.Mock(), port_id=mock.Mock()))
self.assertRaises(
lib_exc.BadRequest,
self.pf_plugin._find_a_router_for_fip_port_forwarding,
self.ctxt, pf_dict, fip_obj)
self.assertTrue(mock_get_subnet.called)
ipv6_port_dict = {'id': 'ID',
'fixed_ips': [{"subnet_id": "test-subnet-id",
"ip_address": "1::1"}]}
mock_get_port.return_value = ipv6_port_dict
self.assertRaises(
lib_exc.BadRequest,
self.pf_plugin._find_a_router_for_fip_port_forwarding,
self.ctxt, pf_dict, fip_obj)
@mock.patch.object(port_forwarding.PortForwarding, 'get_objects')
def test_negative_check_router_match(self, mock_pf_get_objects):
pf_dict = {
'internal_port_id': 'internal_neutron_port',
'internal_ip_address': 'internal_fixed_ip',
'internal_port': 'internal protocol port num'}
fip_obj = mock.Mock()
mock_pf_get_objects.return_value = ['Exist port forwardings']
router_id = 'selected router id'
self.assertRaises(lib_exc.BadRequest,
self.pf_plugin._check_router_match,
self.ctxt, fip_obj, router_id, pf_dict)
mock_pf_get_objects.return_value = []
self.assertRaises(lib_exc.BadRequest,
self.pf_plugin._check_router_match,
self.ctxt, fip_obj, router_id, pf_dict)
@mock.patch.object(router.FloatingIP, 'get_objects')
def test_create_floatingip_port_forwarding_port_in_use(
self, mock_fip_get_objects):
pf_input = {
'port_forwarding':
{'port_forwarding': {
'internal_ip_address': '1.1.1.1',
'internal_port_id': 'internal_neutron_port',
'floatingip_id': 'fip_id_1'}},
'floatingip_id': 'fip_id_1'}
fip_obj = mock.Mock(floating_ip_address="10.10.10.10")
mock_fip_get_objects.return_value = [fip_obj]
self.assertRaises(pf_exc.PortHasBindingFloatingIP,
self.pf_plugin.create_floatingip_port_forwarding,
self.ctxt, **pf_input)
@mock.patch.object(router.FloatingIP, 'get_objects')
def test_update_floatingip_port_forwarding_port_in_use(
self, mock_fip_get_objects):
pf_input = {
'port_forwarding':
{'port_forwarding': {
'internal_ip_address': '1.1.1.1',
'internal_port_id': 'internal_neutron_port',
'floatingip_id': 'fip_id_2'}}}
fip_obj = mock.Mock(floating_ip_address="10.10.10.11")
mock_fip_get_objects.return_value = [fip_obj]
self.assertRaises(pf_exc.PortHasBindingFloatingIP,
self.pf_plugin.update_floatingip_port_forwarding,
self.ctxt, 'fake-pf-id', 'fip_id_2', **pf_input)