filter connections 2

psutil/tests/__init__.py

@@ -47,6 +47,7 @@
 from psutil import SUNOS
 from psutil import WINDOWS
 from psutil._common import bytes2human
+from psutil._common import debug
 from psutil._common import memoize
 from psutil._common import print_color
 from psutil._common import supports_ipv6
@@ -105,7 +106,7 @@
     # sync primitives
     'call_until', 'wait_for_pid', 'wait_for_file',
     # network
-    'check_net_address',
+    'check_net_address', 'filter_proc_connections',
     'get_free_port', 'bind_socket', 'bind_unix_socket', 'tcp_socketpair',
     'unix_socketpair', 'create_sockets',
     # compat
@@ -1871,6 +1872,20 @@ def check_status(conn):
     check_status(conn)
 
 
+def filter_proc_connections(cons):
+    """Our process may start with some open UNIX sockets which are not
+    initialized by us, invalidating unit tests.
+    """
+    new = []
+    for conn in cons:
+        if POSIX and conn.family == socket.AF_UNIX:
+            if MACOS and "/syslog" in conn.raddr:
+                debug("skipping %s" % str(conn))
+                continue
+        new.append(conn)
+    return new
+
+
 # ===================================================================
 # --- compatibility
 # ===================================================================

psutil/tests/test_connections.py

@@ -25,7 +25,6 @@
 from psutil import POSIX
 from psutil import SUNOS
 from psutil import WINDOWS
-from psutil._common import debug
 from psutil._common import supports_ipv6
 from psutil._compat import PY3
 from psutil.tests import AF_UNIX
@@ -36,6 +35,7 @@
 from psutil.tests import bind_unix_socket
 from psutil.tests import check_connection_ntuple
 from psutil.tests import create_sockets
+from psutil.tests import filter_proc_connections
 from psutil.tests import reap_children
 from psutil.tests import retry_on_failure
 from psutil.tests import serialrun
@@ -49,20 +49,10 @@
 
 
 def this_proc_connections(kind):
-    # This process may use some UNIX sockets which are not initialized
-    # by us. Let's filter them.
     cons = psutil.Process().connections(kind=kind)
     if kind in ("all", "unix"):
-        new = []
-        for conn in cons:
-            if POSIX and conn.family == socket.AF_UNIX:
-                if MACOS and "/syslog" in conn.raddr:
-                    debug("skipping %r" % conn)
-                    continue
-            new.append(conn)
-        return new
-    else:
-        return cons
+        return filter_proc_connections(cons)
+    return cons
 
 
 @serialrun

psutil/tests/test_testutils.py

@@ -36,6 +36,7 @@
 from psutil.tests import call_until
 from psutil.tests import chdir
 from psutil.tests import create_sockets
+from psutil.tests import filter_proc_connections
 from psutil.tests import get_free_port
 from psutil.tests import is_namedtuple
 from psutil.tests import mock
@@ -318,7 +319,9 @@ def tcp_tcp_socketpair(self):
     def test_unix_socketpair(self):
         p = psutil.Process()
         num_fds = p.num_fds()
-        self.assertEqual(p.connections(kind='unix'), [])
+        self.assertEqual(
+            filter_proc_connections(p.connections(kind='unix')), []
+        )
         name = self.get_testfn()
         server, client = unix_socketpair(name)
         try: