[Python-checkins] gh-93852: Add test.support.create_unix_domain_name() (#93914)
vstinner
webhook-mailer at python.org
Fri Jun 17 07:17:23 EDT 2022
https://github.com/python/cpython/commit/c5b750dc0b4d4e58047c9d93c635fa26b06562f7
commit: c5b750dc0b4d4e58047c9d93c635fa26b06562f7
branch: main
author: Victor Stinner <vstinner at python.org>
committer: vstinner <vstinner at python.org>
date: 2022-06-17T13:16:51+02:00
summary:
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.
files:
A Misc/NEWS.d/next/Tests/2022-06-16-21-38-18.gh-issue-93852.U_Hl6s.rst
M Lib/test/support/socket_helper.py
M Lib/test/test_asyncio/test_unix_events.py
M Lib/test/test_asyncio/utils.py
M Lib/test/test_logging.py
M Lib/test/test_socket.py
M Lib/test/test_socketserver.py
diff --git a/Lib/test/support/socket_helper.py b/Lib/test/support/socket_helper.py
index 42b2a93398cbf..d2960c9e33347 100644
--- a/Lib/test/support/socket_helper.py
+++ b/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)
diff --git a/Lib/test/test_asyncio/test_unix_events.py b/Lib/test/test_asyncio/test_unix_events.py
index 2f68459d30cd4..23d2e8ad40164 100644
--- a/Lib/test/test_asyncio/test_unix_events.py
+++ b/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',
diff --git a/Lib/test/test_asyncio/utils.py b/Lib/test/test_asyncio/utils.py
index 507daa11c28bc..96be5a1c3bcf7 100644
--- a/Lib/test/test_asyncio/utils.py
+++ b/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
diff --git a/Lib/test/test_logging.py b/Lib/test/test_logging.py
index f4a4324280539..7859c604f0ef5 100644
--- a/Lib/test/test_logging.py
+++ b/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):
diff --git a/Lib/test/test_socket.py b/Lib/test/test_socket.py
index c98190382416c..9a8c3b6766715 100644
--- a/Lib/test/test_socket.py
+++ b/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):
diff --git a/Lib/test/test_socketserver.py b/Lib/test/test_socketserver.py
index 2edb1e0c0e21e..2fa5069423327 100644
--- a/Lib/test/test_socketserver.py
+++ b/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
diff --git a/Misc/NEWS.d/next/Tests/2022-06-16-21-38-18.gh-issue-93852.U_Hl6s.rst b/Misc/NEWS.d/next/Tests/2022-06-16-21-38-18.gh-issue-93852.U_Hl6s.rst
new file mode 100644
index 0000000000000..ce86eead02660
--- /dev/null
+++ b/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.
More information about the Python-checkins
mailing list