openstack
/
neutron
Code Issues Proposed changes

Use method get_ipv6_addr_by_EUI64 from oslo.utils 

Oslo.utils provides same method get_ipv6_addr_by_EUI64, let's use
it and deprecate method get_ipv6_addr_by_EUI64 in neutron.

Closes-Bug: #1597213
Change-Id: If1eda1283b4405e13ccb60dab530ce4cc1b22948
This commit is contained in:
ChangBo Guo(gcb) 2016-06-17 16:14:31 +08:00
parent c52809fce1
commit 7913798a2c
14 changed files with 53 additions and 45 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
neutron/agent/linux/iptables_firewall.py
View File

@ -20,6 +20,7 @@ import netaddr
from neutron_lib import constants
from oslo_config import cfg
from oslo_log import log as logging
from oslo_utils import netutils
import six
from neutron._i18n import _LI
@ -386,7 +387,7 @@ class IptablesFirewallDriver(firewall.FirewallDriver):
mac_ipv4_pairs.append((mac, ip_address))
else:
mac_ipv6_pairs.append((mac, ip_address))
lla = str(ipv6_utils.get_ipv6_addr_by_EUI64(
lla = str(netutils.get_ipv6_addr_by_EUI64(
constants.IPv6_LLA_PREFIX, mac))
mac_ipv6_pairs.append((mac, lla))

4
neutron/agent/linux/openvswitch_firewall/firewall.py
View File

@ -17,13 +17,13 @@ import netaddr
from neutron_lib import constants as lib_const
from neutron_lib import exceptions
from oslo_log import log as logging
from oslo_utils import netutils
from neutron._i18n import _, _LE, _LW
from neutron.agent import firewall
from neutron.agent.linux.openvswitch_firewall import constants as ovsfw_consts
from neutron.agent.linux.openvswitch_firewall import rules
from neutron.common import constants
from neutron.common import ipv6_utils
from neutron.plugins.ml2.drivers.openvswitch.agent.common import constants \
as ovs_consts
@ -86,7 +86,7 @@ class OFPort(object):
self.id = port_dict['device']
self.vlan_tag = vlan_tag
self.mac = ovs_port.vif_mac
self.lla_address = str(ipv6_utils.get_ipv6_addr_by_EUI64(
self.lla_address = str(netutils.get_ipv6_addr_by_EUI64(
lib_const.IPv6_LLA_PREFIX, self.mac))
self.ofport = ovs_port.ofport
self.sec_groups = list()

6
neutron/agent/linux/pd.py
View File

@ -20,6 +20,7 @@ import eventlet
from neutron_lib import constants as n_const
from oslo_config import cfg
from oslo_log import log as logging
from oslo_utils import netutils
import six
from stevedore import driver
@ -28,7 +29,6 @@ from neutron.callbacks import events
from neutron.callbacks import registry
from neutron.callbacks import resources
from neutron.common import constants as l3_constants
from neutron.common import ipv6_utils
from neutron.common import utils
LOG = logging.getLogger(__name__)
@ -175,8 +175,8 @@ class PrefixDelegation(object):
@staticmethod
def _get_lla(mac):
lla = ipv6_utils.get_ipv6_addr_by_EUI64(n_const.IPv6_LLA_PREFIX,
mac)
lla = netutils.get_ipv6_addr_by_EUI64(n_const.IPv6_LLA_PREFIX,
mac)
return lla
def _get_llas(self, gw_ifname, ns_name):

5
neutron/common/ipv6_utils.py
View File

@ -18,6 +18,7 @@ IPv6-related utilities and helper functions.
"""
import os
from debtcollector import removals
import netaddr
from neutron_lib import constants as const
from oslo_log import log
@ -29,6 +30,10 @@ LOG = log.getLogger(__name__)
_IS_IPV6_ENABLED = None
@removals.remove(
message="use get_ipv6_addr_by_EUI64 from oslo_utils.netutils",
version="Newton",
removal_version="Ocata")
def get_ipv6_addr_by_EUI64(prefix, mac):
# Check if the prefix is IPv4 address
isIPv4 = netaddr.valid_ipv4(prefix)

3
neutron/db/ipam_non_pluggable_backend.py
View File

@ -22,6 +22,7 @@ from neutron_lib import constants
from neutron_lib import exceptions as n_exc
from oslo_db import exception as db_exc
from oslo_log import log as logging
from oslo_utils import netutils
from sqlalchemy import and_
from sqlalchemy.orm import exc
@ -358,7 +359,7 @@ class IpamNonPluggableBackend(ipam_backend_mixin.IpamBackendMixin):
def _calculate_ipv6_eui64_addr(self, context, subnet, mac_addr):
prefix = subnet['cidr']
network_id = subnet['network_id']
ip_address = ipv6_utils.get_ipv6_addr_by_EUI64(
ip_address = netutils.get_ipv6_addr_by_EUI64(
prefix, mac_addr).format()
if not self._check_unique_ip(context, network_id,
subnet['id'], ip_address):

6
neutron/db/securitygroups_rpc_base.py
View File

@ -16,10 +16,10 @@
import netaddr
from neutron_lib import constants as const
from oslo_log import log as logging
from oslo_utils import netutils
from sqlalchemy.orm import exc
from neutron._i18n import _, _LW
from neutron.common import ipv6_utils as ipv6
from neutron.common import utils
from neutron.db.models import allowed_address_pair as aap_models
from neutron.db.models import securitygroup as sg_models
@ -319,7 +319,7 @@ class SecurityGroupServerRpcMixin(sg_db.SecurityGroupDbMixin):
for mac_address, network_id, ip in query:
if (netaddr.IPAddress(ip).version == 6
and not netaddr.IPAddress(ip).is_link_local()):
ip = str(ipv6.get_ipv6_addr_by_EUI64(const.IPv6_LLA_PREFIX,
ip = str(netutils.get_ipv6_addr_by_EUI64(const.IPv6_LLA_PREFIX,
mac_address))
if ip not in ips[network_id]:
ips[network_id].append(ip)
@ -379,7 +379,7 @@ class SecurityGroupServerRpcMixin(sg_db.SecurityGroupDbMixin):
LOG.warning(_LW('No valid gateway port on subnet %s is '
'found for IPv6 RA'), subnet['id'])
return
lla_ip = str(ipv6.get_ipv6_addr_by_EUI64(
lla_ip = str(netutils.get_ipv6_addr_by_EUI64(
const.IPv6_LLA_PREFIX,
mac_address))
return lla_ip

4
neutron/ipam/requests.py
View File

@ -15,11 +15,11 @@ import abc
import netaddr
from neutron_lib.api import validators
from neutron_lib import constants
from oslo_utils import netutils
from oslo_utils import uuidutils
import six
from neutron._i18n import _
from neutron.common import ipv6_utils
from neutron.common import utils as common_utils
from neutron.ipam import exceptions as ipam_exc
@ -221,7 +221,7 @@ class AutomaticAddressRequest(SpecificAddressRequest):
reason=_('must provide exactly 2 arguments - cidr and MAC'))
prefix = kwargs['prefix']
mac_address = kwargs['mac']
return ipv6_utils.get_ipv6_addr_by_EUI64(prefix, mac_address)
return netutils.get_ipv6_addr_by_EUI64(prefix, mac_address)
_address_generators = {EUI64: _generate_eui64_address}

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

@ -29,6 +29,7 @@ from oslo_log import log as logging
import oslo_messaging
from oslo_service import loopingcall
from oslo_service import systemd
from oslo_utils import netutils
from osprofiler import profiler
import six
from six import moves
@ -45,7 +46,6 @@ from neutron.api.rpc.callbacks import resources
from neutron.api.rpc.handlers import dvr_rpc
from neutron.common import config
from neutron.common import constants as c_const
from neutron.common import ipv6_utils as ipv6
from neutron.common import topics
from neutron.common import utils as n_utils
from neutron import context
@ -924,7 +924,7 @@ class OVSNeutronAgent(sg_rpc.SecurityGroupAgentRpcCallbackMixin,
ipv6_addresses = {ip for ip in addresses
if netaddr.IPNetwork(ip).version == 6}
# Allow neighbor advertisements for LLA address.
ipv6_addresses |= {str(ipv6.get_ipv6_addr_by_EUI64(
ipv6_addresses |= {str(netutils.get_ipv6_addr_by_EUI64(
n_const.IPv6_LLA_PREFIX, mac))
for mac in mac_addresses}
if not has_zero_prefixlen_address(ipv6_addresses):

6
neutron/tests/tools.py
View File

@ -26,12 +26,12 @@ import fixtures
import mock
import netaddr
from neutron_lib import constants
from oslo_utils import netutils
import six
import unittest2
from neutron.api.v2 import attributes
from neutron.common import constants as n_const
from neutron.common import ipv6_utils
from neutron.common import utils
from neutron.db import common_db_mixin
@ -259,8 +259,8 @@ def get_random_ip_address(version=4):
random.randint(3, 254))
return netaddr.IPAddress(ip_string)
else:
ip = ipv6_utils.get_ipv6_addr_by_EUI64('2001:db8::/64',
get_random_mac())
ip = netutils.get_ipv6_addr_by_EUI64('2001:db8::/64',
get_random_mac())
return ip

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

@ -20,6 +20,7 @@ import mock
from neutron_lib import constants as const
from oslo_config import cfg
import oslo_messaging
from oslo_utils import netutils
from testtools import matchers
import webob.exc
@ -27,7 +28,6 @@ from neutron.agent import firewall as firewall_base
from neutron.agent.linux import iptables_manager
from neutron.agent import securitygroups_rpc as sg_rpc
from neutron.api.rpc.handlers import securitygroups_rpc
from neutron.common import ipv6_utils as ipv6
from neutron.common import rpc as n_rpc
from neutron import context
from neutron.db import securitygroups_rpc_base as sg_db_rpc
@ -578,7 +578,7 @@ class SGServerRpcCallBackTestCase(test_sg.SecurityGroupDBTestCase):
security_groups=[sg1_id])
dhcp_rest = self.deserialize(self.fmt, dhcp_port)
dhcp_mac = dhcp_rest['port']['mac_address']
dhcp_lla_ip = str(ipv6.get_ipv6_addr_by_EUI64(
dhcp_lla_ip = str(netutils.get_ipv6_addr_by_EUI64(
const.IPv6_LLA_PREFIX,
dhcp_mac))
@ -702,7 +702,7 @@ class SGServerRpcCallBackTestCase(test_sg.SecurityGroupDBTestCase):
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(
gateway_lla_ip = str(netutils.get_ipv6_addr_by_EUI64(
const.IPv6_LLA_PREFIX,
gateway_mac))
@ -770,7 +770,7 @@ class SGServerRpcCallBackTestCase(test_sg.SecurityGroupDBTestCase):
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(
gateway_lla_ip = str(netutils.get_ipv6_addr_by_EUI64(
const.IPv6_LLA_PREFIX,
gateway_mac))
# Create another router interface port
@ -845,7 +845,7 @@ class SGServerRpcCallBackTestCase(test_sg.SecurityGroupDBTestCase):
device_owner=const.DEVICE_OWNER_DVR_INTERFACE)
gateway_mac = gateway_res['port']['mac_address']
gateway_port_id = gateway_res['port']['id']
gateway_lla_ip = str(ipv6.get_ipv6_addr_by_EUI64(
gateway_lla_ip = str(netutils.get_ipv6_addr_by_EUI64(
const.IPv6_LLA_PREFIX,
gateway_mac))

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

@ -26,6 +26,7 @@ from neutron_lib import exceptions as lib_exc
from oslo_concurrency import lockutils
from oslo_config import cfg
from oslo_utils import importutils
from oslo_utils import netutils
from oslo_utils import uuidutils
import six
from sqlalchemy import event
@ -1356,8 +1357,8 @@ fixed_ips=ip_address%%3D%s&fixed_ips=ip_address%%3D%s&fixed_ips=subnet_id%%3D%s
if ipv6_utils.is_auto_address_subnet(subnet['subnet']):
port_mac = port['port']['mac_address']
subnet_cidr = subnet['subnet']['cidr']
eui_addr = str(ipv6_utils.get_ipv6_addr_by_EUI64(subnet_cidr,
port_mac))
eui_addr = str(netutils.get_ipv6_addr_by_EUI64(subnet_cidr,
port_mac))
self.assertEqual(port['port']['fixed_ips'][0]['ip_address'],
eui_addr)
@ -1664,8 +1665,8 @@ fixed_ips=ip_address%%3D%s&fixed_ips=ip_address%%3D%s&fixed_ips=subnet_id%%3D%s
self.assertEqual(1, len(ips))
port_mac = port['port']['mac_address']
subnet_cidr = subnet['subnet']['cidr']
eui_addr = str(ipv6_utils.get_ipv6_addr_by_EUI64(subnet_cidr,
port_mac))
eui_addr = str(netutils.get_ipv6_addr_by_EUI64(subnet_cidr,
port_mac))
self.assertEqual(ips[0]['ip_address'], eui_addr)
self.assertEqual(ips[0]['subnet_id'], subnet['subnet']['id'])
@ -1891,7 +1892,7 @@ fixed_ips=ip_address%%3D%s&fixed_ips=ip_address%%3D%s&fixed_ips=subnet_id%%3D%s
res = self._create_port(self.fmt, net_id=net_id)
port = self.deserialize(self.fmt, res)
port_mac = port['port']['mac_address']
eui_addr = str(ipv6_utils.get_ipv6_addr_by_EUI64(
eui_addr = str(netutils.get_ipv6_addr_by_EUI64(
prefix, port_mac))
fixedips = [{'subnet_id': subnet_id, 'ip_address': eui_addr}]
self.assertEqual(fixedips, port['port']['fixed_ips'])
@ -1969,8 +1970,8 @@ fixed_ips=ip_address%%3D%s&fixed_ips=ip_address%%3D%s&fixed_ips=subnet_id%%3D%s
subnet['subnet']['id']}]) as port:
port_mac = port['port']['mac_address']
subnet_cidr = subnet['subnet']['cidr']
eui_addr = str(ipv6_utils.get_ipv6_addr_by_EUI64(subnet_cidr,
port_mac))
eui_addr = str(netutils.get_ipv6_addr_by_EUI64(subnet_cidr,
port_mac))
self.assertEqual(port['port']['fixed_ips'][0]['ip_address'],
eui_addr)
@ -1996,7 +1997,7 @@ fixed_ips=ip_address%%3D%s&fixed_ips=ip_address%%3D%s&fixed_ips=subnet_id%%3D%s
self.assertEqual(2, len(ips))
port_mac = port['port']['mac_address']
subnet_cidr = subnet2['subnet']['cidr']
eui_addr = str(ipv6_utils.get_ipv6_addr_by_EUI64(
eui_addr = str(netutils.get_ipv6_addr_by_EUI64(
subnet_cidr, port_mac))
self.assertIn(ips[0]['ip_address'], network_ip_set)
self.assertIn(ips[1]['ip_address'], network_ip_set)
@ -2026,7 +2027,7 @@ fixed_ips=ip_address%%3D%s&fixed_ips=ip_address%%3D%s&fixed_ips=subnet_id%%3D%s
def _calc_ipv6_addr_by_EUI64(port, subnet):
port_mac = port['port']['mac_address']
subnet_cidr = subnet['subnet']['cidr']
return str(ipv6_utils.get_ipv6_addr_by_EUI64(subnet_cidr, port_mac))
return str(netutils.get_ipv6_addr_by_EUI64(subnet_cidr, port_mac))
def test_ip_allocation_for_ipv6_subnet_slaac_address_mode(self):
res = self._create_network(fmt=self.fmt, name='net',
@ -2197,10 +2198,10 @@ fixed_ips=ip_address%%3D%s&fixed_ips=ip_address%%3D%s&fixed_ips=subnet_id%%3D%s
port_mac = port['port']['mac_address']
cidr_1 = v6_subnet_1['subnet']['cidr']
cidr_2 = v6_subnet_2['subnet']['cidr']
eui_addr_1 = str(ipv6_utils.get_ipv6_addr_by_EUI64(cidr_1,
port_mac))
eui_addr_2 = str(ipv6_utils.get_ipv6_addr_by_EUI64(cidr_2,
port_mac))
eui_addr_1 = str(netutils.get_ipv6_addr_by_EUI64(cidr_1,
port_mac))
eui_addr_2 = str(netutils.get_ipv6_addr_by_EUI64(cidr_2,
port_mac))
self.assertEqual({eui_addr_1, eui_addr_2},
{fixed_ip['ip_address'] for fixed_ip in
port['port']['fixed_ips']})
@ -4392,7 +4393,7 @@ class TestSubnetsV2(NeutronDbPluginV2TestCase):
# create expected fixed_ips
port_mac = port_dict['mac_address']
eui_addr = str(ipv6_utils.get_ipv6_addr_by_EUI64(prefix,
eui_addr = str(netutils.get_ipv6_addr_by_EUI64(prefix,
port_mac))
ips = port_dict['fixed_ips']
for fip in ips:

6
neutron/tests/unit/db/test_ipam_non_pluggable_backend.py
View File

@ -16,9 +16,9 @@
import mock
from neutron_lib import constants as n_const
from oslo_config import cfg
from oslo_utils import netutils
from neutron.common import constants
from neutron.common import ipv6_utils
from neutron.db import db_base_plugin_v2
from neutron.db import ipam_backend_mixin
from neutron.db import ipam_non_pluggable_backend as non_ipam
@ -70,7 +70,7 @@ class TestIpamNonPluggableBackend(base.BaseTestCase):
'device_owner': n_const.DEVICE_OWNER_COMPUTE_PREFIX}}
expected = []
for subnet in subnets:
addr = str(ipv6_utils.get_ipv6_addr_by_EUI64(
addr = str(netutils.get_ipv6_addr_by_EUI64(
subnet['cidr'], port['port']['mac_address']))
expected.append({'ip_address': addr, 'subnet_id': subnet['id']})
@ -103,7 +103,7 @@ class TestIpamNonPluggableBackend(base.BaseTestCase):
'device_owner': n_const.DEVICE_OWNER_COMPUTE_PREFIX}}
expected = []
for subnet in subnets:
addr = str(ipv6_utils.get_ipv6_addr_by_EUI64(
addr = str(netutils.get_ipv6_addr_by_EUI64(
subnet['cidr'], port['port']['mac_address']))
expected.append({'ip_address': addr, 'subnet_id': subnet['id']})

6
neutron/tests/unit/db/test_ipam_pluggable_backend.py
View File

@ -20,11 +20,11 @@ from neutron_lib import constants
from neutron_lib import exceptions as n_exc
from oslo_config import cfg
from oslo_db import exception as db_exc
from oslo_utils import netutils
from oslo_utils import uuidutils
import webob.exc
from neutron.common import constants as n_const
from neutron.common import ipv6_utils
from neutron.db import ipam_backend_mixin
from neutron.db import ipam_pluggable_backend
from neutron.db import models_v2
@ -246,8 +246,8 @@ class TestDbBasePluginIpam(test_db_base.NeutronDbPluginV2TestCase):
'subnet_cidr': '2001:470:abcd::/64',
'mac': '6c:62:6d:de:cf:49',
'eui64_address': True}
eui64_ip = ipv6_utils.get_ipv6_addr_by_EUI64(ip['subnet_cidr'],
ip['mac'])
eui64_ip = netutils.get_ipv6_addr_by_EUI64(ip['subnet_cidr'],
ip['mac'])
mocks['ipam']._ipam_allocate_ips(mock.ANY, mocks['driver'],
mock.ANY, [ip])

6
neutron/tests/unit/ipam/test_requests.py
View File

@ -14,9 +14,9 @@ import mock
import netaddr
from neutron_lib import constants
from oslo_config import cfg
from oslo_utils import netutils
from oslo_utils import uuidutils
from neutron.common import ipv6_utils
from neutron import context
from neutron.ipam import driver
from neutron.ipam import exceptions as ipam_exc
@ -204,8 +204,8 @@ class TestAddressRequest(base.BaseTestCase):
def test_automatic_address_request_eui64(self):
subnet_cidr = '2607:f0d0:1002:51::/64'
port_mac = 'aa:bb:cc:dd:ee:ff'
eui_addr = str(ipv6_utils.get_ipv6_addr_by_EUI64(subnet_cidr,
port_mac))
eui_addr = str(netutils.get_ipv6_addr_by_EUI64(subnet_cidr,
port_mac))
request = ipam_req.AutomaticAddressRequest(
address_type=self.EUI64,
prefix=subnet_cidr,