Put use_uvloop in conftest.py. Only run uvloop if present.

tests/conftest.py

@@ -0,0 +1,20 @@
+# Pytest looks here for fixtures
+
+import asyncio
+
+import pytest
+
+
+try:
+    import uvloop
+    loop_params = (False, True)
+except ImportError:
+    loop_params = (False, )
+
+
+# This runs all the tests one with plain asyncio, then again with uvloop
+@pytest.fixture(scope="session", autouse=True, params=loop_params)
+def use_uvloop(request):
+    if request.param:
+        import uvloop
+        asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())

tests/test_curio.py

@@ -154,7 +154,7 @@ async def test_tg_cm_all():
 
 @pytest.mark.asyncio
 async def test_tg_cm_any():
-    tasks = [await spawn(sleep, x/200) for x in range(5, 0, -1)]
+    tasks = [await spawn(sleep, x/200) for x in (0.1, 0.01, -1)]
     async with TaskGroup(tasks, wait=any) as t:
         pass
     assert all(task.done() for task in tasks)

tests/test_session.py

@@ -81,14 +81,6 @@ def caplog_count(caplog, message):
     return sum(message in record.message for record in caplog.records)
 
 
-# This runs all the tests one with plain asyncio, then again with uvloop
-@pytest.fixture(scope="session", autouse=True, params=(False, True))
-def use_uvloop(request):
-    if request.param:
-        import uvloop
-        asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())
-
-
 @pytest.fixture
 def server_port(unused_tcp_port, event_loop):
     coro = serve_rs(MyServerSession, 'localhost', unused_tcp_port, loop=event_loop)

tests/test_socks.py

@@ -24,13 +24,6 @@
 auth_methods = [None, SOCKSUserAuth('user', 'pass')]
 
 
-# This runs all the tests one with plain asyncio, then again with uvloop
-@pytest.fixture(scope="session", autouse=True, params=(False, True))
-def use_uvloop(request):
-    if request.param:
-        import uvloop
-        asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())
-
 @pytest.fixture(params=SOCKS4_addresses)
 def addr4(request):
     return request.param

tests/test_websocket.py

@@ -7,13 +7,6 @@
 
 from test_session import MyServerSession
 
-# This runs all the tests one with plain asyncio, then again with uvloop
-@pytest.fixture(scope="session", autouse=True, params=(False, True))
-def use_uvloop(request):
-    if request.param:
-        import uvloop
-        asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())
-
 
 @pytest.fixture
 def ws_server(unused_tcp_port, event_loop):