tests: switch from pytest.fixture to pytest_asyncio.fixture for async code

* Closes #5995
* Asynchronous fixtures should use `pytest_asyncio.fixture` rather than
  `pytest.fixture` when attempting to override the default event loop
  scope.
* In practice, this has to be done whenever the fixture scope is not
  "function" (which is the pytest_asyncio default).

Author: oliver-sanders

tests/conftest.py

@@ -21,6 +21,7 @@
 from typing import List, Optional, Tuple
 
 import pytest
+import pytest_asyncio
 
 from cylc.flow import LOG, flags
 from cylc.flow.cfgspec.glbl_cfg import glbl_cfg
@@ -40,7 +41,7 @@ def before_each():
     GraphNodeParser.get_inst().clear()
 
 
-@pytest.fixture(scope='module')
+@pytest_asyncio.fixture(scope='module')
 def mod_monkeypatch():
     """A module-scoped version of the monkeypatch fixture."""
     from _pytest.monkeypatch import MonkeyPatch
@@ -160,7 +161,7 @@ def _log_scan(log, items):
     return _log_scan
 
 
-@pytest.fixture(scope='session')
+@pytest_asyncio.fixture(scope='session')
 def port_range():
     return glbl_cfg().get(['scheduler', 'run hosts', 'ports'])
 

tests/integration/conftest.py

@@ -30,6 +30,7 @@
 )
 
 import pytest
+import pytest_asyncio
 
 from cylc.flow.config import WorkflowConfig
 from cylc.flow.id import Tokens
@@ -112,15 +113,15 @@ def _pytest_passed(request: pytest.FixtureRequest) -> bool:
     ))
 
 
-@pytest.fixture(scope='session')
+@pytest_asyncio.fixture(scope='session')
 def run_dir():
     """The cylc run directory for this host."""
     path = Path(get_cylc_run_dir())
     path.mkdir(exist_ok=True)
     yield path
 
 
-@pytest.fixture(scope='session')
+@pytest_asyncio.fixture(scope='session')
 def ses_test_dir(request, run_dir):
     """The root run dir for test flows in this test session."""
     timestamp = get_current_time_string(use_basic_format=True)
@@ -131,7 +132,7 @@ def ses_test_dir(request, run_dir):
     _rm_if_empty(path)
 
 
-@pytest.fixture(scope='module')
+@pytest_asyncio.fixture(scope='module')
 def mod_test_dir(request, ses_test_dir):
     """The root run dir for test flows in this test module."""
     path = Path(
@@ -163,7 +164,7 @@ def test_dir(request, mod_test_dir):
         _rm_if_empty(path)
 
 
-@pytest.fixture(scope='module')
+@pytest_asyncio.fixture(scope='module')
 def mod_flow(run_dir, mod_test_dir):
     """A function for creating module-level flows."""
     yield partial(_make_flow, run_dir, mod_test_dir)
@@ -175,7 +176,7 @@ def flow(run_dir, test_dir):
     yield partial(_make_flow, run_dir, test_dir)
 
 
-@pytest.fixture(scope='module')
+@pytest_asyncio.fixture(scope='module')
 def mod_scheduler():
     """Return a Scheduler object for a flow.
 
@@ -197,7 +198,7 @@ def scheduler():
         yield _scheduler
 
 
-@pytest.fixture(scope='module')
+@pytest_asyncio.fixture(scope='module')
 def mod_start():
     """Start a scheduler but don't set it running (module scope)."""
     return partial(_start_flow, None)
@@ -209,7 +210,7 @@ def start(caplog: pytest.LogCaptureFixture):
     return partial(_start_flow, caplog)
 
 
-@pytest.fixture(scope='module')
+@pytest_asyncio.fixture(scope='module')
 def mod_run():
     """Start a scheduler and set it running (module scope)."""
     return partial(_run_flow, None)
@@ -235,7 +236,7 @@ def one_conf():
     }
 
 
