Use directly neutron.common.constants constants in l3_dvr_db

l3_dvr_db defines local constants referencing nonlocal[1]
constants but uses local and nonlocal constants. This change
unifies the code by removing local constants and
using [1] constants.

[1] in neutron.common.constants module

Change-Id: I6548c8eb2a434967b37910e39fa79a883da4e280

diff --git a/neutron/db/l3_dvr_db.py b/neutron/db/l3_dvr_db.py
index 16f48c86f33d2d55e3a3d58064eb4e10e4487063..f431af2aa98fe9b3603f3a8b73be02963ceaa625 100644 (file)
--- a/neutron/db/l3_dvr_db.py
+++ b/neutron/db/l3_dvr_db.py
@@ -39,14 +39,6 @@ from neutron.plugins.common import utils as p_utils
 
 
 LOG = logging.getLogger(__name__)
-
-DEVICE_OWNER_DVR_INTERFACE = l3_const.DEVICE_OWNER_DVR_INTERFACE
-DEVICE_OWNER_DVR_SNAT = l3_const.DEVICE_OWNER_ROUTER_SNAT
-FLOATINGIP_AGENT_INTF_KEY = l3_const.FLOATINGIP_AGENT_INTF_KEY
-DEVICE_OWNER_AGENT_GW = l3_const.DEVICE_OWNER_AGENT_GW
-SNAT_ROUTER_INTF_KEY = l3_const.SNAT_ROUTER_INTF_KEY
-
-
 router_distributed_opts = [
     cfg.BoolOpt('router_distributed',
                 default=False,
@@ -62,9 +54,9 @@ class L3_NAT_with_dvr_db_mixin(l3_db.L3_NAT_db_mixin,
 
     router_device_owners = (
         l3_db.L3_NAT_db_mixin.router_device_owners +
-        (DEVICE_OWNER_DVR_INTERFACE,
-         DEVICE_OWNER_DVR_SNAT,
-         DEVICE_OWNER_AGENT_GW))
+        (l3_const.DEVICE_OWNER_DVR_INTERFACE,
+         l3_const.DEVICE_OWNER_ROUTER_SNAT,
+         l3_const.DEVICE_OWNER_AGENT_GW))
 
     extra_attributes = (
         l3_attrs_db.ExtraAttributesMixin.extra_attributes + [{
@@ -111,7 +103,7 @@ class L3_NAT_with_dvr_db_mixin(l3_db.L3_NAT_db_mixin,
         """Update the model to support the dvr case of a router."""
         if data.get('distributed'):
             old_owner = l3_const.DEVICE_OWNER_ROUTER_INTF
-            new_owner = DEVICE_OWNER_DVR_INTERFACE
+            new_owner = l3_const.DEVICE_OWNER_DVR_INTERFACE
             for rp in router_db.attached_ports.filter_by(port_type=old_owner):
                 rp.port_type = new_owner
                 rp.port.device_owner = new_owner
@@ -172,7 +164,7 @@ class L3_NAT_with_dvr_db_mixin(l3_db.L3_NAT_db_mixin,
         if router_is_uuid:
             router = self._get_router(context, router)
         if is_distributed_router(router):
-            return DEVICE_OWNER_DVR_INTERFACE
+            return l3_const.DEVICE_OWNER_DVR_INTERFACE
         return super(L3_NAT_with_dvr_db_mixin,
                      self)._get_device_owner(context, router)
 
@@ -315,7 +307,7 @@ class L3_NAT_with_dvr_db_mixin(l3_db.L3_NAT_db_mixin,
 
     def _port_has_ipv6_address(self, port):
         """Overridden to return False if DVR SNAT port."""
-        if port['device_owner'] == DEVICE_OWNER_DVR_SNAT:
+        if port['device_owner'] == l3_const.DEVICE_OWNER_ROUTER_SNAT:
             return False
         return super(L3_NAT_with_dvr_db_mixin,
                      self)._port_has_ipv6_address(port)
@@ -373,7 +365,7 @@ class L3_NAT_with_dvr_db_mixin(l3_db.L3_NAT_db_mixin,
         qry = context.session.query(l3_db.RouterPort)
         qry = qry.filter(
             l3_db.RouterPort.router_id.in_(router_ids),
-            l3_db.RouterPort.port_type == DEVICE_OWNER_DVR_SNAT
+            l3_db.RouterPort.port_type == l3_const.DEVICE_OWNER_ROUTER_SNAT
         )
         interfaces = collections.defaultdict(list)
         for rp in qry:
@@ -420,7 +412,7 @@ class L3_NAT_with_dvr_db_mixin(l3_db.L3_NAT_db_mixin,
             if router['gw_port_id']:
                 snat_router_intfs = snat_intfs_by_router_id[router['id']]
                 LOG.debug("SNAT ports returned: %s ", snat_router_intfs)
-                router[SNAT_ROUTER_INTF_KEY] = snat_router_intfs
+                router[l3_const.SNAT_ROUTER_INTF_KEY] = snat_router_intfs
         return routers_dict
 
     def _process_floating_ips_dvr(self, context, routers_dict,
@@ -449,7 +441,7 @@ class L3_NAT_with_dvr_db_mixin(l3_db.L3_NAT_db_mixin,
         if not fip_agent_id:
             return []
         filters = {'device_id': [fip_agent_id],
-                   'device_owner': [DEVICE_OWNER_AGENT_GW]}
+                   'device_owner': [l3_const.DEVICE_OWNER_AGENT_GW]}
         interfaces = self._core_plugin.get_ports(context.elevated(), filters)
         LOG.debug("Return the FIP ports: %s ", interfaces)
         return interfaces
@@ -477,8 +469,8 @@ class L3_NAT_with_dvr_db_mixin(l3_db.L3_NAT_db_mixin,
                 ports_to_populate.append(router['gw_port'])
             if router.get(l3_const.FLOATINGIP_AGENT_INTF_KEY):
                 ports_to_populate += router[l3_const.FLOATINGIP_AGENT_INTF_KEY]
-            if router.get(SNAT_ROUTER_INTF_KEY):
-                ports_to_populate += router[SNAT_ROUTER_INTF_KEY]
+            if router.get(l3_const.SNAT_ROUTER_INTF_KEY):
+                ports_to_populate += router[l3_const.SNAT_ROUTER_INTF_KEY]
         ports_to_populate += interfaces
         self._populate_subnets_for_ports(context, ports_to_populate)
         self._process_interfaces(routers_dict, interfaces)
@@ -489,7 +481,7 @@ class L3_NAT_with_dvr_db_mixin(l3_db.L3_NAT_db_mixin,
         vm_port_db = port or self._core_plugin.get_port(context, port_id)
         device_owner = vm_port_db['device_owner'] if vm_port_db else ""
         if (n_utils.is_dvr_serviced(device_owner) or
-            device_owner == DEVICE_OWNER_AGENT_GW):
+            device_owner == l3_const.DEVICE_OWNER_AGENT_GW):
             return vm_port_db[portbindings.HOST_ID]
 
     def _get_agent_gw_ports_exist_for_network(
@@ -502,7 +494,7 @@ class L3_NAT_with_dvr_db_mixin(l3_db.L3_NAT_db_mixin,
         filters = {
             'network_id': [network_id],
             'device_id': [agent_id],
-            'device_owner': [DEVICE_OWNER_AGENT_GW]
+            'device_owner': [l3_const.DEVICE_OWNER_AGENT_GW]
         }
         ports = self._core_plugin.get_ports(context, filters)
         if ports:
@@ -537,7 +529,7 @@ class L3_NAT_with_dvr_db_mixin(l3_db.L3_NAT_db_mixin,
         self, context, host_id, ext_net_id):
         """Function to delete FIP gateway port with given ext_net_id."""
         # delete any fip agent gw port
-        device_filter = {'device_owner': [DEVICE_OWNER_AGENT_GW],
+        device_filter = {'device_owner': [l3_const.DEVICE_OWNER_AGENT_GW],
                          'network_id': [ext_net_id]}
         ports = self._core_plugin.get_ports(context,
                                             filters=device_filter)
@@ -567,7 +559,7 @@ class L3_NAT_with_dvr_db_mixin(l3_db.L3_NAT_db_mixin,
                 port_data = {'tenant_id': '',
                              'network_id': network_id,
                              'device_id': l3_agent_db['id'],
-                             'device_owner': DEVICE_OWNER_AGENT_GW,
+                             'device_owner': l3_const.DEVICE_OWNER_AGENT_GW,
                              'binding:host_id': host,
                              'admin_state_up': True,
                              'name': ''}
@@ -587,7 +579,7 @@ class L3_NAT_with_dvr_db_mixin(l3_db.L3_NAT_db_mixin,
         qry = context.session.query(l3_db.RouterPort)
         qry = qry.filter_by(
             router_id=router_id,
-            port_type=DEVICE_OWNER_DVR_SNAT
+            port_type=l3_const.DEVICE_OWNER_ROUTER_SNAT
         )
 
         ports = [self._core_plugin._make_port_dict(rp.port, None)
@@ -601,7 +593,7 @@ class L3_NAT_with_dvr_db_mixin(l3_db.L3_NAT_db_mixin,
                      'network_id': network_id,
                      'fixed_ips': [{'subnet_id': subnet_id}],
                      'device_id': router.id,
-                     'device_owner': DEVICE_OWNER_DVR_SNAT,
+                     'device_owner': l3_const.DEVICE_OWNER_ROUTER_SNAT,
                      'admin_state_up': True,
                      'name': ''}
         snat_port = p_utils.create_port(self._core_plugin, context,
@@ -614,7 +606,7 @@ class L3_NAT_with_dvr_db_mixin(l3_db.L3_NAT_db_mixin,
             router_port = l3_db.RouterPort(
                 port_id=snat_port['id'],
                 router_id=router.id,
-                port_type=DEVICE_OWNER_DVR_SNAT
+                port_type=l3_const.DEVICE_OWNER_ROUTER_SNAT
             )
             context.session.add(router_port)
 
@@ -639,7 +631,7 @@ class L3_NAT_with_dvr_db_mixin(l3_db.L3_NAT_db_mixin,
         int_ports = (
             rp.port for rp in
             router.attached_ports.filter_by(
-                port_type=DEVICE_OWNER_DVR_INTERFACE
+                port_type=l3_const.DEVICE_OWNER_DVR_INTERFACE
             )
         )
         LOG.info(_LI('SNAT interface port list does not exist,'
@@ -674,7 +666,7 @@ class L3_NAT_with_dvr_db_mixin(l3_db.L3_NAT_db_mixin,
         filters = {'fixed_ips': {'subnet_id': [subnet]}}
         ports = self._core_plugin.get_ports(context, filters=filters)
         for port in ports:
-            if port['device_owner'] == DEVICE_OWNER_DVR_INTERFACE:
+            if port['device_owner'] == l3_const.DEVICE_OWNER_DVR_INTERFACE:
                 router_id = port['device_id']
                 router_dict = self._get_router(context, router_id)
                 if router_dict.extra_attributes.distributed:
@@ -698,7 +690,8 @@ class L3_NAT_with_dvr_db_mixin(l3_db.L3_NAT_db_mixin,
         # changeset size since it is late in cycle
         ports = (
             rp.port.id for rp in
-            router.attached_ports.filter_by(port_type=DEVICE_OWNER_DVR_SNAT)
+            router.attached_ports.filter_by(
+                    port_type=l3_const.DEVICE_OWNER_ROUTER_SNAT)
             if rp.port
         )
 

diff --git a/neutron/tests/functional/services/l3_router/test_l3_dvr_router_plugin.py b/neutron/tests/functional/services/l3_router/test_l3_dvr_router_plugin.py
index 473133b182c54a6c8dce2d47594b7800b096daf3..ccf877a04be80be0bff8b8f189e85bd91d3ae712 100644 (file)
--- a/neutron/tests/functional/services/l3_router/test_l3_dvr_router_plugin.py
+++ b/neutron/tests/functional/services/l3_router/test_l3_dvr_router_plugin.py
@@ -14,7 +14,6 @@
 
 from neutron.api.v2 import attributes
 from neutron.common import constants as l3_const
-from neutron.db import l3_dvr_db
 from neutron.extensions import external_net
 from neutron.tests.common import helpers
 from neutron.tests.unit.plugins.ml2 import base as ml2_test_base
@@ -41,13 +40,13 @@ class L3DvrTestCase(ml2_test_base.ML2TestFramework):
     def test_get_device_owner_distributed_router_object(self):
         router = self._create_router()
         self.assertEqual(
-            l3_dvr_db.DEVICE_OWNER_DVR_INTERFACE,
+            l3_const.DEVICE_OWNER_DVR_INTERFACE,
             self.l3_plugin._get_device_owner(self.context, router))
 
     def test_get_device_owner_distributed_router_id(self):
         router = self._create_router()
         self.assertEqual(
-            l3_dvr_db.DEVICE_OWNER_DVR_INTERFACE,
+            l3_const.DEVICE_OWNER_DVR_INTERFACE,
             self.l3_plugin._get_device_owner(self.context, router['id']))
 
     def test_get_device_owner_centralized(self):
@@ -120,7 +119,7 @@ class L3DvrTestCase(ml2_test_base.ML2TestFramework):
                       'mac_address': attributes.ATTR_NOT_SPECIFIED,
                       'fixed_ips': attributes.ATTR_NOT_SPECIFIED,
                       'device_id': self.l3_agent['id'],
-                      'device_owner': l3_dvr_db.DEVICE_OWNER_AGENT_GW,
+                      'device_owner': l3_const.DEVICE_OWNER_AGENT_GW,
                       'binding:host_id': '',
                       'admin_state_up': True,
                       'name': ''}})

diff --git a/neutron/tests/unit/db/test_l3_dvr_db.py b/neutron/tests/unit/db/test_l3_dvr_db.py
index 70786b66e478b933902e1b49282e8793e2035076..aa071a1495f60b94c7dee7495bde7a9a8d9235fe 100644 (file)
--- a/neutron/tests/unit/db/test_l3_dvr_db.py
+++ b/neutron/tests/unit/db/test_l3_dvr_db.py
@@ -144,7 +144,7 @@ class L3DvrTestCase(test_db_base_plugin_v2.NeutronDbPluginV2TestCase):
     def test__get_device_owner_distributed(self):
         self._test_get_device_owner(
             is_distributed=True,
-            expected=l3_dvr_db.DEVICE_OWNER_DVR_INTERFACE,
+            expected=l3_const.DEVICE_OWNER_DVR_INTERFACE,
             pass_router_id=False)
 
     def _test__is_distributed_router(self, router, expected):