Merge "Use is_valid_cidr and is_valid_ipv6_cidr from oslo_utils"
This commit is contained in:
commit
7f050fef50
|
@ -59,7 +59,7 @@ class CIDR(types.TypeDecorator):
|
|||
def process_bind_param(self, value, dialect):
|
||||
"""Process/Formats the value before insert it into the db."""
|
||||
# NOTE(sdague): normalize all the inserts
|
||||
if utils.is_valid_ipv6_cidr(value):
|
||||
if netutils.is_valid_ipv6_cidr(value):
|
||||
return utils.get_shortened_ipv6_cidr(value)
|
||||
return value
|
||||
|
||||
|
|
|
@ -18,12 +18,12 @@
|
|||
# under the License.
|
||||
|
||||
from oslo_utils import encodeutils
|
||||
from oslo_utils import netutils as utils
|
||||
from six.moves import urllib
|
||||
|
||||
from nova import exception
|
||||
from nova.i18n import _
|
||||
from nova.objects import security_group as security_group_obj
|
||||
from nova import utils
|
||||
|
||||
|
||||
class SecurityGroupBase(object):
|
||||
|
|
|
@ -178,17 +178,6 @@ class GenericUtilsTestCase(test.NoDBTestCase):
|
|||
self.assertEqual('<', utils.xhtml_escape('<'))
|
||||
self.assertEqual('<foo>', utils.xhtml_escape('<foo>'))
|
||||
|
||||
def test_is_valid_ipv6_cidr(self):
|
||||
self.assertTrue(utils.is_valid_ipv6_cidr("2600::/64"))
|
||||
self.assertTrue(utils.is_valid_ipv6_cidr(
|
||||
"abcd:ef01:2345:6789:abcd:ef01:192.168.254.254/48"))
|
||||
self.assertTrue(utils.is_valid_ipv6_cidr(
|
||||
"0000:0000:0000:0000:0000:0000:0000:0001/32"))
|
||||
self.assertTrue(utils.is_valid_ipv6_cidr(
|
||||
"0000:0000:0000:0000:0000:0000:0000:0001"))
|
||||
self.assertFalse(utils.is_valid_ipv6_cidr("foo"))
|
||||
self.assertFalse(utils.is_valid_ipv6_cidr("127.0.0.1"))
|
||||
|
||||
def test_get_shortened_ipv6(self):
|
||||
self.assertEqual("abcd:ef01:2345:6789:abcd:ef01:c0a8:fefe",
|
||||
utils.get_shortened_ipv6(
|
||||
|
|
|
@ -539,14 +539,6 @@ def parse_server_string(server_str):
|
|||
return ('', '')
|
||||
|
||||
|
||||
def is_valid_ipv6_cidr(address):
|
||||
try:
|
||||
netaddr.IPNetwork(address, version=6).cidr
|
||||
return True
|
||||
except (TypeError, netaddr.AddrFormatError):
|
||||
return False
|
||||
|
||||
|
||||
def get_shortened_ipv6(address):
|
||||
addr = netaddr.IPAddress(address, version=6)
|
||||
return str(addr.ipv6())
|
||||
|
@ -557,29 +549,6 @@ def get_shortened_ipv6_cidr(address):
|
|||
return str(net.cidr)
|
||||
|
||||
|
||||
def is_valid_cidr(address):
|
||||
"""Check if address is valid
|
||||
|
||||
The provided address can be a IPv6 or a IPv4
|
||||
CIDR address.
|
||||
"""
|
||||
try:
|
||||
# Validate the correct CIDR Address
|
||||
netaddr.IPNetwork(address)
|
||||
except netaddr.AddrFormatError:
|
||||
return False
|
||||
|
||||
# Prior validation partially verify /xx part
|
||||
# Verify it here
|
||||
ip_segment = address.split('/')
|
||||
|
||||
if (len(ip_segment) <= 1 or
|
||||
ip_segment[1] == ''):
|
||||
return False
|
||||
|
||||
return True
|
||||
|
||||
|
||||
def get_ip_version(network):
|
||||
"""Returns the IP version of a network (IPv4 or IPv6).
|
||||
|
||||
|
|
Loading…
Reference in New Issue