Simplified the tests

Author: agronholm

tests/conftest.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import asyncio
+import platform
 import ssl
 import sys
 from collections.abc import Generator, Iterator
@@ -9,6 +10,7 @@
 from unittest.mock import Mock
 
 import pytest
+import sniffio
 import trustme
 from _pytest.fixtures import SubRequest
 from trustme import CA
@@ -18,7 +20,10 @@
 
 uvloop_marks = []
 try:
-    import uvloop
+    if platform.system() == "Windows":
+        import winloop as uvloop
+    else:
+        import uvloop
 except ImportError:
     uvloop_marks.append(pytest.mark.skip(reason="uvloop not available"))
     uvloop = Mock()
@@ -27,32 +32,22 @@
         uvloop.loop.Loop, "shutdown_default_executor"
     ):
         uvloop_marks.append(
-            pytest.mark.skip(reason="uvloop is missing shutdown_default_executor()")
+            pytest.mark.skip(
+                reason=f"{uvloop.__name__} is missing shutdown_default_executor()"
+            )
         )
 
-winloop_marks = []
-try:
-    import winloop
-except ImportError:
-    winloop_marks.append(pytest.mark.skip(reason="winloop not available"))
-    winloop = Mock()
-
 pytest_plugins = ["pytester"]
 
 asyncio_params = [
     pytest.param(("asyncio", {"debug": True}), id="asyncio"),
     pytest.param(
         (
             "asyncio",
-            {
-                "debug": True,
-                "loop_factory": uvloop.new_event_loop
-                if sys.platform != "win32"
-                else winloop.new_event_loop,
-            },
+            {"debug": True, "loop_factory": uvloop.new_event_loop},
         ),
-        marks=uvloop_marks if sys.platform != "win32" else winloop_marks,
-        id="asyncio+uvloop",
+        marks=uvloop_marks,
+        id=f"asyncio+{uvloop.__name__}",
     ),
 ]
 if sys.version_info >= (3, 12):
@@ -159,3 +154,16 @@ def no_other_refs() -> list[object]:
 
     def no_other_refs() -> list[object]:
         return [sys._getframe(1)]
+
+
+@pytest.fixture
+def event_loop_implementation_name() -> str | None:
+    try:
+        name = sniffio.current_async_library()
+    except sniffio.AsyncLibraryNotFoundError:
+        return None
+
+    if name == "asyncio":
+        return asyncio.get_running_loop().__module__
+
+    return name

tests/test_sockets.py

@@ -2197,28 +2197,19 @@ async def test_getaddrinfo() -> None:
 @pytest.mark.parametrize("sock_type", [socket.SOCK_STREAM, socket.SOCK_STREAM])
 async def test_getaddrinfo_ipv6addr(
     sock_type: Literal[socket.SocketKind.SOCK_STREAM],
+    event_loop_implementation_name: str | None,
 ) -> None:
     # IDNA trips up over raw IPv6 addresses
-    if platform.system() == "Windows":
-        import sniffio
-
-        if sniffio.current_async_library() == "asyncio":
-            import asyncio
-
-            if asyncio.get_running_loop().__module__ == "winloop":
-                proto = 6
-            else:
-                proto = 0
-        else:
-            proto = 0
+    if platform.system() == "Windows" and event_loop_implementation_name != "winloop":
+        expected_proto = 0
     else:
-        proto = 6
+        expected_proto = 6
 
     assert await getaddrinfo("::1", 0, type=sock_type) == [
         (
             socket.AF_INET6,
             socket.SOCK_STREAM,
-            proto,
+            expected_proto,
             "",
             ("::1", 0),
         )

tests/test_subprocesses.py

@@ -40,13 +40,11 @@
     ],
 )
 async def test_run_process(
-    shell: bool, command: str | list[str], anyio_backend_name: str
+    shell: bool, command: str | list[str], event_loop_implementation_name: str | None
 ) -> None:
-    if anyio_backend_name == "asyncio" and sys.platform == "win32":
-        import asyncio
+    if event_loop_implementation_name == "winloop":
+        pytest.skip(reason="winloop produces a non-zero exit with status 1")
 
-        if asyncio.get_event_loop().__module__ == "winloop":
-            pytest.skip(reason="winloop produces a non-zero exit with status 1")
     process = await run_process(command, input=b"abc")
     assert process.returncode == 0
     assert process.stdout.rstrip() == b"cba"
@@ -312,24 +310,21 @@ async def test_process_aexit_cancellation_closes_standard_streams(
 async def test_py39_arguments(
     argname: str,
     argvalue_factory: Callable[[], Any],
-    anyio_backend_name: str,
-    anyio_backend_options: dict[str, Any],
+    event_loop_implementation_name: str | None,
 ) -> None:
     try:
         await run_process(
             [sys.executable, "-c", "print('hello')"],
             **{argname: argvalue_factory()},
         )
     except ValueError as exc:
-        if (
-            "unexpected kwargs" in str(exc)
-            and anyio_backend_name == "asyncio"
-            and anyio_backend_options["loop_factory"]
-            and anyio_backend_options["loop_factory"].__module__
-            in ("uvloop", "winloop")
+        if "unexpected kwargs" in str(exc) and event_loop_implementation_name in (
+            "uvloop",
+            "winloop",
         ):
             pytest.skip(
-                f"the {argname!r} argument is not supported by uvloop or winloop yet"
+                f"the {argname!r} argument is not supported by "
+                f"{event_loop_implementation_name} yet"
             )
 
         raise