Merge "Move ip_netmask_to_cidr to network utils"

cloudbaseinit/metadata/services/baseopenstackservice.py

@@ -16,7 +16,6 @@
 import json
 import posixpath
 
-import netaddr
 from oslo_log import log as oslo_logging
 
 from cloudbaseinit import conf as cloudbaseinit_conf
@@ -25,6 +24,7 @@ from cloudbaseinit.metadata.services import base
 from cloudbaseinit.models import network as network_model
 from cloudbaseinit.utils import debiface
 from cloudbaseinit.utils import encoding
+from cloudbaseinit.utils import network as network_utils
 from cloudbaseinit.utils import x509constants
 
 NETWORK_LINK_TYPE_PHYSICAL = "phy"
@@ -101,14 +101,6 @@ class BaseOpenStackService(base.BaseMetadataService):
 
         return debiface.parse(content)
 
-    @staticmethod
-    def _ip_netmask_to_cidr(ip_address, netmask):
-        if netmask is None:
-            return ip_address
-        prefix_len = netaddr.IPNetwork(
-            u"%s/%s" % (ip_address, netmask)).prefixlen
-        return u"%s/%s" % (ip_address, prefix_len)
-
     @staticmethod
     def _parse_network_data_links(links_data):
         links = []
@@ -195,7 +187,7 @@ class BaseOpenStackService(base.BaseMetadataService):
             link_id = network_data.get("link")
             ip_address = network_data.get("ip_address")
             netmask = network_data.get("netmask")
-            address_cidr = BaseOpenStackService._ip_netmask_to_cidr(
+            address_cidr = network_utils.ip_netmask_to_cidr(
                 ip_address, netmask)
 
             routes = []
@@ -203,7 +195,7 @@ class BaseOpenStackService(base.BaseMetadataService):
                 gateway = route_data.get("gateway")
                 network = route_data.get("network")
                 netmask = route_data.get("netmask")
-                network_cidr = BaseOpenStackService._ip_netmask_to_cidr(
+                network_cidr = network_utils.ip_netmask_to_cidr(
                     network, netmask)
 
                 route = network_model.Route(

cloudbaseinit/tests/utils/test_network.py

@@ -96,3 +96,21 @@ class NetworkUtilsTest(unittest.TestCase):
         res = network.get_local_ip("fake address")
         self.assertEqual(res, "fake name")
         mock_socket().connect.assert_called_with(("fake address", 8000))
+
+    def _test_ip_netmask_to_cidr(self, expected_result, fake_ip_address,
+                                 fake_netmask):
+        result = network.ip_netmask_to_cidr(fake_ip_address, fake_netmask)
+        self.assertEqual(expected_result, result)
+
+    def test_ip_netmask_to_cidr(self):
+        fake_ip_address = '10.1.1.1'
+        expected_result = '10.1.1.1/24'
+        fake_netmask = '255.255.255.0'
+        self._test_ip_netmask_to_cidr(expected_result, fake_ip_address,
+                                      fake_netmask)
+
+    def test_ip_netmask_to_cidr_empty_netmask(self):
+        fake_ip_address = '10.1.1.1'
+        fake_netmask = None
+        self._test_ip_netmask_to_cidr(fake_ip_address, fake_ip_address,
+                                      fake_netmask)

cloudbaseinit/utils/network.py

@@ -14,6 +14,7 @@
 
 
 import binascii
+import netaddr
 import socket
 import struct
 import sys
@@ -89,3 +90,11 @@ def netmask6_to_4_truncate(netmask6):
     mask = "1" * length + "0" * (32 - length)
     network_address = struct.pack("!L", int(mask, 2))
     return socket.inet_ntoa(network_address)
+
+
+def ip_netmask_to_cidr(ip_address, netmask):
+    if not netmask:
+        return ip_address
+    prefix_len = netaddr.IPNetwork(
+        u"%s/%s" % (ip_address, netmask)).prefixlen
+    return u"%s/%s" % (ip_address, prefix_len)