Issue #28683: Fix the tests that bind() a unix socket and raise PermissionError

on Android for a non-root user.

Lib/test/support/__init__.py

@@ -96,6 +96,7 @@
     "setswitchinterval", "android_not_root",
     # network
     "HOST", "IPV6_ENABLED", "find_unused_port", "bind_port", "open_urlresource",
+    "bind_unix_socket",
     # processes
     'temp_umask', "reap_children",
     # logging
@@ -708,6 +709,15 @@ def bind_port(sock, host=HOST):
     port = sock.getsockname()[1]
     return port
 
+def bind_unix_socket(sock, addr):
+    """Bind a unix socket, raising SkipTest if PermissionError is raised."""
+    assert sock.family == socket.AF_UNIX
+    try:
+        sock.bind(addr)
+    except PermissionError:
+        sock.close()
+        raise unittest.SkipTest('cannot bind AF_UNIX sockets')
+
 def _is_ipv6_enabled():
     """Check whether IPv6 is enabled on this host."""
     if socket.has_ipv6:

Lib/test/test_asyncore.py

@@ -95,6 +95,8 @@ def bind_af_aware(sock, addr):
     if HAS_UNIX_SOCKETS and sock.family == socket.AF_UNIX:
         # Make sure the path doesn't exist.
         support.unlink(addr)
+        support.bind_unix_socket(sock, addr)
+    else:
         sock.bind(addr)
 
 

Lib/test/test_pathlib.py

@@ -1888,7 +1888,8 @@ def test_is_socket_true(self):
         try:
             sock.bind(str(P))
         except OSError as e:
-            if "AF_UNIX path too long" in str(e):
+            if (isinstance(e, PermissionError) or
+                    "AF_UNIX path too long" in str(e)):
                 self.skipTest("cannot bind Unix socket: " + str(e))
         self.assertTrue(P.is_socket())
         self.assertFalse(P.is_fifo())

Lib/test/test_socket.py

@@ -278,7 +278,13 @@ def _tearDown(self):
 
     def clientRun(self, test_func):
         self.server_ready.wait()
+        try:
             self.clientSetUp()
+        except BaseException as e:
+            self.queue.put(e)
+            self.clientTearDown()
+            return
+        finally:
             self.client_ready.set()
         if self.server_crashed:
             self.clientTearDown()
@@ -520,8 +526,11 @@ def clientSetUp(self):
         self.serv_conn = self.cli
 
     def clientTearDown(self):
+        try:
             self.serv_conn.close()
             self.serv_conn = None
+        except AttributeError:
+            pass
         super().clientTearDown()
 
 
@@ -540,7 +549,7 @@ def setUp(self):
 
     def bindSock(self, sock):
         path = tempfile.mktemp(dir=self.dir_path)
-        sock.bind(path)
+        support.bind_unix_socket(sock, path)
         self.addCleanup(support.unlink, path)
 
 class UnixStreamBase(UnixSocketTestBase):
@@ -4631,7 +4640,7 @@ def encoded(self, path):
     def bind(self, sock, path):
         # Bind the socket
         try:
-            sock.bind(path)
+            support.bind_unix_socket(sock, path)
         except OSError as e:
             if str(e) == "AF_UNIX path too long":
                 self.skipTest(

Misc/NEWS

@@ -47,6 +47,9 @@ Windows
 Tests
 -----
 
+- Issue #28683: Fix the tests that bind() a unix socket and raise
+  PermissionError on Android for a non-root user.
+
 - Issue #26939: Add the support.setswitchinterval() function to fix
   test_functools hanging on the Android armv7 qemu emulator.