openstack
/
neutron
Code Issues Proposed changes

Use DEVICE_OWNER_* for 'network:*' constants 

Now that we have the constant defined, we should reuse it from other
code to avoid potential typos.

Change-Id: Id7a941c1a461264ba44893d97cc6226f092e9888
This commit is contained in:
Gary Kotton 2015-11-15 00:54:38 -08:00
parent e51450a546
commit 50be190b68
20 changed files with 63 additions and 43 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

23
neutron/common/constants.py
View File

@ -31,19 +31,24 @@ FLOATINGIP_STATUS_DOWN = 'DOWN'
FLOATINGIP_STATUS_ERROR = 'ERROR'
DEVICE_OWNER_COMPUTE_PREFIX = "compute:"
DEVICE_OWNER_NETWORK_PREFIX = "network:"
DEVICE_OWNER_ROUTER_HA_INTF = "network:router_ha_interface"
DEVICE_OWNER_ROUTER_INTF = "network:router_interface"
DEVICE_OWNER_ROUTER_GW = "network:router_gateway"
DEVICE_OWNER_FLOATINGIP = "network:floatingip"
DEVICE_OWNER_DHCP = "network:dhcp"
DEVICE_OWNER_DVR_INTERFACE = "network:router_interface_distributed"
DEVICE_OWNER_AGENT_GW = "network:floatingip_agent_gateway"
DEVICE_OWNER_ROUTER_SNAT = "network:router_centralized_snat"
DEVICE_OWNER_ROUTER_HA_INTF = (DEVICE_OWNER_NETWORK_PREFIX +
"router_ha_interface")
DEVICE_OWNER_ROUTER_INTF = DEVICE_OWNER_NETWORK_PREFIX + "router_interface"
DEVICE_OWNER_ROUTER_GW = DEVICE_OWNER_NETWORK_PREFIX + "router_gateway"
DEVICE_OWNER_FLOATINGIP = DEVICE_OWNER_NETWORK_PREFIX + "floatingip"
DEVICE_OWNER_DHCP = DEVICE_OWNER_NETWORK_PREFIX + "dhcp"
DEVICE_OWNER_DVR_INTERFACE = (DEVICE_OWNER_NETWORK_PREFIX +
"router_interface_distributed")
DEVICE_OWNER_AGENT_GW = (DEVICE_OWNER_NETWORK_PREFIX +
"floatingip_agent_gateway")
DEVICE_OWNER_ROUTER_SNAT = (DEVICE_OWNER_NETWORK_PREFIX +
"router_centralized_snat")
DEVICE_OWNER_LOADBALANCER = "neutron:LOADBALANCER"
DEVICE_OWNER_LOADBALANCERV2 = "neutron:LOADBALANCERV2"
DEVICE_OWNER_PREFIXES = ["network:", "neutron:"]
DEVICE_OWNER_PREFIXES = [DEVICE_OWNER_NETWORK_PREFIX, "neutron:"]
# Collection used to identify devices owned by router interfaces.
# DEVICE_OWNER_ROUTER_HA_INTF is a special case and so is not included.

2
neutron/common/utils.py
View File

@ -448,7 +448,7 @@ def is_port_trusted(port):
Trust is currently based on the device_owner field starting with 'network:'
since we restrict who can use that in the default policy.json file.
"""
return port['device_owner'].startswith('network:')
return port['device_owner'].startswith(n_const.DEVICE_OWNER_NETWORK_PREFIX)
class DelayedStringRenderer(object):

3
neutron/db/db_base_plugin_v2.py
View File

@ -1089,7 +1089,8 @@ class NeutronDbPluginV2(db_base_plugin_common.DbBasePluginCommon,
context.session.delete(subnetpool)
def _check_mac_addr_update(self, context, port, new_mac, device_owner):
if (device_owner and device_owner.startswith('network:')):
if (device_owner and
device_owner.startswith(constants.DEVICE_OWNER_NETWORK_PREFIX)):
raise n_exc.UnsupportedPortDeviceOwner(
op=_("mac address update"), port_id=id,
device_owner=device_owner)

3
neutron/plugins/ml2/drivers/openvswitch/agent/ovs_neutron_agent.py
View File

@ -876,7 +876,8 @@ class OVSNeutronAgent(sg_rpc.SecurityGroupAgentRpcCallbackMixin,
LOG.info(_LI("Skipping ARP spoofing rules for port '%s' because "
"it has port security disabled"), vif.port_name)
return
if port_details['device_owner'].startswith('network:'):
if port_details['device_owner'].startswith(
n_const.DEVICE_OWNER_NETWORK_PREFIX):
LOG.debug("Skipping ARP spoofing rules for network owned port "
"'%s'.", vif.port_name)
return

4
neutron/tests/api/test_dhcp_ipv6.py
View File

@ -20,6 +20,7 @@ import six
from tempest_lib.common.utils import data_utils
from tempest_lib import exceptions as lib_exc
from neutron.common import constants
from neutron.tests.api import base
from neutron.tests.tempest import config
from neutron.tests.tempest import test
@ -65,7 +66,8 @@ class NetworksTestDHCPv6(base.BaseNetworkTest):
body = self.client.list_ports()
ports = body['ports']
for port in ports:
if (port['device_owner'].startswith('network:router_interface')
if (port['device_owner'].startswith(
constants.DEVICE_OWNER_ROUTER_INTF)
and port['device_id'] in [r['id'] for r in self.routers]):
self.client.remove_router_interface_with_port_id(
port['device_id'], port['id']

2
neutron/tests/common/l3_test_common.py
View File

@ -31,7 +31,7 @@ def get_ha_interface(ip='169.254.192.1', mac='12:34:56:78:2b:5d'):
subnet_id = _uuid()
return {'admin_state_up': True,
'device_id': _uuid(),
'device_owner': 'network:router_ha_interface',
'device_owner': l3_constants.DEVICE_OWNER_ROUTER_HA_INTF,
'fixed_ips': [{'ip_address': ip,
'prefixlen': 18,
'subnet_id': subnet_id}],

5
neutron/tests/functional/agent/linux/test_linuxbridge_arp_protect.py
View File

@ -13,8 +13,8 @@
# License for the specific language governing permissions and limitations
# under the License.
from neutron.common import constants
from neutron.plugins.ml2.drivers.linuxbridge.agent import arp_protect
from neutron.tests.common import machine_fixtures
from neutron.tests.common import net_helpers
from neutron.tests.functional import base as functional_base
@ -93,7 +93,8 @@ class LinuxBridgeARPSpoofTestCase(functional_base.BaseSudoTestCase):
self._add_arp_protection(self.source, ['1.1.1.1'])
no_arping(self.observer.namespace, self.source.ip)
self._add_arp_protection(self.source, ['1.1.1.1'],
{'device_owner': 'network:router_gateway'})
{'device_owner':
constants.DEVICE_OWNER_ROUTER_GW})
arping(self.observer.namespace, self.source.ip)
def test_arp_protection_dead_reference_removal(self):

4
neutron/tests/functional/agent/test_l3_agent.py
View File

@ -1223,7 +1223,7 @@ class TestDvrRouter(L3AgentTestFramework):
'gateway_ip': float_subnet['gateway_ip'],
'id': fixed_ip['subnet_id']}],
'network_id': external_gw_port['network_id'],
'device_owner': 'network:floatingip_agent_gateway',
'device_owner': l3_constants.DEVICE_OWNER_AGENT_GW,
'mac_address': 'fa:16:3e:80:8d:89',
'binding:host_id': self.agent.conf.host,
'fixed_ips': [{'subnet_id': fixed_ip['subnet_id'],
@ -1253,7 +1253,7 @@ class TestDvrRouter(L3AgentTestFramework):
'gateway_ip': snat_subnet['gateway_ip'],
'id': fixed_ip['subnet_id']}],
'network_id': port['network_id'],
'device_owner': 'network:router_centralized_snat',
'device_owner': l3_constants.DEVICE_OWNER_ROUTER_SNAT,
'mac_address': 'fa:16:3e:80:8d:89',
'fixed_ips': [{'subnet_id': fixed_ip['subnet_id'],
'ip_address': snat_ip,

5
neutron/tests/functional/agent/test_ovs_flows.py
View File

@ -232,8 +232,9 @@ class _ARPSpoofTestCase(object):
# block first and then disable port security to make sure old rules
# are cleared
self._setup_arp_spoof_for_port(self.dst_p.name, ['192.168.0.3'])
self._setup_arp_spoof_for_port(self.dst_p.name, ['192.168.0.3'],
device_owner='network:router_gateway')
self._setup_arp_spoof_for_port(
self.dst_p.name, ['192.168.0.3'],
device_owner=n_const.DEVICE_OWNER_ROUTER_GW)
self.src_p.addr.add('%s/24' % self.src_addr)
self.dst_p.addr.add('%s/24' % self.dst_addr)
net_helpers.assert_ping(self.src_namespace, self.dst_addr, count=2)

2
neutron/tests/unit/agent/dhcp/test_agent.py
View File

@ -104,7 +104,7 @@ fake_port1 = dhcp.DictModel(dict(id='12345678-1234-aaaa-1234567890ab',
fake_dhcp_port = dhcp.DictModel(dict(id='12345678-1234-aaaa-123456789022',
device_id='dhcp-12345678-1234-aaaa-123456789022',
device_owner='network:dhcp',
device_owner=const.DEVICE_OWNER_DHCP,
allocation_pools=fake_subnet1_allocation_pools,
mac_address='aa:bb:cc:dd:ee:22',
network_id='12345678-1234-5678-1234567890ab',

6
neutron/tests/unit/agent/l3/test_agent.py
View File

@ -142,7 +142,8 @@ class BasicRouterOperationsFramework(base.BaseTestCase):
'gateway_ip': '152.2.0.1',
'id': subnet_id_1}],
'network_id': _uuid(),
'device_owner': 'network:router_centralized_snat',
'device_owner':
l3_constants.DEVICE_OWNER_ROUTER_SNAT,
'mac_address': 'fa:16:3e:80:8d:80',
'fixed_ips': [{'subnet_id': subnet_id_1,
'ip_address': '152.2.0.13',
@ -152,7 +153,8 @@ class BasicRouterOperationsFramework(base.BaseTestCase):
'gateway_ip': '152.10.0.1',
'id': subnet_id_2}],
'network_id': _uuid(),
'device_owner': 'network:router_centralized_snat',
'device_owner':
l3_constants.DEVICE_OWNER_ROUTER_SNAT,
'mac_address': 'fa:16:3e:80:8d:80',
'fixed_ips': [{'subnet_id': subnet_id_2,
'ip_address': '152.10.0.13',

10
neutron/tests/unit/agent/l3/test_dvr_local_router.py
View File

@ -124,7 +124,8 @@ class TestDvrRouterOperations(base.BaseTestCase):
'gateway_ip': '152.2.0.1',
'id': subnet_id_1}],
'network_id': _uuid(),
'device_owner': 'network:router_centralized_snat',
'device_owner':
l3_constants.DEVICE_OWNER_ROUTER_SNAT,
'mac_address': 'fa:16:3e:80:8d:80',
'fixed_ips': [{'subnet_id': subnet_id_1,
'ip_address': '152.2.0.13',
@ -134,7 +135,8 @@ class TestDvrRouterOperations(base.BaseTestCase):
'gateway_ip': '152.10.0.1',
'id': subnet_id_2}],
'network_id': _uuid(),
'device_owner': 'network:router_centralized_snat',
'device_owner':
l3_constants.DEVICE_OWNER_ROUTER_SNAT,
'mac_address': 'fa:16:3e:80:8d:80',
'fixed_ips': [{'subnet_id': subnet_id_2,
'ip_address': '152.10.0.13',
@ -333,7 +335,7 @@ class TestDvrRouterOperations(base.BaseTestCase):
ports = ri.router.get(l3_constants.INTERFACE_KEY, [])
subnet_id = l3_test_common.get_subnet_id(ports[0])
test_ports = [{'mac_address': '00:11:22:33:44:55',
'device_owner': 'network:dhcp',
'device_owner': l3_constants.DEVICE_OWNER_DHCP,
'fixed_ips': [{'ip_address': '1.2.3.4',
'prefixlen': 24,
'subnet_id': subnet_id}]}]
@ -469,7 +471,7 @@ class TestDvrRouterOperations(base.BaseTestCase):
'gateway_ip': '20.0.0.1'}],
'id': _uuid(),
'binding:host_id': 'myhost',
'device_owner': 'network:floatingip_agent_gateway',
'device_owner': l3_constants.DEVICE_OWNER_AGENT_GW,
'network_id': fake_network_id,
'mac_address': 'ca:fe:de:ad:be:ef'}]
)

4
neutron/tests/unit/agent/linux/test_dhcp.py
View File

@ -72,7 +72,7 @@ class Dictable(object):
class FakeDhcpPort(object):
id = 'aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaa'
admin_state_up = True
device_owner = 'network:dhcp'
device_owner = constants.DEVICE_OWNER_DHCP
fixed_ips = [FakeIPAllocation('192.168.0.1',
'dddddddd-dddd-dddd-dddd-dddddddddddd')]
mac_address = '00:00:80:aa:bb:ee'
@ -84,7 +84,7 @@ class FakeDhcpPort(object):
class FakeReservedPort(object):
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:80:aa:bb:ee'

6
neutron/tests/unit/agent/test_securitygroups_rpc.py
View File

@ -688,7 +688,7 @@ class SGServerRpcCallBackTestCase(test_sg.SecurityGroupDBTestCase):
self.fmt, n['network']['id'],
fixed_ips=[{'subnet_id': subnet_v6['subnet']['id'],
'ip_address': fake_gateway}],
device_owner='network:router_interface')
device_owner=const.DEVICE_OWNER_ROUTER_INTF)
gateway_mac = gateway_res['port']['mac_address']
gateway_port_id = gateway_res['port']['id']
gateway_lla_ip = str(ipv6.get_ipv6_addr_by_EUI64(
@ -756,7 +756,7 @@ class SGServerRpcCallBackTestCase(test_sg.SecurityGroupDBTestCase):
self.fmt, n['network']['id'],
fixed_ips=[{'subnet_id': subnet_v6['subnet']['id'],
'ip_address': fake_gateway}],
device_owner='network:router_interface')
device_owner=const.DEVICE_OWNER_ROUTER_INTF)
gateway_mac = gateway_res['port']['mac_address']
gateway_port_id = gateway_res['port']['id']
gateway_lla_ip = str(ipv6.get_ipv6_addr_by_EUI64(
@ -766,7 +766,7 @@ class SGServerRpcCallBackTestCase(test_sg.SecurityGroupDBTestCase):
interface_res = self._make_port(
self.fmt, n['network']['id'],
fixed_ips=[{'subnet_id': subnet_v6['subnet']['id']}],
device_owner='network:router_interface')
device_owner=const.DEVICE_OWNER_ROUTER_INTF)
interface_port_id = interface_res['port']['id']
ports_rest1 = self._make_port(

3
neutron/tests/unit/db/test_db_base_plugin_v2.py
View File

@ -5652,7 +5652,8 @@ class NeutronDbPluginV2AsMixinTestCase(NeutronDbPluginV2TestCase,
with self.network() as net, self.network() as net1:
with self.subnet(network=net, cidr='10.0.0.0/24') as subnet,\
self.subnet(network=net1, cidr='10.0.1.0/24') as subnet1:
with self.port(subnet=subnet, device_owner='network:dhcp'),\
with self.port(subnet=subnet,
device_owner=constants.DEVICE_OWNER_DHCP),\
self.port(subnet=subnet1):
# check that user allocations on another network don't
# affect _subnet_get_user_allocation method

5
neutron/tests/unit/notifiers/test_nova.py
View File

@ -75,7 +75,8 @@ class TestNovaNotify(base.BaseTestCase):
port)
def test_port_without_device_id_no_notify(self):
port = models_v2.Port(id='port-uuid', device_owner="network:dhcp",
port = models_v2.Port(id='port-uuid',
device_owner=n_const.DEVICE_OWNER_DHCP,
status=n_const.PORT_STATUS_ACTIVE)
self._record_port_status_changed_helper(n_const.PORT_STATUS_ACTIVE,
sql_attr.NO_VALUE,
@ -93,7 +94,7 @@ class TestNovaNotify(base.BaseTestCase):
def test_non_compute_instances_no_notify(self):
device_id = '32102d7b-1cf4-404d-b50a-97aae1f55f87'
port = models_v2.Port(id='port-uuid', device_id=device_id,
device_owner="network:dhcp",
device_owner=n_const.DEVICE_OWNER_DHCP,
status=n_const.PORT_STATUS_ACTIVE)
self._record_port_status_changed_helper(n_const.PORT_STATUS_ACTIVE,
sql_attr.NO_VALUE,

3
neutron/tests/unit/plugins/ml2/drivers/openvswitch/agent/test_ovs_neutron_agent.py
View File

@ -1469,7 +1469,8 @@ class TestOvsNeutronAgent(object):
def test_arp_spoofing_network_port(self):
int_br = mock.create_autospec(self.agent.int_br)
self.agent.setup_arp_spoofing_protection(
int_br, FakeVif(), {'device_owner': 'network:router_interface'})
int_br, FakeVif(),
{'device_owner': n_const.DEVICE_OWNER_ROUTER_INTF})
self.assertTrue(int_br.delete_arp_spoofing_protection.called)
self.assertFalse(int_br.install_arp_spoofing_protection.called)

2
neutron/tests/unit/plugins/ml2/test_plugin.py
View File

@ -1601,7 +1601,7 @@ class TestFaultyMechansimDriver(Ml2PluginV2FaultyDriverTestCase):
network['network']['tenant_id'],
'name': 'port1',
'device_owner':
'network:router_interface_distributed',
constants.DEVICE_OWNER_DVR_INTERFACE,
'admin_state_up': 1,
'fixed_ips':
[{'subnet_id': subnet_id}]}}

8
neutron/tests/unit/scheduler/test_l3_agent_scheduler.py
View File

@ -1043,7 +1043,7 @@ class L3DvrSchedulerTestCase(testlib_api.SqlTestCase):
{
'id': 'dvr_port1',
'device_id': 'r1',
'device_owner': 'network:router_interface_distributed',
'device_owner': constants.DEVICE_OWNER_DVR_INTERFACE,
'fixed_ips': [
{
'subnet_id': '80947d4a-fbc8-484b-9f92-623a6bfcf3e0',
@ -1054,7 +1054,7 @@ class L3DvrSchedulerTestCase(testlib_api.SqlTestCase):
{
'id': 'dvr_port2',
'device_id': 'r2',
'device_owner': 'network:router_interface_distributed',
'device_owner': constants.DEVICE_OWNER_DVR_INTERFACE,
'fixed_ips': [
{
'subnet_id': '80947d4a-fbc8-484b-9f92-623a6bfcf3e0',
@ -1089,7 +1089,7 @@ class L3DvrSchedulerTestCase(testlib_api.SqlTestCase):
dvr_port = {
'id': 'dvr_port1',
'device_id': 'r1',
'device_owner': 'network:router_interface_distributed',
'device_owner': constants.DEVICE_OWNER_DVR_INTERFACE,
'fixed_ips': [
{
'subnet_id': '80947d4a-fbc8-484b-9f92-623a6bfcf3e0',
@ -1115,7 +1115,7 @@ class L3DvrSchedulerTestCase(testlib_api.SqlTestCase):
dvr_port = {
'id': 'dvr_port1',
'device_id': 'r1',
'device_owner': 'network:router_interface_distributed',
'device_owner': constants.DEVICE_OWNER_DVR_INTERFACE,
'fixed_ips': [
{
'subnet_id': '80947d4a-fbc8-484b-9f92-623a6bfcf3e0',

6
neutron/tests/unit/test_policy.py
View File

@ -327,8 +327,10 @@ class NeutronPolicyTestCase(base.BaseTestCase):
oslo_policy.PolicyNotAuthorized)
def test_create_port_device_owner_regex(self):
blocked_values = ('network:', 'network:abdef', 'network:dhcp',
'network:router_interface')
blocked_values = (const.DEVICE_OWNER_NETWORK_PREFIX,
'network:abdef',
const.DEVICE_OWNER_DHCP,
const.DEVICE_OWNER_ROUTER_INTF)
for val in blocked_values:
self._test_advsvc_action_on_attr(
'create', 'port', 'device_owner', val,