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

5 files changed, 113 insertions, 0 deletions

diff --git a/oslo/config/cfg.py b/oslo/config/cfg.py
index 21ec818..3d58834 100644
--- a/oslo/config/cfg.py
+++ b/oslo/config/cfg.py
@@ -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.
diff --git a/oslo/config/types.py b/oslo/config/types.py
index a411942..541095b 100644
--- a/oslo/config/types.py
+++ b/oslo/config/types.py
@@ -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)
diff --git a/requirements.txt b/requirements.txt
index a7bc656..7b8af46 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -1,2 +1,3 @@
 argparse
+netaddr>=0.7.6
 six>=1.7.0
diff --git a/tests/test_cfg.py b/tests/test_cfg.py
index 165ae31..268565f 100644
--- a/tests/test_cfg.py
+++ b/tests/test_cfg.py
@@ -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))),
diff --git a/tests/test_types.py b/tests/test_types.py
index 88c8aea..a9f32a7 100644
--- a/tests/test_types.py
+++ b/tests/test_types.py
@@ -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')