Issue #28683: Fix the tests that bind() a unix socket and raise PermissionError
on Android for a non-root user.
diff --git a/Lib/test/support/__init__.py b/Lib/test/support/__init__.py
index ed1af2b..15d8fc8 100644
--- a/Lib/test/support/__init__.py
+++ b/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 @@
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 dbee593..51c6573 100644
--- a/Lib/test/test_asyncore.py
+++ b/Lib/test/test_asyncore.py
@@ -95,7 +95,9 @@
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 65b2d5a..ce1ca15 100644
--- a/Lib/test/test_pathlib.py
+++ b/Lib/test/test_pathlib.py
@@ -1888,7 +1888,8 @@
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 59564c9..6b29e18 100644
--- a/Lib/test/test_socket.py
+++ b/Lib/test/test_socket.py
@@ -278,8 +278,14 @@
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 @@
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 @@
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 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 97502a9..35ad11f 100644
--- a/Misc/NEWS
+++ b/Misc/NEWS
@@ -47,6 +47,9 @@
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.