Remove netaddr module requirement
This patch removes the 'netaddr' module from the Octavia requirements list. It replaces the use of netaddr in Octavia with the python standard library 'ipaddress' module. It also increase unit test coverage for the impacted functions. Change-Id: I9ec023116417e5c8850395d4c06db63e5be1c022
This commit is contained in:
parent
3f4a44db94
commit
085cf1c2e3
@ -13,9 +13,9 @@
|
||||
# under the License.
|
||||
|
||||
import copy
|
||||
import ipaddress
|
||||
|
||||
from dateutil import parser
|
||||
import netaddr
|
||||
from wsme import types as wtypes
|
||||
|
||||
from octavia.common import constants
|
||||
@ -50,8 +50,8 @@ class CidrType(wtypes.UserType):
|
||||
def validate(value):
|
||||
"""Validates whether value is an IPv4 or IPv6 CIDR."""
|
||||
try:
|
||||
return str(netaddr.IPNetwork(value).cidr)
|
||||
except (ValueError, netaddr.core.AddrFormatError) as e:
|
||||
return ipaddress.ip_network(value, strict=False).with_prefixlen
|
||||
except Exception as e:
|
||||
error = 'Value should be IPv4 or IPv6 CIDR format'
|
||||
raise ValueError(error) from e
|
||||
|
||||
|
@ -20,10 +20,10 @@
|
||||
|
||||
import base64
|
||||
import hashlib
|
||||
import ipaddress
|
||||
import re
|
||||
import socket
|
||||
|
||||
import netaddr
|
||||
from oslo_config import cfg
|
||||
from oslo_log import log as logging
|
||||
from oslo_utils import excutils
|
||||
@ -73,47 +73,42 @@ def get_network_driver():
|
||||
|
||||
def is_ipv4(ip_address):
|
||||
"""Check if ip address is IPv4 address."""
|
||||
ip = netaddr.IPAddress(ip_address)
|
||||
ip = ipaddress.ip_address(ip_address)
|
||||
return ip.version == 4
|
||||
|
||||
|
||||
def is_ipv6(ip_address):
|
||||
"""Check if ip address is IPv6 address."""
|
||||
ip = netaddr.IPAddress(ip_address)
|
||||
ip = ipaddress.ip_address(ip_address)
|
||||
return ip.version == 6
|
||||
|
||||
|
||||
def is_cidr_ipv6(cidr):
|
||||
"""Check if CIDR is IPv6 address with subnet prefix."""
|
||||
ip = netaddr.IPNetwork(cidr)
|
||||
ip = ipaddress.ip_network(cidr, strict=False)
|
||||
return ip.version == 6
|
||||
|
||||
|
||||
def is_ipv6_lla(ip_address):
|
||||
"""Check if ip address is IPv6 link local address."""
|
||||
ip = netaddr.IPAddress(ip_address)
|
||||
return ip.version == 6 and ip.is_link_local()
|
||||
ip = ipaddress.ip_address(ip_address)
|
||||
return ip.version == 6 and ip.is_link_local
|
||||
|
||||
|
||||
def ip_port_str(ip_address, port):
|
||||
"""Return IP port as string representation depending on address family."""
|
||||
ip = netaddr.IPAddress(ip_address)
|
||||
ip = ipaddress.ip_address(ip_address)
|
||||
if ip.version == 4:
|
||||
return "{ip}:{port}".format(ip=ip, port=port)
|
||||
return "[{ip}]:{port}".format(ip=ip, port=port)
|
||||
|
||||
|
||||
def netmask_to_prefix(netmask):
|
||||
return netaddr.IPAddress(netmask).netmask_bits()
|
||||
return ipaddress.ip_network(f'0.0.0.0/{netmask}', strict=False).prefixlen
|
||||
|
||||
|
||||
def ip_netmask_to_cidr(ip, netmask):
|
||||
net = netaddr.IPNetwork("0.0.0.0/0")
|
||||
if ip and netmask:
|
||||
net = netaddr.IPNetwork(
|
||||
"{ip}/{netmask}".format(ip=ip, netmask=netmask)
|
||||
)
|
||||
return "{ip}/{netmask}".format(ip=net.network, netmask=net.prefixlen)
|
||||
return ipaddress.ip_network(f'{ip}/{netmask}', strict=False).with_prefixlen
|
||||
|
||||
|
||||
def get_vip_security_group_name(loadbalancer_id):
|
||||
|
@ -22,7 +22,6 @@ Defined here so these can also be used at deeper levels than the API.
|
||||
import ipaddress
|
||||
import re
|
||||
|
||||
import netaddr
|
||||
from oslo_config import cfg
|
||||
import rfc3986
|
||||
from wsme import types as wtypes
|
||||
@ -391,7 +390,7 @@ def network_allowed_by_config(network_id, valid_networks=None):
|
||||
|
||||
|
||||
def is_ip_member_of_cidr(address, cidr):
|
||||
if netaddr.IPAddress(address) in netaddr.IPNetwork(cidr):
|
||||
if ipaddress.ip_address(address) in ipaddress.ip_network(cidr):
|
||||
return True
|
||||
return False
|
||||
|
||||
|
@ -41,6 +41,10 @@ class TestConfig(base.TestCase):
|
||||
self.assertTrue(utils.is_ipv6('2001:db8::1'))
|
||||
self.assertTrue(utils.is_ipv6('fe80::225:90ff:fefb:53ad'))
|
||||
|
||||
def test_is_cidr_ipv6(self):
|
||||
self.assertTrue(utils.is_cidr_ipv6('2001:db8::/32'))
|
||||
self.assertFalse(utils.is_cidr_ipv6('192.0.2.0/32'))
|
||||
|
||||
def test_is_ipv6_lla(self):
|
||||
self.assertFalse(utils.is_ipv6_lla('192.0.2.10'))
|
||||
self.assertFalse(utils.is_ipv6_lla('169.254.0.10'))
|
||||
|
@ -526,3 +526,13 @@ class TestValidations(base.TestCase):
|
||||
exceptions.ValidationException,
|
||||
validate.check_alpn_protocols,
|
||||
[])
|
||||
|
||||
def test_is_ip_member_of_cidr(self):
|
||||
self.assertTrue(validate.is_ip_member_of_cidr('192.0.0.5',
|
||||
'192.0.0.0/24'))
|
||||
self.assertFalse(validate.is_ip_member_of_cidr('198.51.100.5',
|
||||
'192.0.0.0/24'))
|
||||
self.assertTrue(validate.is_ip_member_of_cidr('2001:db8::5',
|
||||
'2001:db8::/32'))
|
||||
self.assertFalse(validate.is_ip_member_of_cidr('::ffff:0:203.0.113.5',
|
||||
'2001:db8::/32'))
|
||||
|
@ -0,0 +1,5 @@
|
||||
---
|
||||
other:
|
||||
- |
|
||||
The netaddr python module has been removed as an Octavia requirement. It
|
||||
has been replaced with the python standard library 'ipaddress' module.
|
@ -47,7 +47,6 @@ tenacity>=5.0.4 # Apache-2.0
|
||||
distro>=1.2.0 # Apache-2.0
|
||||
jsonschema>=3.2.0 # MIT
|
||||
octavia-lib>=2.5.0 # Apache-2.0
|
||||
netaddr>=0.7.19 # BSD
|
||||
simplejson>=3.13.2 # MIT
|
||||
setproctitle>=1.1.10 # BSD
|
||||
python-dateutil>=2.7.0 # BSD
|
||||
|
Loading…
Reference in New Issue
Block a user