DISPATCH-216 - Added code to skip system_tests_protocol_family.py if …

…IPv6 is turned off

(cherry picked from commit 70fa43c5fea58ccbc0cec590b43d08f173d7600a)

tests/router_policy_test.py

@@ -24,7 +24,7 @@
 from qpid_dispatch_internal.policy.policy_util import PolicyError
 from qpid_dispatch_internal.policy.policy_util import PolicyAppConnectionMgr
 from qpid_dispatch_internal.policy.policy_local import PolicyLocal
-from system_test import TestCase, main_module
+from system_test import TestCase, main_module, is_ipv6_enabled
 
 class PolicyHostAddrTest(TestCase):
 
@@ -61,7 +61,7 @@ def test_policy_hostaddr_ipv4(self):
         self.check_hostaddr_match(bbb, "1.1.2.0", False)
 
     def test_policy_hostaddr_ipv6(self):
-        if not HostAddr.has_ipv6:
+        if not is_ipv6_enabled():
             self.skipTest("System IPv6 support is not available")
         # Create simple host and range
         aaa = HostAddr("::1")
@@ -88,7 +88,7 @@ def test_policy_hostaddr_ipv4_wildcard(self):
 
 
     def test_policy_hostaddr_ipv6_wildcard(self):
-        if not HostAddr.has_ipv6:
+        if not is_ipv6_enabled():
             self.skipTest("System IPv6 support is not available")
         aaa = HostAddr("*")
         self.check_hostaddr_match(aaa,"::0")
@@ -101,7 +101,7 @@ def test_policy_malformed_hostaddr_ipv4(self):
         self.expect_deny( "9.9.9.9,8.8.8.8", "a > b")
 
     def test_policy_malformed_hostaddr_ipv6(self):
-        if not HostAddr.has_ipv6:
+        if not is_ipv6_enabled():
             self.skipTest("System IPv6 support is not available")
         self.expect_deny( "1::2::3", "Name or service not known")
         self.expect_deny( "::1,::2,::3", "arg count")

tests/system_test.py

@@ -143,6 +143,20 @@ def get_local_host_socket(protocol_family='IPv4'):
 
     return s, host
 
+def is_ipv6_enabled():
+    """
+    Returns true if IPV6 is enabled, false otherwise
+    """
+    sock = socket.socket(socket.AF_INET6, socket.SOCK_STREAM)
+    ipv6_enabled = True
+    try:
+        sock.bind(('::1', 0))
+    except Exception as e:
+        if "socket.error" in str(type(e)):
+            ipv6_enabled = False
+
+    return ipv6_enabled
+
 def port_available(port, protocol_family='IPv4'):
     """Return true if connecting to host:port gives 'connection refused'."""
     s, host = get_local_host_socket(protocol_family)

tests/system_tests_protocol_family.py

@@ -20,7 +20,7 @@
 import unittest
 from time import sleep
 from proton import Message
-from system_test import TestCase, Qdrouterd, main_module
+from system_test import TestCase, Qdrouterd, main_module, is_ipv6_enabled
 
 try:
     from proton import MODIFIED
@@ -31,6 +31,7 @@
 class ProtocolFamilyTest(TestCase):
     @classmethod
     def setUpClass(cls):
+
         """
         Starts three routers with various listeners and connectors.
         There is a call to wait_router_connected to make sure that the routers are able to communicate with each
@@ -71,6 +72,9 @@ def router(name, connection):
             # and tests each connector
             cls.routers.append(cls.tester.qdrouterd(name, config, wait=True))
 
+        if not is_ipv6_enabled():
+            return
+
         cls.routers = []
 
         inter_router_port = cls.tester.get_port(protocol_family='IPv6')
@@ -105,6 +109,10 @@ def router(name, connection):
     # If this test has started executing, it means that the setUpClass() has successfully executed which means that
     # the routers were able to communicate with each other successfully using the specified protocol family.
     def test_simple_pre_settled(self):
+
+        if not is_ipv6_enabled():
+            return self.skipTest("Skipping test..IPV6 not enabled")
+
         addr = self.routers[0].addresses[4]+"/test/1"
         M1 = self.messenger()
         M2 = self.messenger()