Merge "Add get_my_ip()"

oslo/utils/netutils.py

@@ -21,8 +21,10 @@ import logging
 import socket
 
 import netaddr
+import netifaces
 from six.moves.urllib import parse
 
+from oslo.utils._i18n import _LI
 from oslo.utils._i18n import _LW
 
 LOG = logging.getLogger(__name__)
@@ -112,6 +114,37 @@ def is_valid_ip(address):
     return is_valid_ipv4(address) or is_valid_ipv6(address)
 
 
+def get_my_ipv4():
+    """Returns the actual ipv4 of the local machine.
+
+    This code figures out what source address would be used if some traffic
+    were to be sent out to some well known address on the Internet. In this
+    case, IP from RFC5737 is used, but the specific address does not
+    matter much. No traffic is actually sent.
+    """
+    try:
+        csock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
+        csock.connect(('192.0.2.0', 80))
+        (addr, port) = csock.getsockname()
+        csock.close()
+        return addr
+    except socket.error:
+        return _get_my_ipv4_address()
+
+
+def _get_my_ipv4_address():
+    """Figure out the best ipv4
+    """
+    LOCALHOST = '127.0.0.1'
+    gtw = netifaces.gateways()
+    try:
+        interface = gtw['default'][netifaces.AF_INET][1]
+        return netifaces.ifaddresses(interface)[netifaces.AF_INET][0]['addr']
+    except Exception:
+        LOG.info(_LI("Couldn't get IPv4"))
+    return LOCALHOST
+
+
 class _ModifiedSplitResult(parse.SplitResult):
     """Split results class for urlsplit."""
 

requirements.txt

@@ -7,3 +7,4 @@ six>=1.7.0
 iso8601>=0.1.9
 oslo.i18n>=1.0.0  # Apache-2.0
 netaddr>=0.7.12
+netifaces>=0.10.4

tests/test_netutils.py

@@ -16,6 +16,8 @@
 import socket
 
 import mock
+from mock import patch
+import netifaces
 from oslotest import base as test_base
 
 from oslo.utils import netutils
@@ -174,4 +176,37 @@ class NetworkUtilsTest(test_base.BaseTestCase):
 
         self.assertFalse(netutils.is_valid_ip('::1.2.3.'))
 
-        self.assertFalse(netutils.is_valid_ip(''))
+        self.assertFalse(netutils.is_valid_ip(''))
+
+    def test_get_my_ip(self):
+        sock_attrs = {
+            'return_value.getsockname.return_value': ['1.2.3.4', '']}
+        with mock.patch('socket.socket', **sock_attrs):
+            addr = netutils.get_my_ipv4()
+        self.assertEqual(addr, '1.2.3.4')
+
+    @mock.patch('socket.socket')
+    @mock.patch('oslo.utils.netutils._get_my_ipv4_address')
+    def test_get_my_ip_socket_error(self, ip, mock_socket):
+        mock_socket.side_effect = socket.error
+        ip.return_value = '1.2.3.4'
+        addr = netutils.get_my_ipv4()
+        self.assertEqual(addr, '1.2.3.4')
+
+    @mock.patch('netifaces.gateways')
+    @mock.patch('netifaces.ifaddresses')
+    def test_get_my_ipv4_address_with_default_route(
+            self, ifaddr, gateways):
+        with patch.dict(netifaces.__dict__, {'AF_INET': '0'}):
+            ifaddr.return_value = {'0': [{'addr': '172.18.204.1'}]}
+            addr = netutils._get_my_ipv4_address()
+        self.assertEqual('172.18.204.1', addr)
+
+    @mock.patch('netifaces.gateways')
+    @mock.patch('netifaces.ifaddresses')
+    def test_get_my_ipv4_address_without_default_route(
+            self, ifaddr, gateways):
+        with patch.dict(netifaces.__dict__, {'AF_INET': '0'}):
+            ifaddr.return_value = {}
+            addr = netutils._get_my_ipv4_address()
+        self.assertEqual('127.0.0.1', addr)