From aa6e452039a8bdb0b92ac741bcbfee48c14f95e8 Mon Sep 17 00:00:00 2001 From: Sviatoslav Sydorenko Date: Sun, 9 Jun 2019 15:32:39 +0200 Subject: [PATCH] Migrate unix_sockname fixture to conftest module --- tests/conftest.py | 96 +++++++++++++++++++++++++++++++++++++++- tests/test_connector.py | 86 ----------------------------------- tests/test_web_runner.py | 7 ++- 3 files changed, 97 insertions(+), 92 deletions(-) diff --git a/tests/conftest.py b/tests/conftest.py index c8b3749e1ce..0a296f1ab29 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -1,11 +1,25 @@ -import hashlib +import os +import socket import ssl +import sys +from hashlib import md5, sha256 +from pathlib import Path +from tempfile import TemporaryDirectory import pytest import trustme pytest_plugins = ['aiohttp.pytest_plugin', 'pytester'] +IS_HPUX = sys.platform.startswith('hp-ux') +"""Specifies whether the current runtime is HP-UX.""" +IS_LINUX = sys.platform.startswith('linux') +"""Specifies whether the current runtime is HP-UX.""" +IS_UNIX = hasattr(socket, 'AF_UNIX') +"""Specifies whether the current runtime is *NIX.""" + +needs_unix = pytest.mark.skipif(not IS_UNIX, reason='requires UNIX sockets') + @pytest.fixture def tls_certificate_authority(): @@ -55,4 +69,82 @@ def tls_certificate_pem_bytes(tls_certificate): @pytest.fixture def tls_certificate_fingerprint_sha256(tls_certificate_pem_bytes): tls_cert_der = ssl.PEM_cert_to_DER_cert(tls_certificate_pem_bytes.decode()) - return hashlib.sha256(tls_cert_der).digest() + return sha256(tls_cert_der).digest() + + +@pytest.fixture +@needs_unix +def unix_sockname(tmp_path, tmp_path_factory): + """Generate an fs path to the UNIX domain socket for testing. + + N.B. Different OS kernels have different fs path length limitations + for it. For Linux, it's 108, for HP-UX it's 92 (or higher) depending + on its version. For for most of the BSDs (Open, Free, macOS) it's + mostly 104 but sometimes it can be down to 100. + + Ref: https://github.com/aio-libs/aiohttp/issues/3572 + """ + max_sock_len = 92 if IS_HPUX else 108 if IS_LINUX else 100 + """Amount of bytes allocated for the UNIX socket path by OS kernel. + + Ref: https://unix.stackexchange.com/a/367012/27133 + """ + + sock_file_name = 'unix.sock' + + root_tmp_dir = Path('/tmp').resolve() + os_tmp_dir = Path(os.getenv('TMPDIR', '/tmp')).resolve() + original_base_tmp_path = Path(tmp_path_factory.getbasetemp()) + + original_base_tmp_path_hash = md5( + str(original_base_tmp_path).encode(), + ).hexdigest() + + def make_tmp_dir(base_tmp_dir): + return TemporaryDirectory( + dir=base_tmp_dir, + prefix='pt-', + suffix='-{}'.format(original_base_tmp_path_hash), + ) + + def assert_sock_fits(sock_path): + sock_path_len = len(sock_path.encode()) + # exit-check to verify that it's correct and simplify debugging + # in the future + assert sock_path_len <= max_sock_len, ( + 'Suggested UNIX socket ({sock_path}) is {sock_path_len} bytes ' + 'long but the current kernel only has {max_sock_len} bytes ' + 'allocated to hold it so it must be shorter. ' + 'See https://github.com/aio-libs/aiohttp/issues/3572 ' + 'for more info.' + ).format_map(locals()) + + sock_path = str(tmp_path.resolve() / sock_file_name) + sock_path_len = len(sock_path.encode()) + + if original_base_tmp_path == root_tmp_dir and os_tmp_dir == root_tmp_dir: + assert_sock_fits(sock_path) + + if sock_path_len <= max_sock_len: + yield sock_path + return + + with make_tmp_dir(os_tmp_dir) as tmpd: + sock_path = str(tmpd.resolve() / sock_file_name) + sock_path_len = len(sock_path.encode()) + + if os_tmp_dir == root_tmp_dir: + assert_sock_fits(sock_path) + # exit-check to verify that it's correct and simplify debugging + # in the future + if sock_path_len <= max_sock_len: + yield sock_path + return + + with make_tmp_dir(root_tmp_dir) as tmpd: + sock_path = str(tmpd.resolve() / sock_file_name) + + assert_sock_fits(sock_path) + + yield sock_path + return diff --git a/tests/test_connector.py b/tests/test_connector.py index e39d40f1178..20adf274da5 100644 --- a/tests/test_connector.py +++ b/tests/test_connector.py @@ -3,16 +3,12 @@ import asyncio import gc import hashlib -import os import platform import socket import ssl import sys import uuid from collections import deque -from hashlib import md5 -from pathlib import Path -from tempfile import TemporaryDirectory from unittest import mock import pytest @@ -27,10 +23,6 @@ from aiohttp.test_utils import make_mocked_coro, unused_port from aiohttp.tracing import Trace -IS_HPUX = sys.platform.startswith('hp-ux') -"""Specifies whether the current runtime is HP-UX.""" -IS_LINUX = sys.platform.startswith('linux') -"""Specifies whether the current runtime is HP-UX.""" IS_UNIX = hasattr(socket, 'AF_UNIX') """Specifies whether the current runtime is *NIX.""" @@ -55,84 +47,6 @@ def ssl_key(): return ConnectionKey('localhost', 80, True, None, None, None, None) -@pytest.fixture -@needs_unix -def unix_sockname(tmp_path, tmp_path_factory): - """Generate an fs path to the UNIX domain socket for testing. - - N.B. Different OS kernels have different fs path length limitations - for it. For Linux, it's 108, for HP-UX it's 92 (or higher) depending - on its version. For for most of the BSDs (Open, Free, macOS) it's - mostly 104 but sometimes it can be down to 100. - - Ref: https://github.com/aio-libs/aiohttp/issues/3572 - """ - max_sock_len = 92 if IS_HPUX else 108 if IS_LINUX else 100 - """Amount of bytes allocated for the UNIX socket path by OS kernel. - - Ref: https://unix.stackexchange.com/a/367012/27133 - """ - - sock_file_name = 'unix.sock' - - root_tmp_dir = Path('/tmp').resolve() - os_tmp_dir = Path(os.getenv('TMPDIR', '/tmp')).resolve() - original_base_tmp_path = Path(tmp_path_factory.getbasetemp()) - - original_base_tmp_path_hash = md5( - str(original_base_tmp_path).encode(), - ).hexdigest() - - def make_tmp_dir(base_tmp_dir): - return TemporaryDirectory( - dir=base_tmp_dir, - prefix='pt-', - suffix='-{}'.format(original_base_tmp_path_hash), - ) - - def assert_sock_fits(sock_path): - sock_path_len = len(sock_path.encode()) - # exit-check to verify that it's correct and simplify debugging - # in the future - assert sock_path_len <= max_sock_len, ( - 'Suggested UNIX socket ({sock_path}) is {sock_path_len} bytes ' - 'long but the current kernel only has {max_sock_len} bytes ' - 'allocated to hold it so it must be shorter. ' - 'See https://github.com/aio-libs/aiohttp/issues/3572 ' - 'for more info.' - ).format_map(locals()) - - sock_path = str(tmp_path.resolve() / sock_file_name) - sock_path_len = len(sock_path.encode()) - - if original_base_tmp_path == root_tmp_dir and os_tmp_dir == root_tmp_dir: - assert_sock_fits(sock_path) - - if sock_path_len <= max_sock_len: - yield sock_path - return - - with make_tmp_dir(os_tmp_dir) as tmpd: - sock_path = str(tmpd.resolve() / sock_file_name) - sock_path_len = len(sock_path.encode()) - - if os_tmp_dir == root_tmp_dir: - assert_sock_fits(sock_path) - # exit-check to verify that it's correct and simplify debugging - # in the future - if sock_path_len <= max_sock_len: - yield sock_path - return - - with make_tmp_dir(root_tmp_dir) as tmpd: - sock_path = str(tmpd.resolve() / sock_file_name) - - assert_sock_fits(sock_path) - - yield sock_path - return - - @pytest.fixture def unix_server(loop, unix_sockname): runners = [] diff --git a/tests/test_web_runner.py b/tests/test_web_runner.py index 6a94ae1df82..dbaaf1b17ae 100644 --- a/tests/test_web_runner.py +++ b/tests/test_web_runner.py @@ -102,15 +102,14 @@ def test_non_app() -> None: @pytest.mark.skipif(platform.system() == "Windows", reason="Unix socket support is required") -async def test_addresses(make_runner, tmpdir) -> None: +async def test_addresses(make_runner, unix_sockname) -> None: _sock = get_unused_port_socket('127.0.0.1') runner = make_runner() await runner.setup() tcp = web.SockSite(runner, _sock) await tcp.start() - path = str(tmpdir / 'tmp.sock') - unix = web.UnixSite(runner, path) + unix = web.UnixSite(runner, unix_sockname) await unix.start() actual_addrs = runner.addresses expected_host, expected_post = _sock.getsockname()[:2] - assert actual_addrs == [(expected_host, expected_post), path] + assert actual_addrs == [(expected_host, expected_post), unix_sockname]