Merge remote-tracking branch 'ldoktor/is_port_free2'

Signed-off-by: NCaio Carrara <ccarrara@redhat.com>

avocado/utils/network.py

@@ -23,29 +23,49 @@ from six.moves import xrange as range
 
 from .data_structures import Borg
 
+#: Families taken into account in this class
+FAMILIES = (socket.AF_INET, socket.AF_INET6)
+#: Protocols taken into account in this class
+PROTOCOLS = (socket.SOCK_STREAM, socket.SOCK_DGRAM)
+
 
 def is_port_free(port, address):
     """
     Return True if the given port is available for use.
 
+    Currently we only check for TCP/UDP connections on IPv4/6
+
     :param port: Port number
     :param address: Socket address to bind or connect
     """
+    if address == "localhost" or not address:
+        localhost = True
+        protocols = PROTOCOLS
+    else:
+        localhost = False
+        # sock.connect always connects for UDP
+        protocols = (socket.SOCK_STREAM, )
+    sock = None
     try:
-        s = socket.socket()
-        if address == "localhost":
-            s.bind((address, port))
-            free = True
-        else:
-            s.connect((address, port))
-            free = False
-    except socket.error:
-        if address == "localhost":
-            free = False
-        else:
-            free = True
-    s.close()
-    return free
+        for family in FAMILIES:
+            for protocol in protocols:
+                try:
+                    sock = socket.socket(family, protocol)
+                    if localhost:
+                        sock.bind(("", port))
+                    else:
+                        sock.connect((address, port))
+                        return False
+                except socket.error as exc:
+                    if exc.errno in (93, 94):   # Unsupported combinations
+                        continue
+                    if localhost:
+                        return False
+                sock.close()
+        return True
+    finally:
+        if sock is not None:
+            sock.close()
 
 
 def find_free_port(start_port=1024, end_port=65535, address="localhost", sequent=True):

python-avocado.spec

@@ -128,6 +128,7 @@ BuildRequires: libcdio
 BuildRequires: libvirt-python
 BuildRequires: perl-Test-Harness
 BuildRequires: psmisc
+BuildRequires: python2-netifaces
 %if 0%{?rhel}
 BuildRequires: PyYAML
 %else
@@ -136,6 +137,7 @@ BuildRequires: python2-yaml
 %if %{with_python3}
 BuildRequires: python3-libvirt
 BuildRequires: python3-yaml
+BuildRequires: python3-netifaces
 %endif
 %endif
 

requirements-selftests.txt

@@ -34,3 +34,6 @@ lxml==4.2.4
 # in make develop. The proper solution would be migrate
 # our make develop process to use pip
 libvirt-python==4.6.0
+
+# For avocado.utils.network selftests
+netifaces

selftests/unit/test_utils_network.py

+import netifaces
+import socket
 import unittest
 
 try:
@@ -33,5 +35,64 @@ class PortTrackerTest(unittest.TestCase):
         self.assertNotIn(22, tracker.retained_ports)
 
 
+def get_all_local_addrs():
+    """
+    Returns all ipv4/ipv6 addresses that are associated with this machine
+    """
+    ipv4_addrs = []
+    ipv6_addrs = []
+    for interface in netifaces.interfaces():
+        ifaddresses = netifaces.ifaddresses(interface)
+        ipv4_addrs += [_['addr']
+                       for _ in ifaddresses.get(netifaces.AF_INET, [])]
+        ipv6_addrs += [_['addr']
+                       for _ in ifaddresses.get(netifaces.AF_INET6, [])]
+    return ipv4_addrs, ipv6_addrs
+
+
+class FreePort(unittest.TestCase):
+
+    def test_is_port_free(self):
+        port = network.find_free_port(sequent=False)
+        self.assertTrue(network.is_port_free(port, "localhost"))
+        local_addrs = get_all_local_addrs()
+        ipv4_addrs = ["localhost", ""] + list(local_addrs[0])
+        ipv6_addrs = ["localhost", ""] + list(local_addrs[1])
+        good = []
+        bad = []
+        skip = []
+        sock = None
+        for family in network.FAMILIES:
+            if family == socket.AF_INET:
+                addrs = ipv4_addrs
+            else:
+                addrs = ipv6_addrs
+            for addr in addrs:
+                for protocol in network.PROTOCOLS:
+                    try:
+                        sock = socket.socket(family, protocol)
+                        sock.bind((addr, port))
+                        if network.is_port_free(port, "localhost"):
+                            bad.append("%s, %s, %s: reports free"
+                                       % (family, protocol, addr))
+                        else:
+                            good.append("%s, %s, %s" % (family, protocol,
+                                                        addr))
+                    except Exception as exc:
+                        if getattr(exc, 'errno', None) in (-2, 2, 22, 94):
+                            skip.append("%s, %s, %s: Not supported: %s"
+                                        % (family, protocol, addr, exc))
+                        else:
+                            bad.append("%s, %s, %s: Failed to bind: %s"
+                                       % (family, protocol, addr, exc))
+                    finally:
+                        if sock is not None:
+                            sock.close()
+        self.assertFalse(bad, "Following combinations failed:\n%s\n\n"
+                         "Following combinations passed:\n%s\n\n"
+                         "Following combinations were skipped:\n%s"
+                         % ("\n".join(bad), "\n".join(good), "\n".join(skip)))
+
+
 if __name__ == "__main__":
     unittest.main()