gh-93852: Add test.support.create_unix_domain_name() (#93914)

test_asyncio, test_logging, test_socket and test_socketserver now
create AF_UNIX domains in the current directory to no longer fail
with OSError("AF_UNIX path too long") if the temporary directory (the
TMPDIR environment variable) is too long.

Modify the following tests to use create_unix_domain_name():

* test_asyncio
* test_logging
* test_socket
* test_socketserver

test_asyncio.utils: remove unused time import.

Author: vstinner

Lib/test/support/socket_helper.py

@@ -1,8 +1,10 @@
 import contextlib
 import errno
+import os.path
 import socket
-import unittest
 import sys
+import tempfile
+import unittest
 
 from .. import support
 from . import warnings_helper
@@ -270,3 +272,14 @@ def filter_error(err):
     # __cause__ or __context__?
     finally:
         socket.setdefaulttimeout(old_timeout)
+
+
+def create_unix_domain_name():
+    """
+    Create a UNIX domain name: socket.bind() argument of a AF_UNIX socket.
+
+    Return a path relative to the current directory to get a short path
+    (around 27 ASCII characters).
+    """
+    return tempfile.mktemp(prefix="test_python_", suffix='.sock',
+                           dir=os.path.curdir)

Lib/test/test_asyncio/test_unix_events.py

@@ -315,11 +315,15 @@ def test_create_unix_connection_pathlib(self):
                 self.loop.run_until_complete(coro)
 
     def test_create_unix_server_existing_path_nonsock(self):
-        with tempfile.NamedTemporaryFile() as file:
-            coro = self.loop.create_unix_server(lambda: None, file.name)
-            with self.assertRaisesRegex(OSError,
-                                        'Address.*is already in use'):
-                self.loop.run_until_complete(coro)
+        path = test_utils.gen_unix_socket_path()
+        self.addCleanup(os_helper.unlink, path)
+        # create the file
+        open(path, "wb").close()
+
+        coro = self.loop.create_unix_server(lambda: None, path)
+        with self.assertRaisesRegex(OSError,
+                                    'Address.*is already in use'):
+            self.loop.run_until_complete(coro)
 
     def test_create_unix_server_ssl_bool(self):
         coro = self.loop.create_unix_server(lambda: None, path='spam',
@@ -356,20 +360,18 @@ def test_create_unix_server_path_dgram(self):
                          'no socket.SOCK_NONBLOCK (linux only)')
     @socket_helper.skip_unless_bind_unix_socket
     def test_create_unix_server_path_stream_bittype(self):
-        sock = socket.socket(
-            socket.AF_UNIX, socket.SOCK_STREAM | socket.SOCK_NONBLOCK)
-        with tempfile.NamedTemporaryFile() as file:
-            fn = file.name
-        try:
-            with sock:
-                sock.bind(fn)
-                coro = self.loop.create_unix_server(lambda: None, path=None,
-                                                    sock=sock)
-                srv = self.loop.run_until_complete(coro)
-                srv.close()
-                self.loop.run_until_complete(srv.wait_closed())
-        finally:
-            os.unlink(fn)
+        fn = test_utils.gen_unix_socket_path()
+        self.addCleanup(os_helper.unlink, fn)
+
+        sock = socket.socket(socket.AF_UNIX,
+                             socket.SOCK_STREAM | socket.SOCK_NONBLOCK)
+        with sock:
+            sock.bind(fn)
+            coro = self.loop.create_unix_server(lambda: None, path=None,
+                                                sock=sock)
+            srv = self.loop.run_until_complete(coro)
+            srv.close()
+            self.loop.run_until_complete(srv.wait_closed())
 
     def test_create_unix_server_ssl_timeout_with_plain_sock(self):
         coro = self.loop.create_unix_server(lambda: None, path='spam',

Lib/test/test_asyncio/utils.py

@@ -11,9 +11,7 @@
 import socket
 import socketserver
 import sys
-import tempfile
 import threading
-import time
 import unittest
 import weakref
 
@@ -34,6 +32,7 @@
 from asyncio import tasks
 from asyncio.log import logger
 from test import support
+from test.support import socket_helper
 from test.support import threading_helper
 
 
@@ -251,8 +250,7 @@ class UnixSSLWSGIServer(SSLWSGIServerMixin, SilentUnixWSGIServer):
 
 
     def gen_unix_socket_path():
-        with tempfile.NamedTemporaryFile() as file:
-            return file.name
+        return socket_helper.create_unix_domain_name()
 
 
     @contextlib.contextmanager

Lib/test/test_logging.py

@@ -1828,12 +1828,6 @@ def test_noserver(self):
         time.sleep(self.sock_hdlr.retryTime - now + 0.001)
         self.root_logger.error('Nor this')
 
-def _get_temp_domain_socket():
-    fn = make_temp_file(prefix='test_logging_', suffix='.sock')
-    # just need a name - file can't be present, or we'll get an
-    # 'address already in use' error.
-    os.remove(fn)
-    return fn
 
 @unittest.skipUnless(hasattr(socket, "AF_UNIX"), "Unix sockets required")
 class UnixSocketHandlerTest(SocketHandlerTest):
@@ -1845,13 +1839,10 @@ class UnixSocketHandlerTest(SocketHandlerTest):
 
     def setUp(self):
         # override the definition in the base class
-        self.address = _get_temp_domain_socket()
+        self.address = socket_helper.create_unix_domain_name()
+        self.addCleanup(os_helper.unlink, self.address)
         SocketHandlerTest.setUp(self)
 
-    def tearDown(self):
-        SocketHandlerTest.tearDown(self)
-        os_helper.unlink(self.address)
-
 @support.requires_working_socket()
 @threading_helper.requires_working_threading()
 class DatagramHandlerTest(BaseTest):
@@ -1928,13 +1919,10 @@ class UnixDatagramHandlerTest(DatagramHandlerTest):
 
     def setUp(self):
         # override the definition in the base class
-        self.address = _get_temp_domain_socket()
+        self.address = socket_helper.create_unix_domain_name()
+        self.addCleanup(os_helper.unlink, self.address)
         DatagramHandlerTest.setUp(self)
 
-    def tearDown(self):
-        DatagramHandlerTest.tearDown(self)
-        os_helper.unlink(self.address)
-
 @support.requires_working_socket()
 @threading_helper.requires_working_threading()
 class SysLogHandlerTest(BaseTest):
@@ -2022,13 +2010,10 @@ class UnixSysLogHandlerTest(SysLogHandlerTest):
 
     def setUp(self):
         # override the definition in the base class
-        self.address = _get_temp_domain_socket()
+        self.address = socket_helper.create_unix_domain_name()
+        self.addCleanup(os_helper.unlink, self.address)
         SysLogHandlerTest.setUp(self)
 
-    def tearDown(self):
-        SysLogHandlerTest.tearDown(self)
-        os_helper.unlink(self.address)
-
 @unittest.skipUnless(socket_helper.IPV6_ENABLED,
                      'IPv6 support required for this test.')
 class IPv6SysLogHandlerTest(SysLogHandlerTest):

Lib/test/test_socket.py

@@ -4,31 +4,30 @@
 from test.support import socket_helper
 from test.support import threading_helper
 
+import _thread as thread
+import array
+import contextlib
 import errno
 import io
 import itertools
-import socket
-import select
-import tempfile
-import time
-import traceback
-import queue
-import sys
-import os
-import platform
-import array
-import contextlib
-from weakref import proxy
-import signal
 import math
+import os
 import pickle
-import re
-import struct
+import platform
+import queue
 import random
-import shutil
+import re
+import select
+import signal
+import socket
 import string
-import _thread as thread
+import struct
+import sys
+import tempfile
 import threading
+import time
+import traceback
+from weakref import proxy
 try:
     import multiprocessing
 except ImportError:
@@ -605,17 +604,18 @@ class SocketTestBase(unittest.TestCase):
 
     def setUp(self):
         self.serv = self.newSocket()
+        self.addCleanup(self.close_server)
         self.bindServer()
 
+    def close_server(self):
+        self.serv.close()
+        self.serv = None
+
     def bindServer(self):
         """Bind server socket and set self.serv_addr to its address."""
         self.bindSock(self.serv)
         self.serv_addr = self.serv.getsockname()
 
-    def tearDown(self):
-        self.serv.close()
-        self.serv = None
-
 
 class SocketListeningTestMixin(SocketTestBase):
     """Mixin to listen on the server socket."""
@@ -700,15 +700,10 @@ class UnixSocketTestBase(SocketTestBase):
     # can't send anything that might be problematic for a privileged
     # user running the tests.
 
-    def setUp(self):
-        self.dir_path = tempfile.mkdtemp()
-        self.addCleanup(os.rmdir, self.dir_path)
-        super().setUp()
-
     def bindSock(self, sock):
-        path = tempfile.mktemp(dir=self.dir_path)
-        socket_helper.bind_unix_socket(sock, path)
+        path = socket_helper.create_unix_domain_name()
         self.addCleanup(os_helper.unlink, path)
+        socket_helper.bind_unix_socket(sock, path)
 
 class UnixStreamBase(UnixSocketTestBase):
     """Base class for Unix-domain SOCK_STREAM tests."""
@@ -1905,17 +1900,18 @@ def test_socket_fileno(self):
             self._test_socket_fileno(s, socket.AF_INET6, socket.SOCK_STREAM)
 
         if hasattr(socket, "AF_UNIX"):
-            tmpdir = tempfile.mkdtemp()
-            self.addCleanup(shutil.rmtree, tmpdir)
+            unix_name = socket_helper.create_unix_domain_name()
+            self.addCleanup(os_helper.unlink, unix_name)
+
             s = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
-            self.addCleanup(s.close)
-            try:
-                s.bind(os.path.join(tmpdir, 'socket'))
-            except PermissionError:
-                pass
-            else:
-                self._test_socket_fileno(s, socket.AF_UNIX,
-                                         socket.SOCK_STREAM)
+            with s:
+                try:
+                    s.bind(unix_name)
+                except PermissionError:
+                    pass
+                else:
+                    self._test_socket_fileno(s, socket.AF_UNIX,
+                                             socket.SOCK_STREAM)
 
     def test_socket_fileno_rejects_float(self):
         with self.assertRaises(TypeError):

Lib/test/test_socketserver.py

@@ -8,7 +8,6 @@
 import select
 import signal
 import socket
-import tempfile
 import threading
 import unittest
 import socketserver
@@ -98,8 +97,7 @@ def pickaddr(self, proto):
         else:
             # XXX: We need a way to tell AF_UNIX to pick its own name
             # like AF_INET provides port==0.
-            dir = None
-            fn = tempfile.mktemp(prefix='unix_socket.', dir=dir)
+            fn = socket_helper.create_unix_domain_name()
             self.test_files.append(fn)
             return fn
 

Misc/NEWS.d/next/Tests/2022-06-16-21-38-18.gh-issue-93852.U_Hl6s.rst

@@ -0,0 +1,4 @@
+test_asyncio, test_logging, test_socket and test_socketserver now create
+AF_UNIX domains in the current directory to no longer fail with
+``OSError("AF_UNIX path too long")`` if the temporary directory (the
+:envvar:`TMPDIR` environment variable) is too long. Patch by Victor Stinner.