diff --git a/neutron/agent/linux/dhcp.py b/neutron/agent/linux/dhcp.py index 0139205740..eefba2ec70 100644 --- a/neutron/agent/linux/dhcp.py +++ b/neutron/agent/linux/dhcp.py @@ -512,7 +512,7 @@ class Dnsmasq(DhcpLocalProcess): # provides all dnsmasq ip as dns-server if there is more than # one dnsmasq for a subnet and there is no dns-server submitted # by the server - if port.device_owner == 'network:dhcp': + if port.device_owner == constants.DEVICE_OWNER_DHCP: for ip in port.fixed_ips: i = subnet_idx_map.get(ip.subnet_id) if i is None: diff --git a/neutron/agent/metadata/agent.py b/neutron/agent/metadata/agent.py index f7a1c982c9..23f4f27937 100644 --- a/neutron/agent/metadata/agent.py +++ b/neutron/agent/metadata/agent.py @@ -43,8 +43,6 @@ from neutron import wsgi LOG = logging.getLogger(__name__) -DEVICE_OWNER_ROUTER_INTF = "network:router_interface" - class MetadataProxyHandler(object): OPTS = [ @@ -133,7 +131,7 @@ class MetadataProxyHandler(object): else: internal_ports = qclient.list_ports( device_id=router_id, - device_owner=DEVICE_OWNER_ROUTER_INTF)['ports'] + device_owner=n_const.DEVICE_OWNER_ROUTER_INTF)['ports'] networks = [p['network_id'] for p in internal_ports] diff --git a/neutron/db/db_base_plugin_v2.py b/neutron/db/db_base_plugin_v2.py index cd88227fae..5f8a87e2cf 100644 --- a/neutron/db/db_base_plugin_v2.py +++ b/neutron/db/db_base_plugin_v2.py @@ -36,8 +36,6 @@ from neutron.openstack.common import uuidutils LOG = logging.getLogger(__name__) -AGENT_OWNER_PREFIX = 'network:' - # Ports with the following 'device_owner' values will not prevent # network deletion. If delete_network() finds that all ports on a # network have these owners, it will explicitly delete each port @@ -45,7 +43,7 @@ AGENT_OWNER_PREFIX = 'network:' # finds out that all existing IP Allocations are associated with ports # with these owners, it will allow subnet deletion to proceed with the # IP allocations being cleaned up by cascade. -AUTO_DELETE_PORT_OWNERS = ['network:dhcp'] +AUTO_DELETE_PORT_OWNERS = [constants.DEVICE_OWNER_DHCP] class CommonDbMixin(object): diff --git a/neutron/db/dhcp_rpc_base.py b/neutron/db/dhcp_rpc_base.py index b945fec27d..2fce470c09 100644 --- a/neutron/db/dhcp_rpc_base.py +++ b/neutron/db/dhcp_rpc_base.py @@ -191,7 +191,7 @@ class DhcpRpcCallbackMixin(object): tenant_id=network['tenant_id'], mac_address=attributes.ATTR_NOT_SPECIFIED, name='', - device_owner='network:dhcp', + device_owner=constants.DEVICE_OWNER_DHCP, fixed_ips=[dict(subnet_id=s) for s in dhcp_enabled_subnet_ids]) retval = self._port_action(plugin, context, {'port': port_dict}, diff --git a/neutron/plugins/bigswitch/plugin.py b/neutron/plugins/bigswitch/plugin.py index 785e3b0b35..89c4f760e8 100644 --- a/neutron/plugins/bigswitch/plugin.py +++ b/neutron/plugins/bigswitch/plugin.py @@ -214,7 +214,7 @@ class NeutronRestProxyV2Base(db_base_plugin_v2.NeutronDbPluginV2, interfaces = [] mapped_router = self._map_state_and_status(router) router_filter = { - 'device_owner': ["network:router_interface"], + 'device_owner': [const.DEVICE_OWNER_ROUTER_INTF], 'device_id': [router.get('id')] } router_ports = self.get_ports(admin_context, @@ -674,7 +674,7 @@ class NeutronRestProxyV2(NeutronRestProxyV2Base, net = super(NeutronRestProxyV2, self).get_network(context, new_port["network_id"]) if self.add_meta_server_route: - if new_port['device_owner'] == 'network:dhcp': + if new_port['device_owner'] == const.DEVICE_OWNER_DHCP: destination = METADATA_SERVER_IP + '/32' self._add_host_route(context, destination, new_port) diff --git a/neutron/plugins/cisco/n1kv/n1kv_neutron_plugin.py b/neutron/plugins/cisco/n1kv/n1kv_neutron_plugin.py index 6d78b21450..f38db82a4a 100644 --- a/neutron/plugins/cisco/n1kv/n1kv_neutron_plugin.py +++ b/neutron/plugins/cisco/n1kv/n1kv_neutron_plugin.py @@ -24,6 +24,7 @@ import eventlet from neutron.api.rpc.agentnotifiers import dhcp_rpc_agent_api from neutron.api.rpc.agentnotifiers import l3_rpc_agent_api from neutron.api.v2 import attributes +from neutron.common import constants from neutron.common import exceptions as n_exc from neutron.common import rpc as q_rpc from neutron.common import topics @@ -1180,7 +1181,7 @@ class N1kvNeutronPluginV2(db_base_plugin_v2.NeutronDbPluginV2, :returns: port object """ if ('device_id' in port['port'] and port['port']['device_owner'] in - ['network:dhcp', 'network:router_interface']): + [constants.DEVICE_OWNER_DHCP, constants.DEVICE_OWNER_ROUTER_INTF]): p_profile_name = c_conf.CISCO_N1K.network_node_policy_profile p_profile = self._get_policy_profile_by_name(p_profile_name) if p_profile: diff --git a/neutron/plugins/embrane/common/utils.py b/neutron/plugins/embrane/common/utils.py index 8e4a92abab..5fa20eb59d 100644 --- a/neutron/plugins/embrane/common/utils.py +++ b/neutron/plugins/embrane/common/utils.py @@ -19,6 +19,7 @@ from heleosapi import info as h_info +from neutron.common import constants from neutron.db import models_v2 from neutron.openstack.common import log as logging @@ -47,7 +48,8 @@ def retrieve_ip_allocation_info(context, neutron_port): return subnet = retrieve_subnet(context, subnet_id) allocated_ip = neutron_port["fixed_ips"][0]["ip_address"] - is_gw_port = neutron_port["device_owner"] == "network:router_gateway" + is_gw_port = (neutron_port["device_owner"] == + constants.DEVICE_OWNER_ROUTER_GW) gateway_ip = subnet["gateway_ip"] ip_allocation_info = h_info.IpAllocationInfo( diff --git a/neutron/plugins/embrane/l2base/fake/fakeplugin_support.py b/neutron/plugins/embrane/l2base/fake/fakeplugin_support.py index 51a4d0b733..7818d28de4 100644 --- a/neutron/plugins/embrane/l2base/fake/fakeplugin_support.py +++ b/neutron/plugins/embrane/l2base/fake/fakeplugin_support.py @@ -19,6 +19,7 @@ from heleosapi import info as h_info +from neutron.common import constants from neutron import manager from neutron.plugins.embrane.l2base import support_base as base @@ -32,7 +33,8 @@ class FakePluginSupport(base.SupportBase): plugin = manager.NeutronManager.get_plugin() network_id = neutron_port["network_id"] network = plugin._get_network(context, network_id) - is_gw = neutron_port["device_owner"] == "network:router_gateway" + is_gw = (neutron_port["device_owner"] == + constants.DEVICE_OWNER_ROUTER_GW) result = h_info.UtifInfo(vlan=0, network_name=network["name"], network_id=network["id"], diff --git a/neutron/plugins/embrane/l2base/openvswitch/openvswitch_support.py b/neutron/plugins/embrane/l2base/openvswitch/openvswitch_support.py index d342abaef3..f37a6b81aa 100644 --- a/neutron/plugins/embrane/l2base/openvswitch/openvswitch_support.py +++ b/neutron/plugins/embrane/l2base/openvswitch/openvswitch_support.py @@ -19,6 +19,7 @@ from heleosapi import info as h_info +from neutron.common import constants from neutron import manager from neutron.plugins.embrane.l2base import support_base as base from neutron.plugins.embrane.l2base import support_exceptions as exc @@ -45,7 +46,8 @@ class OpenvswitchSupport(base.SupportBase): err_msg=_("No segmentation_id found for the network, " "please be sure that tenant_network_type is vlan")) network = plugin._get_network(context, network_id) - is_gw = neutron_port["device_owner"] == "network:router_gateway" + is_gw = (neutron_port["device_owner"] == + constants.DEVICE_OWNER_ROUTER_GW) result = h_info.UtifInfo(vlan=network_binding["segmentation_id"], network_name=network["name"], network_id=network["id"], diff --git a/neutron/plugins/midonet/plugin.py b/neutron/plugins/midonet/plugin.py index c7190b402d..2db6287040 100644 --- a/neutron/plugins/midonet/plugin.py +++ b/neutron/plugins/midonet/plugin.py @@ -163,7 +163,7 @@ def _is_vif_port(port): def _is_dhcp_port(port): """Check whether the given port is a DHCP port.""" device_owner = port['device_owner'] - return device_owner.startswith('network:dhcp') + return device_owner.startswith(constants.DEVICE_OWNER_DHCP) def _check_resource_exists(func, id, name, raise_exc=False): @@ -1024,7 +1024,7 @@ class MidonetPluginV2(db_base_plugin_v2.NeutronDbPluginV2, rport_qry = context.session.query(models_v2.Port) dhcp_ports = rport_qry.filter_by( network_id=subnet["network_id"], - device_owner='network:dhcp').all() + device_owner=constants.DEVICE_OWNER_DHCP).all() if dhcp_ports and dhcp_ports[0].fixed_ips: metadata_gw_ip = dhcp_ports[0].fixed_ips[0].ip_address else: diff --git a/neutron/plugins/plumgrid/plumgrid_plugin/plumgrid_plugin.py b/neutron/plugins/plumgrid/plumgrid_plugin/plumgrid_plugin.py index 1c23dccb71..12f8e31fa3 100644 --- a/neutron/plugins/plumgrid/plumgrid_plugin/plumgrid_plugin.py +++ b/neutron/plugins/plumgrid/plumgrid_plugin/plumgrid_plugin.py @@ -25,6 +25,7 @@ import netaddr from oslo.config import cfg from neutron.api.v2 import attributes +from neutron.common import constants from neutron.db import db_base_plugin_v2 from neutron.db import external_net_db from neutron.db import l3_db @@ -184,7 +185,7 @@ class NeutronPluginPLUMgridV2(db_base_plugin_v2.NeutronDbPluginV2, port_db = super(NeutronPluginPLUMgridV2, self).create_port(context, port) device_id = port_db["device_id"] - if port_db["device_owner"] == "network:router_gateway": + if port_db["device_owner"] == constants.DEVICE_OWNER_ROUTER_GW: router_db = self._get_router(context, device_id) else: router_db = None @@ -212,7 +213,7 @@ class NeutronPluginPLUMgridV2(db_base_plugin_v2.NeutronDbPluginV2, port_db = super(NeutronPluginPLUMgridV2, self).update_port( context, port_id, port) device_id = port_db["device_id"] - if port_db["device_owner"] == "network:router_gateway": + if port_db["device_owner"] == constants.DEVICE_OWNER_ROUTER_GW: router_db = self._get_router(context, device_id) else: router_db = None @@ -242,7 +243,7 @@ class NeutronPluginPLUMgridV2(db_base_plugin_v2.NeutronDbPluginV2, self.disassociate_floatingips(context, port_id) super(NeutronPluginPLUMgridV2, self).delete_port(context, port_id) - if port_db["device_owner"] == "network:router_gateway": + if port_db["device_owner"] == constants.DEVICE_OWNER_ROUTER_GW: device_id = port_db["device_id"] router_db = self._get_router(context, device_id) else: diff --git a/neutron/plugins/vmware/plugins/base.py b/neutron/plugins/vmware/plugins/base.py index 08d725229b..2e18d28083 100644 --- a/neutron/plugins/vmware/plugins/base.py +++ b/neutron/plugins/vmware/plugins/base.py @@ -1028,7 +1028,7 @@ class NsxPluginV2(addr_pair_db.AllowedAddressPairsMixin, # Before deleting ports, ensure the peer of a NSX logical # port with a patch attachment is removed too port_filter = {'network_id': [id], - 'device_owner': ['network:router_interface']} + 'device_owner': [constants.DEVICE_OWNER_ROUTER_INTF]} router_iface_ports = self.get_ports(context, filters=port_filter) for port in router_iface_ports: nsx_switch_id, nsx_port_id = nsx_utils.get_nsx_switch_and_port_id( diff --git a/neutron/tests/unit/nec/test_nec_plugin.py b/neutron/tests/unit/nec/test_nec_plugin.py index a0e27da394..310c344a2b 100644 --- a/neutron/tests/unit/nec/test_nec_plugin.py +++ b/neutron/tests/unit/nec/test_nec_plugin.py @@ -585,11 +585,12 @@ class TestNecPluginOfcManager(NecPluginV2TestCase): with self.network() as network: with self.subnet(network=network): net = network['network'] - p = self._create_resource('port', - {'network_id': net['id'], - 'tenant_id': net['tenant_id'], - 'device_owner': 'network:dhcp', - 'device_id': 'dhcp-port1'}) + p = self._create_resource( + 'port', + {'network_id': net['id'], + 'tenant_id': net['tenant_id'], + 'device_owner': constants.DEVICE_OWNER_DHCP, + 'device_id': 'dhcp-port1'}) # Make sure that the port is created on OFC. portinfo = {'id': p['id'], 'port_no': 123} self.rpcapi_update_ports(added=[portinfo]) diff --git a/neutron/tests/unit/test_db_plugin.py b/neutron/tests/unit/test_db_plugin.py index cc33abbba6..a23a3a7bc2 100644 --- a/neutron/tests/unit/test_db_plugin.py +++ b/neutron/tests/unit/test_db_plugin.py @@ -31,6 +31,7 @@ from neutron.api.v2 import attributes from neutron.api.v2.attributes import ATTR_NOT_SPECIFIED from neutron.api.v2.router import APIRouter from neutron.common import config +from neutron.common import constants from neutron.common import exceptions as n_exc from neutron.common.test_lib import test_config from neutron import context @@ -1132,7 +1133,8 @@ fixed_ips=ip_address%%3D%s&fixed_ips=ip_address%%3D%s&fixed_ips=subnet_id%%3D%s admin_state_up=True) network = self.deserialize(self.fmt, res) network_id = network['network']['id'] - self._create_port(self.fmt, network_id, device_owner='network:dhcp') + self._create_port(self.fmt, network_id, + device_owner=constants.DEVICE_OWNER_DHCP) req = self.new_delete_request('networks', network_id) res = req.get_response(self.api) self.assertEqual(res.status_int, webob.exc.HTTPNoContent.code) @@ -2398,7 +2400,7 @@ class TestSubnetsV2(NeutronDbPluginV2TestCase): cidr, ip_version=4) self._create_port(self.fmt, network['network']['id'], - device_owner='network:dhcp') + device_owner=constants.DEVICE_OWNER_DHCP) req = self.new_delete_request('subnets', subnet['subnet']['id']) res = req.get_response(self.api) self.assertEqual(res.status_int, webob.exc.HTTPNoContent.code) diff --git a/neutron/tests/unit/test_db_rpc_base.py b/neutron/tests/unit/test_db_rpc_base.py index 69180e19b8..7de3ece8ab 100644 --- a/neutron/tests/unit/test_db_rpc_base.py +++ b/neutron/tests/unit/test_db_rpc_base.py @@ -15,6 +15,7 @@ import mock +from neutron.common import constants from neutron.common import exceptions as n_exc from neutron.db import dhcp_rpc_base from neutron.openstack.common.db import exception as db_exc @@ -54,7 +55,7 @@ class TestDhcpRpcCallbackMixin(base.BaseTestCase): def _test__port_action_with_failures(self, exc=None, action=None): port = { 'network_id': 'foo_network_id', - 'device_owner': 'network:dhcp', + 'device_owner': constants.DEVICE_OWNER_DHCP, 'fixed_ips': [{'subnet_id': 'foo_subnet_id'}] } self.plugin.create_port.side_effect = exc @@ -187,7 +188,7 @@ class TestDhcpRpcCallbackMixin(base.BaseTestCase): create_spec = dict(tenant_id='tenantid', device_id='devid', network_id='netid', name='', admin_state_up=True, - device_owner='network:dhcp', + device_owner=constants.DEVICE_OWNER_DHCP, mac_address=mock.ANY) create_retval = create_spec.copy() create_retval['id'] = 'port_id' diff --git a/neutron/tests/unit/test_extension_extraroute.py b/neutron/tests/unit/test_extension_extraroute.py index 539159a7e2..48ce3f782b 100644 --- a/neutron/tests/unit/test_extension_extraroute.py +++ b/neutron/tests/unit/test_extension_extraroute.py @@ -19,6 +19,7 @@ import contextlib from oslo.config import cfg from webob import exc +from neutron.common import constants from neutron.db import extraroute_db from neutron.extensions import extraroute from neutron.extensions import l3 @@ -392,7 +393,6 @@ class ExtraRouteDBTestCaseBase(object): p['port']['id']) def test_router_update_on_external_port(self): - DEVICE_OWNER_ROUTER_GW = "network:router_gateway" with self.router() as r: with self.subnet(cidr='10.0.1.0/24') as s: self._set_net_external(s['subnet']['network_id']) @@ -402,11 +402,12 @@ class ExtraRouteDBTestCaseBase(object): body = self._show('routers', r['router']['id']) net_id = body['router']['external_gateway_info']['network_id'] self.assertEqual(net_id, s['subnet']['network_id']) - port_res = self._list_ports('json', - 200, - s['subnet']['network_id'], - tenant_id=r['router']['tenant_id'], - device_own=DEVICE_OWNER_ROUTER_GW) + port_res = self._list_ports( + 'json', + 200, + s['subnet']['network_id'], + tenant_id=r['router']['tenant_id'], + device_own=constants.DEVICE_OWNER_ROUTER_GW) port_list = self.deserialize('json', port_res) self.assertEqual(len(port_list['ports']), 1) diff --git a/neutron/tests/unit/test_l3_plugin.py b/neutron/tests/unit/test_l3_plugin.py index 5360fe5295..a29ade31e2 100644 --- a/neutron/tests/unit/test_l3_plugin.py +++ b/neutron/tests/unit/test_l3_plugin.py @@ -1146,7 +1146,8 @@ class L3NatTestCaseBase(L3NatTestCaseMixin): port_id=private_port['port']['id']) self.assertEqual(res.status_int, 400) for p in self._list('ports')['ports']: - if p['device_owner'] == 'network:floatingip': + if (p['device_owner'] == + l3_constants.DEVICE_OWNER_FLOATINGIP): self.fail('garbage port is not deleted') self._remove_external_gateway_from_router( r['router']['id'], @@ -1184,7 +1185,8 @@ class L3NatTestCaseBase(L3NatTestCaseMixin): self.assertEqual(res.status_int, exc.HTTPConflict.code) for p in self._list('ports')['ports']: - if p['device_owner'] == 'network:floatingip': + if (p['device_owner'] == + l3_constants.DEVICE_OWNER_FLOATINGIP): self.fail('garbage port is not deleted') self._remove_external_gateway_from_router( @@ -1341,7 +1343,7 @@ class L3NatTestCaseBase(L3NatTestCaseMixin): found = False with self.floatingip_with_assoc(): for p in self._list('ports')['ports']: - if p['device_owner'] == 'network:floatingip': + if p['device_owner'] == l3_constants.DEVICE_OWNER_FLOATINGIP: self._delete('ports', p['id'], expected_code=exc.HTTPConflict.code) found = True @@ -1524,7 +1526,7 @@ class L3NatTestCaseBase(L3NatTestCaseMixin): found = False with self.floatingip_with_assoc(): for p in self._list('ports')['ports']: - if p['device_owner'] == 'network:router_interface': + if p['device_owner'] == l3_constants.DEVICE_OWNER_ROUTER_INTF: subnet_id = p['fixed_ips'][0]['subnet_id'] router_id = p['device_id'] self._router_interface_action( @@ -1538,7 +1540,7 @@ class L3NatTestCaseBase(L3NatTestCaseMixin): found = False with self.floatingip_with_assoc(): for p in self._list('ports')['ports']: - if p['device_owner'] == 'network:router_interface': + if p['device_owner'] == l3_constants.DEVICE_OWNER_ROUTER_INTF: router_id = p['device_id'] self._router_interface_action( 'remove', router_id, None, p['id'], diff --git a/neutron/tests/unit/test_linux_dhcp.py b/neutron/tests/unit/test_linux_dhcp.py index 1baa17c595..9a6c7e19f6 100644 --- a/neutron/tests/unit/test_linux_dhcp.py +++ b/neutron/tests/unit/test_linux_dhcp.py @@ -23,6 +23,7 @@ from oslo.config import cfg from neutron.agent.common import config from neutron.agent.linux import dhcp from neutron.common import config as base_config +from neutron.common import constants from neutron.openstack.common import log as logging from neutron.tests import base @@ -80,7 +81,7 @@ class FakePort3: class FakeRouterPort: id = 'rrrrrrrr-rrrr-rrrr-rrrr-rrrrrrrrrrrr' admin_state_up = True - device_owner = 'network:router_interface' + device_owner = constants.DEVICE_OWNER_ROUTER_INTF fixed_ips = [FakeIPAllocation('192.168.0.1', 'dddddddd-dddd-dddd-dddd-dddddddddddd')] mac_address = '00:00:0f:rr:rr:rr' @@ -92,7 +93,7 @@ class FakeRouterPort: class FakePortMultipleAgents1: id = 'rrrrrrrr-rrrr-rrrr-rrrr-rrrrrrrrrrrr' admin_state_up = True - device_owner = 'network:dhcp' + device_owner = constants.DEVICE_OWNER_DHCP fixed_ips = [FakeIPAllocation('192.168.0.5', 'dddddddd-dddd-dddd-dddd-dddddddddddd')] mac_address = '00:00:0f:dd:dd:dd' @@ -104,7 +105,7 @@ class FakePortMultipleAgents1: class FakePortMultipleAgents2: id = 'aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa' admin_state_up = True - device_owner = 'network:dhcp' + device_owner = constants.DEVICE_OWNER_DHCP fixed_ips = [FakeIPAllocation('192.168.0.6', 'dddddddd-dddd-dddd-dddd-dddddddddddd')] mac_address = '00:00:0f:ee:ee:ee' diff --git a/neutron/tests/unit/test_metadata_agent.py b/neutron/tests/unit/test_metadata_agent.py index 68cb2b9a28..9cba5ddda6 100644 --- a/neutron/tests/unit/test_metadata_agent.py +++ b/neutron/tests/unit/test_metadata_agent.py @@ -23,6 +23,7 @@ import testtools import webob from neutron.agent.metadata import agent +from neutron.common import constants from neutron.common import utils from neutron.tests import base @@ -112,7 +113,7 @@ class TestMetadataProxyHandler(base.BaseTestCase): expected.append( mock.call().list_ports( device_id=router_id, - device_owner='network:router_interface' + device_owner=constants.DEVICE_OWNER_ROUTER_INTF ) ) diff --git a/neutron/tests/unit/vmware/test_dhcpmeta.py b/neutron/tests/unit/vmware/test_dhcpmeta.py index 160aee6eea..84ddb184cd 100644 --- a/neutron/tests/unit/vmware/test_dhcpmeta.py +++ b/neutron/tests/unit/vmware/test_dhcpmeta.py @@ -17,6 +17,7 @@ import mock from oslo.config import cfg +from neutron.common import constants as n_consts from neutron.common import exceptions as n_exc from neutron import context from neutron.db import api as db @@ -940,12 +941,12 @@ class DhcpAgentNotifyAPITestCase(base.BaseTestCase): def test_notify_ports_update_with_special_ports(self): ports = [{'fixed_ips': [], 'device_id': '', - 'device_owner': 'network:dhcp', + 'device_owner': n_consts.DEVICE_OWNER_DHCP, 'mac_address': 'fa:16:3e:da:1d:46'}, {'fixed_ips': [{'subnet_id': 'foo_subnet_id', 'ip_address': '1.2.3.4'}], 'device_id': 'foo_device_id', - 'device_owner': 'network:router_gateway', + 'device_owner': n_consts.DEVICE_OWNER_ROUTER_GW, 'mac_address': 'fa:16:3e:da:1d:46'}] call_args = mock.call( mock.ANY, 'foo_network_id', 'foo_subnet_id', dhcp=[], meta=[]) @@ -1014,7 +1015,7 @@ class DhcpAgentNotifyAPITestCase(base.BaseTestCase): 'admin_state_up': True, 'network_id': 'foo_network_id', 'tenant_id': 'foo_tenant_id', - 'device_owner': 'network:dhcp', + 'device_owner': n_consts.DEVICE_OWNER_DHCP, 'mac_address': mock.ANY, 'fixed_ips': [{'subnet_id': 'foo_subnet_id'}], 'device_id': '' @@ -1102,7 +1103,7 @@ class DhcpAgentNotifyAPITestCase(base.BaseTestCase): self.notifier.notify(mock.ANY, {'subnet': subnet}, 'subnet.delete.end') filters = { 'network_id': [subnet['network_id']], - 'device_owner': ['network:dhcp'] + 'device_owner': [n_consts.DEVICE_OWNER_DHCP] } self.plugin.get_ports.assert_called_once_with( mock.ANY, filters=filters) @@ -1149,7 +1150,7 @@ class DhcpTestCase(base.BaseTestCase): } port = { 'id': 'foo_port_id', - 'device_owner': 'network:dhcp', + 'device_owner': n_consts.DEVICE_OWNER_DHCP, 'mac_address': 'aa:bb:cc:dd:ee:ff', 'network_id': 'foo_network_id', 'fixed_ips': [{'subnet_id': subnet['id']}] @@ -1184,7 +1185,7 @@ class DhcpTestCase(base.BaseTestCase): def test_handle_delete_dhcp_owner_port(self): port = { 'id': 'foo_port_id', - 'device_owner': 'network:dhcp', + 'device_owner': n_consts.DEVICE_OWNER_DHCP, 'network_id': 'foo_network_id', 'fixed_ips': [], 'mac_address': 'aa:bb:cc:dd:ee:ff' @@ -1297,15 +1298,15 @@ class MetadataTestCase(base.BaseTestCase): def test_handle_port_metadata_access_dhcp_port(self): self._test_handle_port_metadata_access_special_owners( - 'network:dhcp', [{'subnet_id': 'foo_subnet'}]) + n_consts.DEVICE_OWNER_DHCP, [{'subnet_id': 'foo_subnet'}]) def test_handle_port_metadata_access_router_port(self): self._test_handle_port_metadata_access_special_owners( - 'network:router_interface', [{'subnet_id': 'foo_subnet'}]) + n_consts.DEVICE_OWNER_ROUTER_INTF, [{'subnet_id': 'foo_subnet'}]) def test_handle_port_metadata_access_no_device_id(self): self._test_handle_port_metadata_access_special_owners( - 'network:dhcp', '') + n_consts.DEVICE_OWNER_DHCP, '') def test_handle_port_metadata_access_no_fixed_ips(self): self._test_handle_port_metadata_access_special_owners( diff --git a/neutron/tests/unit/vmware/test_nsx_plugin.py b/neutron/tests/unit/vmware/test_nsx_plugin.py index 1464b325d5..c40bd47251 100644 --- a/neutron/tests/unit/vmware/test_nsx_plugin.py +++ b/neutron/tests/unit/vmware/test_nsx_plugin.py @@ -221,7 +221,7 @@ class TestPortsV2(NsxPluginV2TestCase, with mock.patch.object(nsx_db, 'add_neutron_nsx_port_mapping', side_effect=db_exception): with self.network() as net: - with self.port(device_owner='network:dhcp'): + with self.port(device_owner=constants.DEVICE_OWNER_DHCP): self._verify_no_orphan_left(net['network']['id']) def test_create_port_maintenance_returns_503(self): @@ -909,7 +909,7 @@ class TestL3NatTestCase(L3NatTest, subnets = self._list('subnets')['subnets'] with self.subnet() as s: with self.port(subnet=s, device_id='1234', - device_owner='network:dhcp'): + device_owner=constants.DEVICE_OWNER_DHCP): subnets = self._list('subnets')['subnets'] self.assertEqual(len(subnets), 1) self.assertEqual(subnets[0]['host_routes'][0]['nexthop'],