Relocate port finder into a new socket module.

James William Pye · James William Pye · commit 0f249fcbde3f · 2009-02-24T11:09:18.000-07:00
diff --git a/postgresql/python/socket.py b/postgresql/python/socket.py
@@ -0,0 +1,48 @@
+##
+# copyright 2009, James William Pye
+# http://python.projects.postgresql.org
+##
+"""
+socket tools
+"""
+import random
+import socket
+import math
+import errno
+
+__all__ = ['find_available_port']
+
+def find_available_port(
+	interface : "attempt to bind to interface" = 'localhost',
+	address_family : "address family to use (default: AF_INET)" = socket.AF_INET,
+	limit : "Number tries to make before giving up" = 1024,
+	port_range = (6600, 56600)
+) -> (int, None):
+	"""
+	Find an available port on the given interface for the given address family.
+
+	Returns a port number that was successfully bound to or `None` if the attempt
+	limit was reached.
+	"""
+	i = 0
+	while i < limit:
+		i += 1
+		port = (
+			math.floor(
+				random.random() * (port_range[1] - port_range[0])
+			) + port_range[0]
+		)
+		s = socket.socket(address_family, socket.SOCK_STREAM,)
+		try:
+			s.bind(('localhost', port))
+		except socket.error as e:
+			if e.errno in (errno.EACCES, errno.EADDRINUSE, errno.EINTR):
+				# try again
+				continue
+		finally:
+			s.close()
+		break
+	else:
+		port = None
+
+	return port
diff --git a/postgresql/unittest.py b/postgresql/unittest.py
@@ -7,16 +7,15 @@
 """
 import sys
 import os
-import random
-import socket
-import math
 import atexit
 import unittest
 
 from . import exceptions as pg_exc
 from . import cluster as pg_cluster
 from . import installation as pg_inn
 
+from .python.socket import find_available_port
+
 class TestCaseWithCluster(unittest.TestCase):
 	"""
 	postgresql.driver *interface* tests.
@@ -42,30 +41,11 @@ def __init__(self, *args, **kw):
 		if self.cluster.initialized():
 			self.cluster.drop()
 
-	def _gen_cluster_port(self):
-		i = 0
-		limit = 1024
-		while i < limit:
-			i += 1
-			port = (math.floor(random.random() * 50000) + 1024)
-			s = socket.socket(socket.AF_INET, socket.SOCK_STREAM,)
-			try:
-				s.bind(('localhost', port))
-			except socket.error as e:
-				if e.errno in (errno.EACCES, errno.EADDRINUSE, errno.EINTR):
-					# try again
-					continue
-			s.close()
-			break
-		else:
-			port = None
-		return port
-
 	def configure_cluster(self):
-		self.cluster_port = self._gen_cluster_port()
+		self.cluster_port = find_available_port()
 		if self.cluster_port is None:
 			e = pg_exc.ClusterError(
-				'failed to find a port for the test cluster'
+				'failed to find a port for the test cluster on localhost'
 			)
 			self.cluster.ife_descend(e)
 			e.raise_exception()
