tests - fix pytest-asyncio detected an unclosed event loop when tearing down the event_loop

 - using a different policy
 - fixture event_loop in conftest.py - fixes pytest-dev/pytest-asyncio#257

Author: commonism

tests/apiv1_test.py

@@ -21,15 +21,9 @@ def config(unused_tcp_port_factory):
     return c
 
 
-@pytest.fixture(scope="session")
-def event_loop(request):
-    loop = asyncio.get_event_loop_policy().new_event_loop()
-    yield loop
-    loop.close()
-
-
 @pytest_asyncio.fixture(scope="session")
 async def server(event_loop, config):
+    policy = asyncio.get_event_loop_policy()
     uvloop.install()
     try:
         sd = asyncio.Event()
@@ -38,6 +32,7 @@ async def server(event_loop, config):
     finally:
         sd.set()
         await task
+    asyncio.set_event_loop_policy(policy)
 
 
 @pytest_asyncio.fixture(scope="session")

tests/apiv2_test.py

@@ -35,15 +35,9 @@ def config(unused_tcp_port_factory):
     return c
 
 
-@pytest.fixture(scope="session")
-def event_loop(request):
-    loop = asyncio.get_event_loop_policy().new_event_loop()
-    yield loop
-    loop.close()
-
-
 @pytest_asyncio.fixture(scope="session")
 async def server(event_loop, config):
+    policy = asyncio.get_event_loop_policy()
     uvloop.install()
     try:
         sd = asyncio.Event()
@@ -52,6 +46,7 @@ async def server(event_loop, config):
     finally:
         sd.set()
         await task
+    asyncio.set_event_loop_policy(policy)
 
 
 @pytest.fixture(scope="session", params=[2])

tests/conftest.py

@@ -1,3 +1,4 @@
+import asyncio
 import os
 import dataclasses
 import sys
@@ -16,6 +17,13 @@
 URLBASE = "/"
 
 
+@pytest.fixture(scope="session")
+def event_loop(request):
+    loop = asyncio.get_event_loop_policy().new_event_loop()
+    yield loop
+    loop.close()
+
+
 @pytest.fixture(autouse=True)
 def skip_env(request):
     if request.node.get_closest_marker("skip_env"):

tests/forms_test.py

@@ -269,24 +269,19 @@ def config(unused_tcp_port_factory):
     return c
 
 
-@pytest.fixture(scope="session")
-def event_loop(request):
-    loop = asyncio.get_event_loop_policy().new_event_loop()
-    yield loop
-    loop.close()
-
-
 @pytest_asyncio.fixture(scope="session")
 async def server(event_loop, config, app):
-    uvloop.install()
+    policy = asyncio.get_event_loop_policy()
     try:
         sd = asyncio.Event()
         asgi = WsgiToAsgi(app)
         task = event_loop.create_task(serve(asgi, config, shutdown_trigger=sd.wait))
         yield config
     finally:
         sd.set()
+        del asgi
         await task
+    asyncio.set_event_loop_policy(policy)
 
 
 @pytest.fixture(scope="session", params=["application/x-www-form-urlencoded", "multipart/form-data"])

tests/linode_test.py

@@ -1,5 +1,3 @@
-import asyncio
-
 import aiopenapi3.plugin
 from aiopenapi3 import OpenAPI
 import pytest
@@ -70,13 +68,6 @@ def parsed(self, ctx):
         return ctx
 
 
-@pytest_asyncio.fixture(scope="session")
-def event_loop(request):
-    loop = asyncio.get_event_loop_policy().new_event_loop()
-    yield loop
-    loop.close()
-
-
 @pytest_asyncio.fixture(scope="session")
 async def api():
     from aiopenapi3.loader import NullLoader, YAMLCompatibilityLoader