diff --git a/tests/test_sockets.py b/tests/test_sockets.py index f34a0381..1a748c5a 100644 --- a/tests/test_sockets.py +++ b/tests/test_sockets.py @@ -5,7 +5,9 @@ import io import os import platform +import random import socket +import string import sys import threading import time @@ -14,7 +16,7 @@ from socket import AddressFamily from ssl import SSLContext, SSLError from threading import Thread -from typing import Any, Iterable, Iterator, NoReturn, TypeVar, cast +from typing import Any, Generator, Iterable, Iterator, NoReturn, TypeVar, cast import psutil import pytest @@ -702,14 +704,41 @@ async def test_bind_link_local(self) -> None: pass +class TestUNIX: + # MacOS requires unix socket paths under 107 bytes, but the default mktemp + # implementation generates paths that are too long under /private/var/folders + # + # Linux has a similar limitation of 108 bytes, but doesn't normally hit this + # case because the mktemp implementation creates folders in /tmp. However, there + # are cases where mount points could cause /tmp to resolve to a path over the + # limit + # + # Reference: https://github.com/python/cpython/issues/93852 + @classmethod + def _short_socket_path(cls) -> Path: + name = "".join(random.choice(string.ascii_letters) for _ in range(10)) + socket_path = Path("/tmp") / name + return socket_path + + # Harcode an implementation that should work in both cases by locating the + # socket directly under /tmp, and removing it when done + @pytest.fixture + def socket_path(self) -> Generator[Path, None, None]: + socket_path = TestUNIX._short_socket_path() + yield socket_path + socket_path.unlink(missing_ok=True) + + @pytest.fixture + def peer_socket_path(self) -> Generator[Path, None, None]: + socket_path = TestUNIX._short_socket_path() + yield socket_path + socket_path.unlink(missing_ok=True) + + @pytest.mark.skipif( sys.platform == "win32", reason="UNIX sockets are not available on Windows" ) -class TestUNIXStream: - @pytest.fixture - def socket_path(self, tmp_path_factory: TempPathFactory) -> Path: - return tmp_path_factory.mktemp("unix").joinpath("socket") - +class TestUNIXStream(TestUNIX): @pytest.fixture(params=[False, True], ids=["str", "path"]) def socket_path_or_str(self, request: SubRequest, socket_path: Path) -> Path | str: return socket_path if request.param else str(socket_path) @@ -1024,11 +1053,7 @@ async def test_connecting_with_non_utf8(self, socket_path: Path) -> None: @pytest.mark.skipif( sys.platform == "win32", reason="UNIX sockets are not available on Windows" ) -class TestUNIXListener: - @pytest.fixture - def socket_path(self, tmp_path_factory: TempPathFactory) -> Path: - return tmp_path_factory.mktemp("unix").joinpath("socket") - +class TestUNIXListener(TestUNIX): @pytest.fixture(params=[False, True], ids=["str", "path"]) def socket_path_or_str(self, request: SubRequest, socket_path: Path) -> Path | str: return socket_path if request.param else str(socket_path) @@ -1421,19 +1446,11 @@ async def test_send_after_close(self, family: AnyIPAddressFamily) -> None: @pytest.mark.skipif( sys.platform == "win32", reason="UNIX sockets are not available on Windows" ) -class TestUNIXDatagramSocket: - @pytest.fixture - def socket_path(self, tmp_path_factory: TempPathFactory) -> Path: - return tmp_path_factory.mktemp("unix").joinpath("socket") - +class TestUNIXDatagramSocket(TestUNIX): @pytest.fixture(params=[False, True], ids=["str", "path"]) def socket_path_or_str(self, request: SubRequest, socket_path: Path) -> Path | str: return socket_path if request.param else str(socket_path) - @pytest.fixture - def peer_socket_path(self, tmp_path_factory: TempPathFactory) -> Path: - return tmp_path_factory.mktemp("unix").joinpath("peer_socket") - async def test_extra_attributes(self, socket_path: Path) -> None: async with await create_unix_datagram_socket(local_path=socket_path) as unix_dg: raw_socket = unix_dg.extra(SocketAttribute.raw_socket) @@ -1543,19 +1560,11 @@ async def test_local_path_invalid_ascii(self, socket_path: Path) -> None: @pytest.mark.skipif( sys.platform == "win32", reason="UNIX sockets are not available on Windows" ) -class TestConnectedUNIXDatagramSocket: - @pytest.fixture - def socket_path(self, tmp_path_factory: TempPathFactory) -> Path: - return tmp_path_factory.mktemp("unix").joinpath("socket") - +class TestConnectedUNIXDatagramSocket(TestUNIX): @pytest.fixture(params=[False, True], ids=["str", "path"]) def socket_path_or_str(self, request: SubRequest, socket_path: Path) -> Path | str: return socket_path if request.param else str(socket_path) - @pytest.fixture - def peer_socket_path(self, tmp_path_factory: TempPathFactory) -> Path: - return tmp_path_factory.mktemp("unix").joinpath("peer_socket") - @pytest.fixture(params=[False, True], ids=["peer_str", "peer_path"]) def peer_socket_path_or_str( self, request: SubRequest, peer_socket_path: Path