define a reusable bind_unix_socket() test utility

3 files changed, 49 insertions, 42 deletions

diff --git a/psutil/tests/__init__.py b/psutil/tests/__init__.py
index 42db5e1d..04f434ee 100644
--- a/psutil/tests/__init__.py
+++ b/psutil/tests/__init__.py
@@ -43,6 +43,7 @@ except ImportError:
 
 import psutil
 from psutil import LINUX
+from psutil import OSX
 from psutil import POSIX
 from psutil import WINDOWS
 from psutil._compat import PY3
@@ -850,6 +851,29 @@ def check_connection_ntuple(conn):
                     assert dupsock.type == conn.type
 
 
+def bind_unix_socket(type=socket.SOCK_STREAM, suffix="", mode=0o600):
+    """Creates a listening unix socket.
+    Return a (sock, filemame) tuple.
+    """
+    # TODO: for some reason on OSX a UNIX socket cannot be
+    # deleted once created (EACCES) so we create a temp file
+    # which will remain around. :-\
+    if OSX:
+        file = tempfile.mktemp(prefix=TESTFILE_PREFIX, suffix=suffix)
+    else:
+        file = TESTFN + suffix
+    assert not os.path.exists(file), file
+    sock = socket.socket(socket.AF_UNIX, type)
+    try:
+        sock.bind(file)
+    except Exception:
+        sock.close()
+        raise
+    if mode is not None:
+        os.chmod(file, mode)
+    return (sock, file)
+
+
 # ===================================================================
 # --- others
 # ===================================================================
diff --git a/psutil/tests/test_process.py b/psutil/tests/test_process.py
index 8d25aee8..3e971cc8 100755
--- a/psutil/tests/test_process.py
+++ b/psutil/tests/test_process.py
@@ -44,6 +44,7 @@ from psutil._compat import PY3
 from psutil._compat import unicode
 from psutil.tests import AF_UNIX
 from psutil.tests import APPVEYOR
+from psutil.tests import bind_unix_socket
 from psutil.tests import call_until
 from psutil.tests import check_connection_ntuple
 from psutil.tests import create_exe
@@ -1101,25 +1102,21 @@ class TestProcess(unittest.TestCase):
     def test_connections_unix(self):
         def check(type):
             safe_rmpath(TESTFN)
-            # TODO: for some reason on OSX a UNIX socket cannot be
-            # deleted once created (EACCES) so we create a temp file
-            # which will remain around. :-\
-            tfile = tempfile.mktemp(prefix=TESTFILE_PREFIX) if OSX else TESTFN
-            sock = socket.socket(AF_UNIX, type)
-            with contextlib.closing(sock):
-                sock.bind(tfile)
-                cons = psutil.Process().connections(kind='unix')
-                conn = cons[0]
-                check_connection_ntuple(conn)
-                if conn.fd != -1:  # != sunos and windows
-                    self.assertEqual(conn.fd, sock.fileno())
-                self.assertEqual(conn.family, AF_UNIX)
-                self.assertEqual(conn.type, type)
-                self.assertEqual(conn.laddr, tfile)
-                if not SUNOS:
-                    # XXX Solaris can't retrieve system-wide UNIX
-                    # sockets.
-                    self.compare_proc_sys_cons(os.getpid(), cons)
+            sock, name = bind_unix_socket(type=type)
+            self.addCleanup(sock.close)
+            self.addCleanup(safe_rmpath, name)
+            cons = psutil.Process().connections(kind='unix')
+            conn = cons[0]
+            check_connection_ntuple(conn)
+            if conn.fd != -1:  # != sunos and windows
+                self.assertEqual(conn.fd, sock.fileno())
+            self.assertEqual(conn.family, AF_UNIX)
+            self.assertEqual(conn.type, type)
+            self.assertEqual(conn.laddr, name)
+            if not SUNOS:
+                # XXX Solaris can't retrieve system-wide UNIX
+                # sockets.
+                self.compare_proc_sys_cons(os.getpid(), cons)
 
         check(SOCK_STREAM)
         check(SOCK_DGRAM)
diff --git a/psutil/tests/test_unicode.py b/psutil/tests/test_unicode.py
index a885668b..f79131bb 100644
--- a/psutil/tests/test_unicode.py
+++ b/psutil/tests/test_unicode.py
@@ -50,16 +50,14 @@ returning variable str/unicode types, see:
 https://github.com/giampaolo/psutil/issues/655#issuecomment-136131180
 """
 
-import contextlib
 import os
 import socket
-import tempfile
 
 from psutil import BSD
-from psutil import OSX
 from psutil import WINDOWS
 from psutil._compat import PY3
 from psutil.tests import ASCII_FS
+from psutil.tests import bind_unix_socket
 from psutil.tests import chdir
 from psutil.tests import create_exe
 from psutil.tests import get_test_subprocess
@@ -67,7 +65,6 @@ from psutil.tests import reap_children
 from psutil.tests import run_test_module_by_name
 from psutil.tests import safe_mkdir
 from psutil.tests import safe_rmpath
-from psutil.tests import TESTFILE_PREFIX
 from psutil.tests import TESTFN
 from psutil.tests import TESTFN_UNICODE
 from psutil.tests import unittest
@@ -150,25 +147,14 @@ class _BaseFSAPIsTests(object):
 
     @unittest.skipUnless(hasattr(socket, "AF_UNIX"), "AF_UNIX not supported")
     def test_connections(self):
-        safe_rmpath(TESTFN)
-        # TODO: for some reason on OSX a UNIX socket cannot be
-        # deleted once created (EACCES) so we create a temp file
-        # which will remain around. :-\
-        if OSX:
-            tfile = tempfile.mktemp(
-                prefix=TESTFILE_PREFIX + self.funky_name)
-        else:
-            tfile = self.funky_name
-        self.addCleanup(safe_rmpath, tfile)
-
-        sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
-        with contextlib.closing(sock):
-            try:
-                sock.bind(tfile)
-            except (socket.error, UnicodeEncodeError):
-                raise unittest.SkipTest("not supported")
-            conn = psutil.Process().connections(kind='unix')[0]
-            self.assertEqual(conn.laddr, tfile)
+        try:
+            sock, name = bind_unix_socket(suffix=self.funky_name)
+        except (socket.error, UnicodeEncodeError):
+            raise unittest.SkipTest("not supported")
+        self.addCleanup(safe_rmpath, name)
+        self.addCleanup(sock.close)
+        conn = psutil.Process().connections(kind='unix')[0]
+        self.assertEqual(conn.laddr, name)
 
     def test_disk_usage(self):
         safe_mkdir(self.funky_name)