import contextlib
import errno
+import os.path
import socket
-import unittest
import sys
+import tempfile
+import unittest
from .. import support
from . import warnings_helper
# __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)
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',
'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',
import socket
import socketserver
import sys
-import tempfile
import threading
-import time
import unittest
import weakref
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
def gen_unix_socket_path():
- with tempfile.NamedTemporaryFile() as file:
- return file.name
+ return socket_helper.create_unix_domain_name()
@contextlib.contextmanager
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):
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):
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):
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):
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:
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."""
# 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."""
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):
import select
import signal
import socket
-import tempfile
import threading
import unittest
import socketserver
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
--- /dev/null
+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.
projects / thirdparty / Python / cpython.git / commitdiff
