diff --git a/Lib/test/support/__init__.py b/Lib/test/support/__init__.py index ed1af2b5a5d..15d8fc849b9 100644 --- a/Lib/test/support/__init__.py +++ b/Lib/test/support/__init__.py @@ -96,6 +96,7 @@ __all__ = [ "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: diff --git a/Lib/test/test_asyncore.py b/Lib/test/test_asyncore.py index dbee593c544..51c65737a44 100644 --- a/Lib/test/test_asyncore.py +++ b/Lib/test/test_asyncore.py @@ -95,7 +95,9 @@ 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) - sock.bind(addr) + support.bind_unix_socket(sock, addr) + else: + sock.bind(addr) class HelperFunctionTests(unittest.TestCase): diff --git a/Lib/test/test_pathlib.py b/Lib/test/test_pathlib.py index 65b2d5abda1..ce1ca153ed0 100644 --- a/Lib/test/test_pathlib.py +++ b/Lib/test/test_pathlib.py @@ -1888,7 +1888,8 @@ class _BasePathTest(object): 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()) diff --git a/Lib/test/test_socket.py b/Lib/test/test_socket.py index 59564c9063d..6b29e182348 100644 --- a/Lib/test/test_socket.py +++ b/Lib/test/test_socket.py @@ -278,8 +278,14 @@ class ThreadableTest: def clientRun(self, test_func): self.server_ready.wait() - self.clientSetUp() - self.client_ready.set() + 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() return @@ -520,8 +526,11 @@ class ConnectedStreamTestMixin(SocketListeningTestMixin, self.serv_conn = self.cli def clientTearDown(self): - self.serv_conn.close() - self.serv_conn = None + try: + self.serv_conn.close() + self.serv_conn = None + except AttributeError: + pass super().clientTearDown() @@ -540,7 +549,7 @@ class UnixSocketTestBase(SocketTestBase): 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 @@ class TestUnixDomain(unittest.TestCase): 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( diff --git a/Misc/NEWS b/Misc/NEWS index 97502a999b0..35ad11ff507 100644 --- a/Misc/NEWS +++ b/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.