1) A new test case that fails with 0.12.0, and pass with this commit.

   test_async_fixtures_with_finalizer_scope.py:
2) Main problem is due to side effects of asyncio.get_event_loop(). See:
   https://github.com/python/cpython/blob/3.8/Lib/asyncio/events.py#L636
   This method could either return an existing loop or create a new one.

   This commit replaces all asyncio.get_event_loop() with previous pytest-asyncio code (0.10.0), so plugin uses the loop provided by event_loop fixture (instead of calling asyncio.get_event_loop())

   Except following block, that has not been modified in this commit (as it behaves similar to 0.10.0)
   https://github.com/pytest-dev/pytest-asyncio/blob/v0.12.0/pytest_asyncio/plugin.py#L54-L66

Changes are for using always the new loop provided by event_loop fixture.
   Instead of calling get_event_loop() that:
   - either returns global recorded loop (with set_event_loop()) in case there is one,
   - or otherwise creates and record a new one

pytest_asyncio/plugin.py

@@ -69,13 +69,20 @@ def pytest_fixture_setup(fixturedef, request):
         # This is an async generator function. Wrap it accordingly.
         generator = fixturedef.func
 
+        strip_event_loop = False
+        if 'event_loop' not in fixturedef.argnames:
+            fixturedef.argnames += ('event_loop', )
+            strip_event_loop = True
         strip_request = False
         if 'request' not in fixturedef.argnames:
             fixturedef.argnames += ('request', )
             strip_request = True
 
         def wrapper(*args, **kwargs):
+            loop = kwargs['event_loop']
             request = kwargs['request']
+            if strip_event_loop:
+                del kwargs['event_loop']
             if strip_request:
                 del kwargs['request']
 
@@ -96,21 +103,30 @@ async def async_finalizer():
                         msg = "Async generator fixture didn't stop."
                         msg += "Yield only once."
                         raise ValueError(msg)
-                asyncio.get_event_loop().run_until_complete(async_finalizer())
+                loop.run_until_complete(async_finalizer())
 
             request.addfinalizer(finalizer)
-            return asyncio.get_event_loop().run_until_complete(setup())
+            return loop.run_until_complete(setup())
 
         fixturedef.func = wrapper
     elif inspect.iscoroutinefunction(fixturedef.func):
         coro = fixturedef.func
 
+        strip_event_loop = False
+        if 'event_loop' not in fixturedef.argnames:
+            fixturedef.argnames += ('event_loop', )
+            strip_event_loop = True
+
         def wrapper(*args, **kwargs):
+            loop = kwargs['event_loop']
+            if strip_event_loop:
+                del kwargs['event_loop']
+
             async def setup():
                 res = await coro(*args, **kwargs)
                 return res
 
-            return asyncio.get_event_loop().run_until_complete(setup())
+            return loop.run_until_complete(setup())
 
         fixturedef.func = wrapper
     yield
@@ -144,15 +160,9 @@ def wrap_in_sync(func, _loop):
     def inner(**kwargs):
         coro = func(**kwargs)
         if coro is not None:
+            task = asyncio.ensure_future(coro, loop=_loop)
             try:
-                loop = asyncio.get_event_loop()
-            except RuntimeError as exc:
-                if 'no current event loop' not in str(exc):
-                    raise
-                loop = _loop
-            task = asyncio.ensure_future(coro, loop=loop)
-            try:
-                loop.run_until_complete(task)
+                _loop.run_until_complete(task)
             except BaseException:
                 # run_until_complete doesn't get the result from exceptions
                 # that are not subclasses of `Exception`. Consume all

tests/async_fixtures/test_async_fixtures_with_finalizer_scope.py

@@ -0,0 +1,38 @@
+import asyncio
+import contextlib
+import functools
+import pytest
+
+
+@pytest.mark.asyncio
+async def test_module_scope(port):
+    await asyncio.sleep(0.01)
+    assert port
+
+@pytest.fixture(scope="module")
+def event_loop():
+    """Change event_loop fixture to module level."""
+    policy = asyncio.get_event_loop_policy()
+    loop = policy.new_event_loop()
+    yield loop
+    loop.close()
+
+
+@pytest.fixture(scope="module")
+async def port(request, event_loop):
+    def port_finalizer(finalizer):
+        async def port_afinalizer():
+            await finalizer(None, None, None)
+        event_loop.run_until_complete(port_afinalizer())
+
+    context_manager = port_map()
+    port = await context_manager.__aenter__()
+    request.addfinalizer(functools.partial(port_finalizer, context_manager.__aexit__))
+    return True
+
+
+@contextlib.asynccontextmanager
+async def port_map():
+    worker = asyncio.create_task(asyncio.sleep(0.2))
+    yield
+    await worker