Migrate unix_sockname fixture to conftest module

aio-libs · Jun 9, 2019 · aa6e452 · aa6e452
1 parent 52db4a1
commit aa6e452
Show file tree

Hide file tree

Showing 3 changed files with 97 additions and 92 deletions.
diff --git 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
@@ -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
@@ -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]