test: Check for host availability of IPv6 support properly

elprans · elprans · commit ab728dcaac4e · 2011-04-09T15:36:37.000-04:00
Check if host supports IPv6 protocol, and if not, skip the IPv6-related
bits.  Also, make sure that server listens on IPv6 loopback properly:
'localhost' may not always resolve to ::1.  In fact, a properly
configured system would most likely have 'localhost6' instead.  In any
case, it's better to use the IP addresses directly.
diff --git a/postgresql/test/test_connect.py b/postgresql/test/test_connect.py
@@ -5,6 +5,8 @@
 import os
 import unittest
 import atexit
+import socket
+import errno
 
 from ..python.socket import find_available_port
 
@@ -16,8 +18,31 @@
 from .. import driver as pg_driver
 from .. import open as pg_open
 
+
+def check_for_ipv6():
+	result = False
+	if socket.has_ipv6:
+		try:
+			socket.socket(socket.AF_INET6, socket.SOCK_STREAM)
+			result = True
+		except socket.error as e:
+			errs = [errno.EAFNOSUPPORT]
+			WSAEAFNOSUPPORT = getattr(errno, 'WSAEAFNOSUPPORT', None)
+			if WSAEAFNOSUPPORT is not None:
+				errs.append(WSAEAFNOSUPPORT)
+			if e.errno not in errs:
+				raise
+	return result
+
+
 msw = sys.platform in ('win32', 'win64')
 
+# win32 binaries don't appear to be built with ipv6
+has_ipv6 = check_for_ipv6() and not msw
+
+has_unix_sock = not msw
+
+
 class TestCaseWithCluster(unittest.TestCase):
 	"""
 	postgresql.driver *interface* tests.
@@ -50,11 +75,15 @@ def configure_cluster(self):
 				'failed to find a port for the test cluster on localhost',
 				creator = self.cluster
 			).raise_exception()
+
+		listen_addresses = '127.0.0.1'
+		if has_ipv6:
+			listen_addresses += ',::1'
 		self.cluster.settings.update(dict(
 			port = str(self.cluster_port),
 			max_connections = '6',
 			shared_buffers = '24',
-			listen_addresses = 'localhost',
+			listen_addresses = listen_addresses,
 			log_destination = 'stderr',
 			log_min_messages = 'FATAL',
 			silent_mode = 'off',
@@ -123,7 +152,6 @@ def __init__(self, *args, **kw):
 		super().__init__(*args,**kw)
 		# 8.4 nixed this.
 		self.do_crypt = self.cluster.installation.version_info < (8,4)
-		self.do_unix = sys.platform != msw
 
 	def configure_cluster(self):
 		super().configure_cluster()
@@ -135,7 +163,7 @@ def configure_cluster(self):
 		# Configure the hba file with the supported methods.
 		with open(self.cluster.hba_file, 'w') as hba:
 			hosts = ['0.0.0.0/0',]
-			if not msw:
+			if has_ipv6:
 				hosts.append('0::0/0')
 			methods = ['md5', 'password'] + (['crypt'] if self.do_crypt else [])
 			for h in hosts:
@@ -148,11 +176,11 @@ def configure_cluster(self):
 			# trusted
 			hba.writelines(["local all all trust\n"])
 			hba.writelines(["host test trusted 0.0.0.0/0 trust\n"])
-			if not msw:
+			if has_ipv6:
 				hba.writelines(["host test trusted 0::0/0 trust\n"])
 			# admin lines
 			hba.writelines(["host all test 0.0.0.0/0 trust\n"])
-			if not msw:
+			if has_ipv6:
 				hba.writelines(["host all test 0::0/0 trust\n"])
 
 	def initialize_database(self):
@@ -379,9 +407,7 @@ def test_IP4_connect(self):
 		with C() as c:
 			self.assertEqual(c.prepare('select 1').first(), 1)
 
-	if not msw:
-		# win32 binaries don't appear to be built with ipv6
-		# so filter this test on windows.
+	if has_ipv6:
 		def test_IP6_connect(self):
 			C = pg_driver.default.ip6(
 				user = 'test',
@@ -445,7 +471,7 @@ def test_trusted_connect(self):
 			self.assertEqual(c.prepare('select current_user').first(), 'trusted')
 
 	def test_Unix_connect(self):
-		if msw:
+		if not has_unix_sock:
 			return
 		unix_domain_socket = os.path.join(
 			self.cluster.data_directory,
@@ -460,7 +486,7 @@ def test_Unix_connect(self):
 			self.assertEqual(c.client_address, None)
 
 	def test_pg_open_unix(self):
-		if msw:
+		if not has_unix_sock:
 			return
 		unix_domain_socket = os.path.join(
 			self.cluster.data_directory,
