Merge "Canonizes IPv6 before insert it into the db"

nova/db/sqlalchemy/types.py

@@ -18,9 +18,21 @@
 """Custom SQLAlchemy types."""
 
 from sqlalchemy.dialects import postgresql
-from sqlalchemy import String
+from sqlalchemy import types
+
+from nova import utils
 
 
-def IPAddress():
+class IPAddress(types.TypeDecorator):
     """An SQLAlchemy type representing an IP-address."""
-    return String(39).with_variant(postgresql.INET(), 'postgresql')
+    impl = types.String(39).with_variant(postgresql.INET(), 'postgresql')
+
+    def process_bind_param(self, value, dialect):
+        """Process/Formats the value before insert it into the db."""
+        if dialect.name == 'postgresql':
+            return value
+        # NOTE(maurosr): The purpose here is to convert ipv6 to the shortened
+        # form, not validate it.
+        elif utils.is_valid_ipv6(value):
+            return utils.get_shortened_ipv6(value)
+        return value

nova/tests/test_utils.py

@@ -25,6 +25,7 @@ import StringIO
 import tempfile
 
 import mox
+import netaddr
 
 import nova
 from nova import exception
@@ -481,6 +482,29 @@ class GenericUtilsTestCase(test.TestCase):
         self.assertFalse(utils.is_valid_ipv4('::1'))
         self.assertFalse(utils.is_valid_ipv4('bacon'))
 
+    def test_is_valid_ipv6(self):
+        self.assertTrue(utils.is_valid_ipv6("::1"))
+        self.assertTrue(utils.is_valid_ipv6(
+                            "abcd:ef01:2345:6789:abcd:ef01:192.168.254.254"))
+        self.assertTrue(utils.is_valid_ipv6(
+                                    "0000:0000:0000:0000:0000:0000:0000:0001"))
+        self.assertFalse(utils.is_valid_ipv6("foo"))
+        self.assertFalse(utils.is_valid_ipv6("127.0.0.1"))
+
+    def test_get_shortened_ipv6(self):
+        self.assertEquals("abcd:ef01:2345:6789:abcd:ef01:c0a8:fefe",
+                          utils.get_shortened_ipv6(
+                            "abcd:ef01:2345:6789:abcd:ef01:192.168.254.254"))
+        self.assertEquals("::1", utils.get_shortened_ipv6(
+                                    "0000:0000:0000:0000:0000:0000:0000:0001"))
+        self.assertEquals("caca::caca:0:babe:201:102",
+                          utils.get_shortened_ipv6(
+                                    "caca:0000:0000:caca:0000:babe:0201:0102"))
+        self.assertRaises(netaddr.AddrFormatError, utils.get_shortened_ipv6,
+                          "127.0.0.1")
+        self.assertRaises(netaddr.AddrFormatError, utils.get_shortened_ipv6,
+                          "failure")
+
 
 class MonkeyPatchTestCase(test.TestCase):
     """Unit test for utils.monkey_patch()."""

nova/utils.py

@@ -892,6 +892,15 @@ def is_valid_ipv4(address):
         return False
 
 
+def is_valid_ipv6(address):
+    return netaddr.valid_ipv6(address)
+
+
+def get_shortened_ipv6(address):
+    addr = netaddr.IPAddress(address, version=6)
+    return str(addr.ipv6())
+
+
 def is_valid_cidr(address):
     """Check if the provided ipv4 or ipv6 address is a valid
     CIDR address or not"""