"Fossies" - the Fresh Open Source Software Archive

Member "neutron-14.0.3/neutron/tests/unit/db/test_l3_db.py" (22 Oct 2019, 24051 Bytes) of package /linux/misc/openstack/neutron-14.0.3.tar.gz:


As a special service "Fossies" has tried to format the requested source page into HTML format using (guessed) Python source code syntax highlighting (style: standard) with prefixed line numbers. Alternatively you can here view or download the uninterpreted source code file. See also the latest Fossies "Diffs" side-by-side code changes report for "test_l3_db.py": 14.0.2_vs_14.0.3.

    1 # Copyright 2015 Hewlett-Packard Development Company, L.P.
    2 #
    3 # Licensed under the Apache License, Version 2.0 (the "License");
    4 # you may not use this file except in compliance with the License.
    5 # You may obtain a copy of the License at
    6 #
    7 #    http://www.apache.org/licenses/LICENSE-2.0
    8 #
    9 # Unless required by applicable law or agreed to in writing, software
   10 # distributed under the License is distributed on an "AS IS" BASIS,
   11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
   12 # implied.
   13 # See the License for the specific language governing permissions and
   14 # limitations under the License.
   15 
   16 import mock
   17 import netaddr
   18 from neutron_lib.callbacks import events
   19 from neutron_lib.callbacks import registry
   20 from neutron_lib.callbacks import resources
   21 from neutron_lib import constants as n_const
   22 from neutron_lib import context
   23 from neutron_lib import exceptions as n_exc
   24 from neutron_lib.exceptions import l3 as l3_exc
   25 from neutron_lib.plugins import constants as plugin_constants
   26 from neutron_lib.plugins import directory
   27 from neutron_lib.plugins import utils as plugin_utils
   28 from oslo_utils import uuidutils
   29 import testtools
   30 
   31 from neutron.db import l3_db
   32 from neutron.db.models import l3 as l3_models
   33 from neutron.objects import base as base_obj
   34 from neutron.objects import network as network_obj
   35 from neutron.objects import ports as port_obj
   36 from neutron.objects import router as l3_obj
   37 from neutron.objects import subnet as subnet_obj
   38 from neutron.tests import base
   39 from neutron.tests.unit.db import test_db_base_plugin_v2
   40 
   41 
   42 class TestL3_NAT_dbonly_mixin(base.BaseTestCase):
   43     def setUp(self):
   44         super(TestL3_NAT_dbonly_mixin, self).setUp()
   45         self.db = l3_db.L3_NAT_dbonly_mixin()
   46 
   47     def test__each_port_having_fixed_ips_none(self):
   48         """Be sure the method returns an empty list when None is passed"""
   49         filtered = l3_db.L3_NAT_dbonly_mixin._each_port_having_fixed_ips(None)
   50         self.assertEqual([], list(filtered))
   51 
   52     def test__new__passes_args(self):
   53         class T(l3_db.L3_NAT_db_mixin):
   54             def __init__(self, *args, **kwargs):
   55                 self.args = args
   56                 self.kwargs = kwargs
   57 
   58         t = T(1, 2, a=3)
   59         self.assertEqual((1, 2), t.args)
   60         self.assertEqual({'a': 3}, t.kwargs)
   61 
   62     def test__each_port_having_fixed_ips(self):
   63         """Basic test that ports without fixed ips are filtered out"""
   64         ports = [{'id': 'a', 'fixed_ips': [mock.sentinel.fixedip]},
   65                  {'id': 'b'}]
   66         filtered = l3_db.L3_NAT_dbonly_mixin._each_port_having_fixed_ips(ports)
   67         ids = [p['id'] for p in filtered]
   68         self.assertEqual(['a'], ids)
   69 
   70     def test__get_subnets_by_network_no_query(self):
   71         """Basic test that no query is performed if no Ports are passed"""
   72         context = mock.Mock()
   73         with mock.patch.object(directory, 'get_plugin') as get_p:
   74             self.db._get_subnets_by_network_list(context, [])
   75         self.assertFalse(context.session.query.called)
   76         self.assertFalse(get_p.called)
   77 
   78     def test__get_subnets_by_network(self):
   79         """Basic test that the right query is called"""
   80         context = mock.MagicMock()
   81         query = context.session.query().outerjoin().filter()
   82         query.__iter__.return_value = [(mock.sentinel.subnet_db,
   83                                         mock.sentinel.address_scope_id)]
   84 
   85         with mock.patch.object(directory, 'get_plugin') as get_p:
   86             get_p()._make_subnet_dict.return_value = {
   87                 'network_id': mock.sentinel.network_id}
   88             subnets = self.db._get_subnets_by_network_list(
   89                 context, [mock.sentinel.network_id])
   90         self.assertEqual({
   91             mock.sentinel.network_id: [{
   92                 'address_scope_id': mock.sentinel.address_scope_id,
   93                 'network_id': mock.sentinel.network_id}]}, subnets)
   94 
   95     def test__get_mtus_by_network_list(self):
   96         """Basic test that the query get_networks is correctly"""
   97         network = {'id': mock.sentinel.network_id,
   98                    'name': mock.sentinel.name,
   99                    'mtu': mock.sentinel.mtu}
  100         with mock.patch.object(directory, 'get_plugin') as get_p:
  101             get_p().get_networks.return_value = [network]
  102             result = self.db._get_mtus_by_network_list(
  103                 mock.sentinel.context, [mock.sentinel.network_id])
  104             get_p().get_networks.assert_called_once_with(
  105                 mock.sentinel.context,
  106                 filters={'id': [mock.sentinel.network_id]},
  107                 fields=['id', 'mtu'])
  108             self.assertEqual({mock.sentinel.network_id: mock.sentinel.mtu},
  109                              result)
  110 
  111     def test__populate_ports_for_subnets_none(self):
  112         """Basic test that the method runs correctly with no ports"""
  113         ports = []
  114         with mock.patch.object(directory, 'get_plugin') as get_p:
  115             get_p().get_networks.return_value = []
  116             self.db._populate_mtu_and_subnets_for_ports(mock.sentinel.context,
  117                                                         ports)
  118         self.assertEqual([], ports)
  119 
  120     @mock.patch.object(l3_db.L3_NAT_dbonly_mixin,
  121                        '_get_subnets_by_network_list')
  122     def test__populate_ports_for_subnets(self, get_subnets_by_network):
  123         cidr = "2001:db8::/64"
  124         subnet = {'id': mock.sentinel.subnet_id,
  125                   'cidr': cidr,
  126                   'gateway_ip': mock.sentinel.gateway_ip,
  127                   'dns_nameservers': mock.sentinel.dns_nameservers,
  128                   'ipv6_ra_mode': mock.sentinel.ipv6_ra_mode,
  129                   'subnetpool_id': mock.sentinel.subnetpool_id,
  130                   'address_scope_id': mock.sentinel.address_scope_id}
  131         get_subnets_by_network.return_value = {'net_id': [subnet]}
  132 
  133         ports = [{'network_id': 'net_id',
  134                   'id': 'port_id',
  135                   'fixed_ips': [{'subnet_id': mock.sentinel.subnet_id}]}]
  136         with mock.patch.object(directory, 'get_plugin') as get_p:
  137             get_p().get_networks.return_value = [{'id': 'net_id', 'mtu': 1446}]
  138             self.db._populate_mtu_and_subnets_for_ports(mock.sentinel.context,
  139                                                         ports)
  140             keys = ('id', 'cidr', 'gateway_ip', 'ipv6_ra_mode',
  141                     'subnetpool_id', 'dns_nameservers')
  142             address_scopes = {4: None, 6: mock.sentinel.address_scope_id}
  143             self.assertEqual([{'extra_subnets': [],
  144                                'fixed_ips': [{'subnet_id':
  145                                               mock.sentinel.subnet_id,
  146                                               'prefixlen': 64}],
  147                                'id': 'port_id',
  148                                'mtu': 1446,
  149                                'network_id': 'net_id',
  150                                'subnets': [{k: subnet[k] for k in keys}],
  151                                'address_scopes': address_scopes}], ports)
  152 
  153     def test__get_sync_floating_ips_no_query(self):
  154         """Basic test that no query is performed if no router ids are passed"""
  155         db = l3_db.L3_NAT_dbonly_mixin()
  156         context = mock.Mock()
  157         db._get_sync_floating_ips(context, [])
  158         self.assertFalse(context.session.query.called)
  159 
  160     @mock.patch.object(l3_db.L3_NAT_dbonly_mixin, '_make_floatingip_dict')
  161     def test__make_floatingip_dict_with_scope(self, make_fip_dict):
  162         db = l3_db.L3_NAT_dbonly_mixin()
  163         make_fip_dict.return_value = {'id': mock.sentinel.fip_ip}
  164         result = db._make_floatingip_dict_with_scope(
  165             mock.sentinel.floating_ip_db, mock.sentinel.address_scope_id)
  166         self.assertEqual({
  167             'fixed_ip_address_scope': mock.sentinel.address_scope_id,
  168             'id': mock.sentinel.fip_ip}, result)
  169 
  170     def test__unique_floatingip_iterator(self):
  171         context = mock.MagicMock()
  172         query = mock.MagicMock()
  173         query.order_by().__iter__.return_value = [
  174             ({'id': 'id1'}, 'scope1'),
  175             ({'id': 'id1'}, 'scope1'),
  176             ({'id': 'id2'}, 'scope2'),
  177             ({'id': 'id2'}, 'scope2'),
  178             ({'id': 'id2'}, 'scope2'),
  179             ({'id': 'id3'}, 'scope3')]
  180         query.reset_mock()
  181         with mock.patch.object(
  182                 l3_obj.FloatingIP, '_load_object',
  183                 side_effect=({'id': 'id1'}, {'id': 'id2'}, {'id': 'id3'})):
  184             result = list(
  185                 l3_obj.FloatingIP._unique_floatingip_iterator(context, query))
  186             query.order_by.assert_called_once_with(l3_models.FloatingIP.id)
  187             self.assertEqual([({'id': 'id1'}, 'scope1'),
  188                               ({'id': 'id2'}, 'scope2'),
  189                               ({'id': 'id3'}, 'scope3')], result)
  190 
  191     @mock.patch.object(directory, 'get_plugin')
  192     def test_prevent_l3_port_deletion_port_not_found(self, gp):
  193         # port not found doesn't prevent
  194         gp.return_value.get_port.side_effect = n_exc.PortNotFound(port_id='1')
  195         self.db.prevent_l3_port_deletion(None, None)
  196 
  197     @mock.patch.object(directory, 'get_plugin')
  198     def test_prevent_l3_port_device_owner_not_router(self, gp):
  199         # ignores other device owners
  200         gp.return_value.get_port.return_value = {'device_owner': 'cat'}
  201         self.db.prevent_l3_port_deletion(None, None)
  202 
  203     @mock.patch.object(directory, 'get_plugin')
  204     def test_prevent_l3_port_no_fixed_ips(self, gp):
  205         # without fixed IPs is allowed
  206         gp.return_value.get_port.return_value = {
  207             'device_owner': n_const.DEVICE_OWNER_ROUTER_INTF, 'fixed_ips': [],
  208             'id': 'f'
  209         }
  210         self.db.prevent_l3_port_deletion(None, None)
  211 
  212     @mock.patch.object(directory, 'get_plugin')
  213     def test_prevent_l3_port_no_router(self, gp):
  214         # without router is allowed
  215         gp.return_value.get_port.return_value = {
  216             'device_owner': n_const.DEVICE_OWNER_ROUTER_INTF,
  217             'device_id': '44', 'id': 'f',
  218             'fixed_ips': [{'ip_address': '1.1.1.1', 'subnet_id': '4'}]}
  219         self.db.get_router = mock.Mock()
  220         self.db.get_router.side_effect = l3_exc.RouterNotFound(router_id='44')
  221         self.db.prevent_l3_port_deletion(mock.Mock(), None)
  222 
  223     @mock.patch.object(directory, 'get_plugin')
  224     def test_prevent_l3_port_existing_router(self, gp):
  225         gp.return_value.get_port.return_value = {
  226             'device_owner': n_const.DEVICE_OWNER_ROUTER_INTF,
  227             'device_id': 'some_router', 'id': 'f',
  228             'fixed_ips': [{'ip_address': '1.1.1.1', 'subnet_id': '4'}]}
  229         self.db.get_router = mock.Mock()
  230         with testtools.ExpectedException(n_exc.ServicePortInUse):
  231             self.db.prevent_l3_port_deletion(mock.Mock(), None)
  232 
  233     @mock.patch.object(directory, 'get_plugin')
  234     def test_prevent_l3_port_existing_floating_ip(self, gp):
  235         ctx = context.get_admin_context()
  236         gp.return_value.get_port.return_value = {
  237             'device_owner': n_const.DEVICE_OWNER_FLOATINGIP,
  238             'device_id': 'some_flip', 'id': 'f',
  239             'fixed_ips': [{'ip_address': '1.1.1.1', 'subnet_id': '4'}]}
  240         with mock.patch.object(l3_obj.FloatingIP, 'objects_exist',
  241                                return_value=mock.Mock()),\
  242             testtools.ExpectedException(n_exc.ServicePortInUse):
  243 
  244             self.db.prevent_l3_port_deletion(ctx, None)
  245 
  246     @mock.patch.object(directory, 'get_plugin')
  247     def test_subscribe_address_scope_of_subnetpool(self, gp):
  248         l3_db.L3RpcNotifierMixin()
  249         registry.publish(resources.SUBNETPOOL_ADDRESS_SCOPE,
  250                          events.AFTER_UPDATE, mock.ANY,
  251                          payload=events.DBEventPayload(
  252                              mock.MagicMock(), resource_id='fake_id'))
  253         self.assertTrue(gp.return_value.notify_routers_updated.called)
  254 
  255     def test__check_and_get_fip_assoc_with_extra_association_no_change(self):
  256         fip = {'extra_key': 'value'}
  257         context = mock.MagicMock()
  258         floatingip_obj = l3_obj.FloatingIP(
  259             context,
  260             id=uuidutils.generate_uuid(),
  261             floating_network_id=uuidutils.generate_uuid(),
  262             floating_ip_address=netaddr.IPAddress('8.8.8.8'),
  263             fixed_port_id=uuidutils.generate_uuid(),
  264             floating_port_id=uuidutils.generate_uuid())
  265         with mock.patch.object(
  266                 l3_db.L3_NAT_dbonly_mixin,
  267                 '_get_assoc_data',
  268                 return_value=('1', '2', '3')) as mock_get_assoc_data:
  269             self.db._check_and_get_fip_assoc(context, fip, floatingip_obj)
  270             context.session.query.assert_not_called()
  271             mock_get_assoc_data.assert_called_once_with(
  272                 mock.ANY, fip, floatingip_obj)
  273 
  274     def test__notify_attaching_interface(self):
  275         with mock.patch.object(l3_db.registry, 'notify') as mock_notify:
  276             context = mock.MagicMock()
  277             router_id = 'router_id'
  278             net_id = 'net_id'
  279             router_db = mock.Mock()
  280             router_db.id = router_id
  281             port = {'network_id': net_id}
  282             intf = {}
  283             self.db._notify_attaching_interface(context, router_db, port, intf)
  284             kwargs = {'context': context, 'router_id': router_id,
  285                       'network_id': net_id, 'interface_info': intf,
  286                       'router_db': router_db, 'port': port}
  287             mock_notify.assert_called_once_with(
  288                 resources.ROUTER_INTERFACE, events.BEFORE_CREATE, self.db,
  289                 **kwargs)
  290 
  291     def test__create_gw_port(self):
  292         router_id = '2afb8434-7380-43a2-913f-ba3a5ad5f349'
  293         router = l3_models.Router(id=router_id)
  294         new_network_id = 'net-id'
  295         ext_ips = [{'subnet_id': 'subnet-id', 'ip_address': '1.1.1.1'}]
  296         gw_port = {'fixed_ips': [{'subnet_id': 'subnet-id',
  297                                   'ip_address': '1.1.1.1'}],
  298                    'id': '8742d007-6f05-4b7e-abdb-11818f608959'}
  299         ctx = context.get_admin_context()
  300 
  301         with mock.patch.object(directory, 'get_plugin') as get_p, \
  302                 mock.patch.object(get_p(), 'get_subnets_by_network',
  303                                   return_value=mock.ANY), \
  304                 mock.patch.object(get_p(), '_get_port',
  305                                   return_value=gw_port), \
  306                 mock.patch.object(l3_db.L3_NAT_dbonly_mixin,
  307                                   '_check_for_dup_router_subnets') as cfdrs,\
  308                 mock.patch.object(plugin_utils, 'create_port',
  309                                   return_value=gw_port), \
  310                 mock.patch.object(ctx.session, 'add'), \
  311                 mock.patch.object(base_obj.NeutronDbObject, 'create'), \
  312                 mock.patch.object(l3_db.registry, 'notify') as mock_notify:
  313 
  314             self.db._create_gw_port(ctx, router_id=router_id,
  315                                     router=router,
  316                                     new_network_id=new_network_id,
  317                                     ext_ips=ext_ips)
  318 
  319             expected_gw_ips = ['1.1.1.1']
  320 
  321             self.assertTrue(cfdrs.called)
  322             mock_notify.assert_called_with(
  323                 resources.ROUTER_GATEWAY, events.AFTER_CREATE,
  324                 self.db._create_gw_port, context=ctx,
  325                 gw_ips=expected_gw_ips, network_id=new_network_id,
  326                 router_id=router_id
  327             )
  328 
  329 
  330 class L3_NAT_db_mixin(base.BaseTestCase):
  331     def setUp(self):
  332         super(L3_NAT_db_mixin, self).setUp()
  333         self.db = l3_db.L3_NAT_db_mixin()
  334 
  335     def _test_create_router(self, external_gateway_info=None):
  336         router_db = l3_models.Router(id='123')
  337         router_dict = {'id': '123', 'tenant_id': '456',
  338                        'external_gateway_info': external_gateway_info}
  339         # Need to use a copy here as the create_router method pops the gateway
  340         # information
  341         router_input = {'router': router_dict.copy()}
  342 
  343         with mock.patch.object(l3_db.L3_NAT_dbonly_mixin, '_create_router_db',
  344                                return_value=router_db) as crd,\
  345             mock.patch.object(l3_db.L3_NAT_dbonly_mixin, '_make_router_dict',
  346                               return_value=router_dict),\
  347             mock.patch.object(l3_db.L3_NAT_dbonly_mixin,
  348                               '_update_router_gw_info') as urgi,\
  349             mock.patch.object(l3_db.L3_NAT_dbonly_mixin, '_get_router',
  350                               return_value=router_db),\
  351             mock.patch.object(l3_db.L3_NAT_db_mixin, 'notify_router_updated')\
  352             as nru:
  353 
  354             self.db.create_router(mock.Mock(), router_input)
  355             self.assertTrue(crd.called)
  356             if external_gateway_info:
  357                 self.assertTrue(urgi.called)
  358                 self.assertTrue(nru.called)
  359             else:
  360                 self.assertFalse(urgi.called)
  361                 self.assertFalse(nru.called)
  362 
  363     def test_create_router_no_gateway(self):
  364         self._test_create_router()
  365 
  366     def test_create_router_gateway(self):
  367         ext_gateway_info = {'network_id': 'net-id', 'enable_snat': True,
  368                             'external_fixed_ips': [
  369                                 {'subnet_id': 'subnet-id',
  370                                  'ip_address': 'ip'}]}
  371         self._test_create_router(ext_gateway_info)
  372 
  373     def test_add_router_interface_no_interface_info(self):
  374         router_db = l3_models.Router(id='123')
  375         with mock.patch.object(l3_db.L3_NAT_dbonly_mixin, '_get_router',
  376                                return_value=router_db):
  377             self.assertRaises(
  378                 n_exc.BadRequest,
  379                 self.db.add_router_interface, mock.Mock(), router_db.id)
  380 
  381 
  382 class FakeL3Plugin(l3_db.L3_NAT_dbonly_mixin):
  383     pass
  384 
  385 
  386 class L3TestCase(test_db_base_plugin_v2.NeutronDbPluginV2TestCase):
  387 
  388     GET_PORTS_BY_ROUTER_MSG = (
  389         'The following ports, assigned to router %(router_id)s, do not have a '
  390         '"routerport" register: %(port_ids)s')
  391 
  392     def setUp(self, *args, **kwargs):
  393         super(L3TestCase, self).setUp(plugin='ml2')
  394         self.core_plugin = directory.get_plugin()
  395         self.ctx = context.get_admin_context()
  396         self.mixin = FakeL3Plugin()
  397         directory.add_plugin(plugin_constants.L3, self.mixin)
  398         self.network = self.create_network()
  399         self.subnets = []
  400         self.subnets.append(self.create_subnet(self.network, '1.1.1.1',
  401                                                '1.1.1.0/24'))
  402         self.subnets.append(self.create_subnet(self.network, '1.1.2.1',
  403                                                '1.1.2.0/24'))
  404         router = {'router': {'name': 'foo_router', 'admin_state_up': True,
  405                              'tenant_id': 'foo_tenant'}}
  406         self.router = self.create_router(router)
  407         self.ports = []
  408         for subnet in self.subnets:
  409             ipa = str(netaddr.IPNetwork(subnet['subnet']['cidr']).ip + 10)
  410             fixed_ips = [{'subnet_id': subnet['subnet']['id'],
  411                           'ip_address': ipa}]
  412             self.ports.append(self.create_port(
  413                 self.network['network']['id'], {'fixed_ips': fixed_ips}))
  414         self.addCleanup(self._clean_objs)
  415 
  416     def _clean_objs(self):
  417         port_obj.Port.delete_objects(
  418             self.ctx, network_id=self.network['network']['id'])
  419         subnet_obj.Subnet.delete_objects(
  420             self.ctx, network_id=self.network['network']['id'])
  421         network_obj.Network.get_object(
  422             self.ctx, id=self.network['network']['id']).delete()
  423         l3_obj.Router.get_object(self.ctx, id=self.router['id']).delete()
  424 
  425     def create_router(self, router):
  426         with self.ctx.session.begin(subtransactions=True):
  427             return self.mixin.create_router(self.ctx, router)
  428 
  429     def create_port(self, net_id, port_info):
  430         with self.ctx.session.begin(subtransactions=True):
  431             return self._make_port(self.fmt, net_id, **port_info)
  432 
  433     def create_network(self, name=None, **kwargs):
  434         name = name or 'network1'
  435         with self.ctx.session.begin(subtransactions=True):
  436             return self._make_network(self.fmt, name, True, **kwargs)
  437 
  438     def create_subnet(self, network, gateway, cidr, **kwargs):
  439         with self.ctx.session.begin(subtransactions=True):
  440             return self._make_subnet(self.fmt, network, gateway, cidr,
  441                                      **kwargs)
  442 
  443     def _add_router_interfaces(self):
  444         return [self.mixin.add_router_interface(
  445             self.ctx, self.router['id'],
  446             interface_info={'port_id': port['port']['id']})
  447             for port in self.ports]
  448 
  449     def _check_routerports(self, ri_statuses):
  450         port_ids = []
  451         for idx, ri_status in enumerate(ri_statuses):
  452             rp_obj = l3_obj.RouterPort.get_object(
  453                 self.ctx, port_id=self.ports[idx]['port']['id'],
  454                 router_id=self.router['id'])
  455             if ri_status:
  456                 self.assertEqual(self.ports[idx]['port']['id'], rp_obj.port_id)
  457                 port_ids.append(rp_obj.port_id)
  458             else:
  459                 self.assertIsNone(rp_obj)
  460 
  461         _router_obj = l3_obj.Router.get_object(self.ctx, id=self.router['id'])
  462         router_port_ids = [rp.port_id for rp in
  463                            _router_obj.db_obj.attached_ports]
  464         self.assertEqual(sorted(port_ids), sorted(router_port_ids))
  465 
  466     @mock.patch.object(port_obj, 'LOG')
  467     def test_remove_router_interface_by_port(self, mock_log):
  468         self._add_router_interfaces()
  469         self._check_routerports((True, True))
  470 
  471         interface_info = {'port_id': self.ports[0]['port']['id']}
  472         self.mixin.remove_router_interface(self.ctx, self.router['id'],
  473                                            interface_info)
  474         mock_log.warning.assert_not_called()
  475         self._check_routerports((False, True))
  476 
  477     @mock.patch.object(port_obj, 'LOG')
  478     def test_remove_router_interface_by_port_removed_rport(self, mock_log):
  479         self._add_router_interfaces()
  480         self._check_routerports((True, True))
  481 
  482         rp_obj = l3_obj.RouterPort.get_object(
  483             self.ctx, router_id=self.router['id'],
  484             port_id=self.ports[0]['port']['id'])
  485         rp_obj.delete()
  486 
  487         interface_info = {'port_id': self.ports[0]['port']['id']}
  488         self.mixin.remove_router_interface(self.ctx, self.router['id'],
  489                                            interface_info)
  490         msg_vars = {'router_id': self.router['id'],
  491                     'port_ids': {self.ports[0]['port']['id']}}
  492         mock_log.warning.assert_called_once_with(self.GET_PORTS_BY_ROUTER_MSG,
  493                                                  msg_vars)
  494         self._check_routerports((False, True))
  495 
  496     @mock.patch.object(port_obj, 'LOG')
  497     def test_remove_router_interface_by_subnet(self, mock_log):
  498         self._add_router_interfaces()
  499         self._check_routerports((True, True))
  500 
  501         interface_info = {'subnet_id': self.subnets[1]['subnet']['id']}
  502         self.mixin.remove_router_interface(self.ctx, self.router['id'],
  503                                            interface_info)
  504         mock_log.warning.not_called_once()
  505         self._check_routerports((True, False))
  506 
  507     @mock.patch.object(port_obj, 'LOG')
  508     def test_remove_router_interface_by_subnet_removed_rport(self, mock_log):
  509         self._add_router_interfaces()
  510         self._check_routerports((True, True))
  511 
  512         rp_obj = l3_obj.RouterPort.get_object(
  513             self.ctx, router_id=self.router['id'],
  514             port_id=self.ports[0]['port']['id'])
  515         rp_obj.delete()
  516 
  517         interface_info = {'subnet_id': self.subnets[0]['subnet']['id']}
  518         self.mixin.remove_router_interface(self.ctx, self.router['id'],
  519                                            interface_info)
  520         msg_vars = {'router_id': self.router['id'],
  521                     'port_ids': {self.ports[0]['port']['id']}}
  522         mock_log.warning.assert_called_once_with(self.GET_PORTS_BY_ROUTER_MSG,
  523                                                  msg_vars)
  524         self._check_routerports((False, True))