Introduce Opts for IP addresses
In order to validate correct input values for IP addresses new Opt was introduced. Validation is done on parsing level so there is no need to explicitly check for valid ip address in the code. Requirement for netaddr package was added. DocImpact Change-Id: I9adc30d9b989e8c636fefd435885c4c363ca540c Partial-Bug: #1284684
This commit is contained in:
parent
ef6f9bc46a
commit
afab1f5266
@ -950,6 +950,15 @@ class DictOpt(Opt):
|
||||
super(DictOpt, self).__init__(name, type=types.Dict(), **kwargs)
|
||||
|
||||
|
||||
class IPOpt(Opt):
|
||||
|
||||
"""Opt with IPAddress type (either IPv4, IPv6 or both)."""
|
||||
|
||||
def __init__(self, name, version=None, **kwargs):
|
||||
super(IPOpt, self).__init__(name, type=types.IPAddress(version),
|
||||
**kwargs)
|
||||
|
||||
|
||||
class MultiOpt(Opt):
|
||||
|
||||
"""Multi-value option.
|
||||
|
@ -18,6 +18,7 @@ Use these classes as values for the `type` argument to
|
||||
:class:`oslo.config.cfg.Opt` and its subclasses.
|
||||
|
||||
"""
|
||||
import netaddr
|
||||
|
||||
|
||||
class String(object):
|
||||
@ -332,3 +333,52 @@ class Dict(object):
|
||||
(self.__class__ == other.__class__) and
|
||||
(self.value_type == other.value_type)
|
||||
)
|
||||
|
||||
|
||||
class IPAddress(object):
|
||||
|
||||
"""IP address type
|
||||
|
||||
Represents either ipv4 or ipv6. Without specifying version parameter both
|
||||
versions are checked
|
||||
|
||||
:param version: defines which version should be explicitly checked (4 or 6)
|
||||
|
||||
"""
|
||||
|
||||
def __init__(self, version=None):
|
||||
version_checkers = {
|
||||
None: self._check_both_versions,
|
||||
4: self._check_ipv4,
|
||||
6: self._check_ipv6
|
||||
}
|
||||
|
||||
self.version_checker = version_checkers.get(version)
|
||||
if self.version_checker is None:
|
||||
raise TypeError("%s is not a valid IP version." % version)
|
||||
|
||||
def __call__(self, value):
|
||||
value = str(value)
|
||||
if not value:
|
||||
raise ValueError("IP address cannot be an empty string")
|
||||
self.version_checker(value)
|
||||
return value
|
||||
|
||||
def __repr__(self):
|
||||
return "IPAddress"
|
||||
|
||||
def __eq__(self, other):
|
||||
return self.__class__ == other.__class__
|
||||
|
||||
def _check_ipv4(self, address):
|
||||
if not netaddr.valid_ipv4(address, netaddr.core.INET_PTON):
|
||||
raise ValueError("%s is not an IPv4 address" % address)
|
||||
|
||||
def _check_ipv6(self, address):
|
||||
if not netaddr.valid_ipv6(address, netaddr.core.INET_PTON):
|
||||
raise ValueError("%s is not an IPv6 address" % address)
|
||||
|
||||
def _check_both_versions(self, address):
|
||||
if not (netaddr.valid_ipv4(address, netaddr.core.INET_PTON) or
|
||||
netaddr.valid_ipv6(address, netaddr.core.INET_PTON)):
|
||||
raise ValueError("%s is not IPv4 or IPv6 address" % address)
|
||||
|
@ -1,2 +1,3 @@
|
||||
argparse
|
||||
netaddr>=0.7.6
|
||||
six>=1.7.0
|
||||
|
@ -14,6 +14,7 @@
|
||||
|
||||
import argparse
|
||||
import errno
|
||||
import functools
|
||||
import os
|
||||
import shutil
|
||||
import sys
|
||||
@ -265,6 +266,9 @@ class CliOptsTestCase(BaseTestCase):
|
||||
deps - a tuple of deprecated name/group
|
||||
"""
|
||||
|
||||
IPv4Opt = functools.partial(cfg.IPOpt, version=4)
|
||||
IPv6Opt = functools.partial(cfg.IPOpt, version=6)
|
||||
|
||||
scenarios = [
|
||||
('str_default',
|
||||
dict(opt_class=cfg.StrOpt, default=None, cli_args=[], value=None,
|
||||
@ -364,6 +368,22 @@ class CliOptsTestCase(BaseTestCase):
|
||||
('float_arg_deprecated_group_and_name',
|
||||
dict(opt_class=cfg.FloatOpt, default=None,
|
||||
cli_args=['--old-oof', '2.0'], value=2.0, deps=('oof', 'old'))),
|
||||
('ipv4addr_arg',
|
||||
dict(opt_class=IPv4Opt, default=None,
|
||||
cli_args=['--foo', '192.168.0.1'], value='192.168.0.1',
|
||||
deps=(None, None))),
|
||||
('ipaddr_arg_implicitv4',
|
||||
dict(opt_class=cfg.IPOpt, default=None,
|
||||
cli_args=['--foo', '192.168.0.1'], value='192.168.0.1',
|
||||
deps=(None, None))),
|
||||
('ipaddr_arg_implicitv6',
|
||||
dict(opt_class=cfg.IPOpt, default=None,
|
||||
cli_args=['--foo', 'abcd:ef::1'], value='abcd:ef::1',
|
||||
deps=(None, None))),
|
||||
('ipv6addr_arg',
|
||||
dict(opt_class=IPv6Opt, default=None,
|
||||
cli_args=['--foo', 'abcd:ef::1'], value='abcd:ef::1',
|
||||
deps=(None, None))),
|
||||
('list_default',
|
||||
dict(opt_class=cfg.ListOpt, default=['bar'],
|
||||
cli_args=[], value=['bar'], deps=(None, None))),
|
||||
|
@ -376,3 +376,36 @@ class DictTypeTests(TypeTestHelper, unittest.TestCase):
|
||||
|
||||
def test_not_equal_to_other_class(self):
|
||||
self.assertFalse(types.Dict() == types.Integer())
|
||||
|
||||
|
||||
class IPAddressTypeTests(TypeTestHelper, unittest.TestCase):
|
||||
type = types.IPAddress()
|
||||
|
||||
def test_ipv4_address(self):
|
||||
self.assertConvertedValue('192.168.0.1', '192.168.0.1')
|
||||
|
||||
def test_ipv6_address(self):
|
||||
self.assertConvertedValue('abcd:ef::1', 'abcd:ef::1')
|
||||
|
||||
def test_strings(self):
|
||||
self.assertInvalid('')
|
||||
self.assertInvalid('foo')
|
||||
|
||||
def test_numbers(self):
|
||||
self.assertInvalid(1)
|
||||
self.assertInvalid(-1)
|
||||
self.assertInvalid(3.14)
|
||||
|
||||
|
||||
class IPv4AddressTypeTests(IPAddressTypeTests):
|
||||
type = types.IPAddress(4)
|
||||
|
||||
def test_ipv6_address(self):
|
||||
self.assertInvalid('abcd:ef::1')
|
||||
|
||||
|
||||
class IPv6AddressTypeTests(IPAddressTypeTests):
|
||||
type = types.IPAddress(6)
|
||||
|
||||
def test_ipv4_address(self):
|
||||
self.assertInvalid('192.168.0.1')
|
||||
|
Loading…
Reference in New Issue
Block a user