-@pytest.fixture(scope='module')
+@pytest_asyncio.fixture(scope='module')
 def mod_one_conf():
     return {
         'scheduler': {
@@ -257,38 +258,13 @@ def one(one_conf, flow, scheduler):
     return schd
 
 
-@pytest.fixture(scope='module')
+@pytest_asyncio.fixture(scope='module')
 def mod_one(mod_one_conf, mod_flow, mod_scheduler):
     id_ = mod_flow(mod_one_conf)
     schd = mod_scheduler(id_)
     return schd
 
 
-@pytest.fixture(scope='module')
-def event_loop():
-    """This fixture defines the event loop used for each test.
-
-    The default scoping for this fixture is "function" which means that all
-    async fixtures must have "function" scoping.
-
-    Defining `event_loop` as a module scoped fixture opens the door to
-    module scoped fixtures but means all tests in a module will run in the same
-    event loop. This is fine, it's actually an efficiency win but also
-    something to be aware of.
-
-    See: https://github.com/pytest-dev/pytest-asyncio/issues/171
-
-    """
-    loop = asyncio.get_event_loop_policy().new_event_loop()
-    yield loop
-    # gracefully exit async generators
-    loop.run_until_complete(loop.shutdown_asyncgens())
-    # cancel any tasks still running in this event loop
-    for task in asyncio.all_tasks(loop):
-        task.cancel()
-    loop.close()
-
-
 @pytest.fixture
 def db_select():
     """Select columns from workflow database.
@@ -382,7 +358,7 @@ def _validate(id_: Union[str, Path], **kwargs) -> WorkflowConfig:
     return _validate
 
 
-@pytest.fixture(scope='module')
+@pytest_asyncio.fixture(scope='module')
 def mod_validate(run_dir):
     """Provides a function for validating workflow configurations.
 
@@ -465,7 +441,7 @@ def run_job_cmd(
     return _disable_polling
 
 
-@pytest.fixture(scope='module')
+@pytest_asyncio.fixture(scope='module')
 def mod_workflow_source(mod_flow, tmp_path_factory):
     """Create a workflow source directory.
 
@@ -686,7 +662,7 @@ def complete():
     return _complete
 
 
-@pytest.fixture(scope='module')
+@pytest_asyncio.fixture(scope='module')
 def mod_complete():
     return _complete
 

tests/integration/network/test_client.py

@@ -15,16 +15,19 @@
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
 """Test cylc.flow.client.WorkflowRuntimeClient."""
+
 import json
 from unittest.mock import Mock
+
 import pytest
+import pytest_asyncio
 
 from cylc.flow.exceptions import ClientError
 from cylc.flow.network.client import WorkflowRuntimeClient
 from cylc.flow.network.server import PB_METHOD_MAP
 
 
-@pytest.fixture(scope='module')
+@pytest_asyncio.fixture(scope='module')
 async def harness(mod_flow, mod_scheduler, mod_run, mod_one_conf):
     id_ = mod_flow(mod_one_conf)
     schd = mod_scheduler(id_)

tests/integration/network/test_graphql.py

@@ -16,9 +16,11 @@
 
 """Test the top-level (root) GraphQL queries."""
 
-import pytest
 from typing import TYPE_CHECKING
 
+import pytest
+import pytest_asyncio
+
 from cylc.flow.id import Tokens
 from cylc.flow.network.client import WorkflowRuntimeClient
 
@@ -68,7 +70,7 @@ def job_db_row():
     ]
 
 
-@pytest.fixture(scope='module')
+@pytest_asyncio.fixture(scope='module')
 async def harness(mod_flow, mod_scheduler, mod_run):
     flow_def = {
         'scheduler': {

tests/integration/network/test_resolvers.py

@@ -19,6 +19,7 @@
 from unittest.mock import Mock
 
 import pytest
+import pytest_asyncio
 
 from cylc.flow.data_store_mgr import EDGES, TASK_PROXIES
 from cylc.flow.id import Tokens
@@ -50,7 +51,7 @@ def node_args():
     }
 
 
-@pytest.fixture(scope='module')
+@pytest_asyncio.fixture(scope='module')
 async def mock_flow(
     mod_flow: Callable[..., str],
     mod_scheduler: Callable[..., Scheduler],

tests/integration/network/test_scan.py

@@ -23,6 +23,7 @@
 from typing import List
 
 import pytest
+import pytest_asyncio
 
 from cylc.flow.network.scan import (
     filter_name,
@@ -81,7 +82,7 @@ def make_src(name):
         make_src(name)
 
 
-@pytest.fixture(scope='session')
+@pytest_asyncio.fixture(scope='session')
 def sample_run_dir():
     tmp_path = Path(TemporaryDirectory().name)
     tmp_path.mkdir()
@@ -111,7 +112,7 @@ def badly_messed_up_cylc_run_dir(
     return tmp_path
 
 
-@pytest.fixture(scope='session')
+@pytest_asyncio.fixture(scope='session')
 def run_dir_with_symlinks():
     tmp_path = Path(TemporaryDirectory().name)
     tmp_path.mkdir()
@@ -133,7 +134,7 @@ def run_dir_with_symlinks():
     rmtree(tmp_path)
 
 
-@pytest.fixture(scope='session')
+@pytest_asyncio.fixture(scope='session')
 def run_dir_with_nasty_symlinks():
     tmp_path = Path(TemporaryDirectory().name)
     tmp_path.mkdir()
@@ -148,7 +149,7 @@ def run_dir_with_nasty_symlinks():
     rmtree(tmp_path)
 
 
-@pytest.fixture(scope='session')
+@pytest_asyncio.fixture(scope='session')
 def nested_dir():
     tmp_path = Path(TemporaryDirectory().name)
     tmp_path.mkdir()

tests/integration/network/test_server.py

@@ -19,6 +19,7 @@
 from typing import Callable
 
 import pytest
+import pytest_asyncio
 
 from cylc.flow import __version__ as CYLC_VERSION
 from cylc.flow.network.server import PB_METHOD_MAP
@@ -31,7 +32,7 @@
     from async_timeout import timeout
 
 
-@pytest.fixture(scope='module')
+@pytest_asyncio.fixture(scope='module')
 async def myflow(mod_flow, mod_scheduler, mod_run, mod_one_conf):
     id_ = mod_flow(mod_one_conf)
     schd = mod_scheduler(id_)

tests/integration/network/test_zmq.py

@@ -15,6 +15,7 @@
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
 import pytest
+import pytest_asyncio
 import zmq
 
 from cylc.flow.exceptions import CylcError
@@ -23,7 +24,7 @@
 from .key_setup import setup_keys
 
 
-@pytest.fixture(scope='module')
+@pytest_asyncio.fixture(scope='module')
 def myflow(mod_flow, mod_one_conf):
     return mod_flow(mod_one_conf)
 

tests/integration/run_modes/test_simulation.py

@@ -18,8 +18,10 @@
 
 import logging
 from pathlib import Path
+
 import pytest
 from pytest import param
+import pytest_asyncio
 
 from cylc.flow import commands
 from cylc.flow.cycling.iso8601 import ISO8601Point
@@ -88,7 +90,7 @@ def _run_simjob(schd, point, task):
     return _run_simjob
 
 
-@pytest.fixture(scope='module')
+@pytest_asyncio.fixture(scope='module')
 async def sim_time_check_setup(
     mod_flow, mod_scheduler, mod_start, mod_one_conf,
 ):

tests/integration/scripts/test_list.py

@@ -17,6 +17,7 @@
 """Test the "cylc list" command."""
 
 import pytest
+import pytest_asyncio
 
 from cylc.flow.exceptions import InputError
 from cylc.flow.option_parsers import Options
@@ -29,7 +30,7 @@
 ListOptions = Options(get_option_parser())
 
 
-@pytest.fixture(scope='module')
+@pytest_asyncio.fixture(scope='module')
 async def cylc_list(mod_flow, mod_scheduler, mod_start):
     id_ = mod_flow(
         {