fileinstall: remove asyncio logic

cylc/rose/entry_points.py

@@ -76,7 +76,10 @@
 
 def rose_stem():
     """Implements the "rose stem" command."""
-    from cylc.rose.stem import get_rose_stem_opts
+    import asyncio
+    from cylc.rose.stem import get_rose_stem_opts, rose_stem
 
     parser, opts = get_rose_stem_opts()
-    rose_stem(parser, opts)
+    asyncio.run(
+        rose_stem(parser, opts)
+    )

cylc/rose/fileinstall.py

@@ -40,14 +40,12 @@ def rose_fileinstall(
 
     if any(i.startswith('file') for i in config_tree.node.value):
         try:
-            startpoint = os.getcwd()
+            # NOTE: Cylc will chdir back for us afterwards
             os.chdir(rundir)
         except FileNotFoundError as exc:
             raise exc
         else:
             # Carry out imports.
-            import asyncio
-
             from metomi.rose.config_processor import ConfigProcessorsManager
             from metomi.rose.fs_util import FileSystemUtil
             from metomi.rose.popen import RosePopener
@@ -64,19 +62,8 @@ def rose_fileinstall(
             fs_util = FileSystemUtil(event_handler)
             popen = RosePopener(event_handler)
 
-            # Get an Asyncio loop if one doesn't exist:
-            #   Rose may need an event loop to invoke async interfaces,
-            #   doing this here incase we want to go async in cylc-rose.
-            # See https://github.com/cylc/cylc-rose/pull/130/files
-            try:
-                asyncio.get_event_loop()
-            except RuntimeError:
-                asyncio.set_event_loop(asyncio.new_event_loop())
-
             # Process fileinstall.
             config_pm = ConfigProcessorsManager(event_handler, popen, fs_util)
             config_pm(config_tree, "file")
-        finally:
-            os.chdir(startpoint)
 
     return config_tree.node

cylc/rose/stem.py

@@ -609,13 +609,13 @@ def get_rose_stem_opts():
     return parser, opts
 
 
-def rose_stem(parser, opts):
+async def rose_stem(parser, opts):
     try:
         # modify the CLI options to add whatever rose stem would like to add
         opts = StemRunner(opts).process()
 
         # call cylc install
-        cylc_install(opts, opts.workflow_conf_dir)
+        await cylc_install(opts, opts.workflow_conf_dir)
 
     except CylcError as exc:
         if opts.verbosity > 1:

setup.cfg

@@ -68,6 +68,8 @@ include = cylc*
 tests =
     coverage>=5.0.0
     pytest
+    # https://github.com/pytest-dev/pytest-asyncio/issues/705
+    pytest-asyncio==0.21.*
     pytest-cov
     pytest-xdist>=2
 lint =

tests/conftest.py

@@ -14,19 +14,52 @@
 # You should have received a copy of the GNU General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
+import asyncio
+from pathlib import Path
+from shutil import rmtree
 from types import SimpleNamespace
 from uuid import uuid4
 
+import pytest
+
 from cylc.flow import __version__ as CYLC_VERSION
 from cylc.flow.option_parsers import Options
-from cylc.flow.pathutil import get_workflow_run_dir
+from cylc.flow.pathutil import get_cylc_run_dir
 from cylc.flow.scripts.install import get_option_parser as install_gop
 from cylc.flow.scripts.install import install_cli as cylc_install
 from cylc.flow.scripts.reinstall import get_option_parser as reinstall_gop
 from cylc.flow.scripts.reinstall import reinstall_cli as cylc_reinstall
-from cylc.flow.scripts.validate import _main as cylc_validate
+from cylc.flow.scripts.validate import run as cylc_validate
 from cylc.flow.scripts.validate import get_option_parser as validate_gop
-import pytest
+from cylc.flow.wallclock import get_current_time_string
+
+
+CYLC_RUN_DIR = Path(get_cylc_run_dir())
+
+
+@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()
@@ -98,15 +131,34 @@ def pytest_runtest_makereport(item, call):
     item.module._module_outcomes = _module_outcomes
 
 
+def _rm_if_empty(path):
+    """Convenience wrapper for removing empty directories."""
+    try:
+        path.rmdir()
+    except OSError:
+        return False
+    return True
+
+
+def _pytest_passed(request: pytest.FixtureRequest) -> bool:
+    """Returns True if the test(s) a fixture was used in passed."""
+    if hasattr(request.node, '_function_outcome'):
+        return request.node._function_outcome.outcome in {'passed', 'skipped'}
+    return all((
+        report.outcome in {'passed', 'skipped'}
+        for report in request.node.obj._module_outcomes.values()
+    ))
+
+
 def _cylc_validate_cli(capsys, caplog):
     """Access the validate CLI"""
-    def _inner(srcpath, args=None):
+    async def _inner(srcpath, args=None):
         parser = validate_gop()
         options = Options(parser, args)()
         output = SimpleNamespace()
 
         try:
-            cylc_validate(parser, options, str(srcpath))
+            await cylc_validate(parser, options, str(srcpath))
             output.ret = 0
             output.exc = ''
         except Exception as exc:
@@ -120,53 +172,77 @@ def _inner(srcpath, args=None):
     return _inner
 
 
-def _cylc_install_cli(capsys, caplog, workflow_name):
+def _cylc_install_cli(capsys, caplog, test_dir):
     """Access the install CLI"""
-    def _inner(srcpath, args=None):
+    async def _inner(srcpath, workflow_name=None, opts=None):
         """Install a workflow.
 
         Args:
             srcpath:
-            args: Dictionary of arguments.
+                The workflow to install
+            workflow_name:
+                The workflow ID prefix to install this workflow as.
+
+                If you leave this blank, it will use the module/function's
+                test directory as appropriate.
+            opts:
+                Dictionary of arguments for cylc install.
+
         """
-        options = Options(install_gop(), args)()
+        nonlocal capsys, caplog, test_dir
+        if not workflow_name:
+            workflow_name = str(
+                (test_dir / str(uuid4())[:4]).relative_to(CYLC_RUN_DIR)
+            )
+        options = Options(
+            install_gop(), opts or {}
+        )(workflow_name=workflow_name)
         output = SimpleNamespace()
         if not options.workflow_name:
             options.workflow_name = workflow_name
-        if not args or not args.get('no_run_name', ''):
+        if not opts or not opts.get('no_run_name', ''):
             options.no_run_name = True
 
         try:
-            output.name, output.id = cylc_install(options, str(srcpath))
+            output.name, output.id = await cylc_install(options, str(srcpath))
             output.ret = 0
             output.exc = ''
         except Exception as exc:
             output.ret = 1
             output.exc = exc
         output.logging = '\n'.join([i.message for i in caplog.records])
         output.out, output.err = capsys.readouterr()
-        output.run_dir = get_workflow_run_dir(output.id)
         return output
     return _inner
 
 
-def _cylc_reinstall_cli(capsys, caplog):
+def _cylc_reinstall_cli(capsys, caplog, test_dir):
     """Access the reinstall CLI"""
-    def _inner(workflow_id, opts=None):
+    async def _inner(workflow_id=None, opts=None):
         """Install a workflow.
 
         Args:
-            srcpath:
-            args: Dictionary of arguments.
+            workflow_id:
+                The workflow ID to reinstall.
+
+                If you leave this blank, it will use the module/function's
+                test directory as appropriate.
+            args:
+                Dictionary of arguments for cylc reinstall.
+
         """
-        options = Options(reinstall_gop(), opts)()
+        nonlocal capsys, caplog, test_dir
+        if not workflow_id:
+            workflow_id = str(test_dir.relative_to(CYLC_RUN_DIR))
+        options = Options(reinstall_gop(), opts or {})()
         output = SimpleNamespace()
 
         try:
-            cylc_reinstall(options, workflow_id)
+            await cylc_reinstall(options, workflow_id)
             output.ret = 0
             output.exc = ''
         except Exception as exc:
+            # raise
             output.ret = 1
             output.exc = exc
         output.logging = '\n'.join([i.message for i in caplog.records])
@@ -176,24 +252,23 @@ def _inner(workflow_id, opts=None):
 
 
 @pytest.fixture
-def cylc_install_cli(capsys, caplog, workflow_name):
-    return _cylc_install_cli(capsys, caplog, workflow_name)
+def cylc_install_cli(capsys, caplog, test_dir):
+    return _cylc_install_cli(capsys, caplog, test_dir)
 
 
 @pytest.fixture(scope='module')
-def mod_cylc_install_cli(mod_capsys, mod_caplog, mod_workflow_name):
-    return _cylc_install_cli(
-        mod_capsys, mod_caplog, mod_workflow_name)
+def mod_cylc_install_cli(mod_capsys, mod_caplog):
+    return _cylc_install_cli(mod_capsys, mod_caplog, mod_test_dir)
 
 
 @pytest.fixture
-def cylc_reinstall_cli(capsys, caplog):
-    return _cylc_reinstall_cli(capsys, caplog)
+def cylc_reinstall_cli(capsys, caplog, test_dir):
+    return _cylc_reinstall_cli(capsys, caplog, test_dir)
 
 
 @pytest.fixture(scope='module')
-def mod_cylc_reinstall_cli(mod_capsys, mod_caplog):
-    return _cylc_reinstall_cli(mod_capsys, mod_caplog)
+def mod_cylc_reinstall_cli(mod_capsys, mod_caplog, mod_test_dir):
+    return _cylc_reinstall_cli(mod_capsys, mod_caplog, mod_test_dir)
 
 
 @pytest.fixture
@@ -204,3 +279,49 @@ def cylc_validate_cli(capsys, caplog):
 @pytest.fixture(scope='module')
 def mod_cylc_validate_cli(mod_capsys, mod_caplog):
     return _cylc_validate_cli(mod_capsys, mod_caplog)
+
+
+@pytest.fixture(scope='session')
+def run_dir():
+    """The cylc run directory for this host."""
+    CYLC_RUN_DIR.mkdir(exist_ok=True)
+    yield CYLC_RUN_DIR
+
+
+@pytest.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)
+    uuid = f'cylc-rose-test-{timestamp}-{str(uuid4())[:4]}'
+    path = Path(run_dir, uuid)
+    path.mkdir(exist_ok=True)
+    yield path
+    _rm_if_empty(path)
+
+
+@pytest.fixture(scope='module')
+def mod_test_dir(request, ses_test_dir):
+    """The root run dir for test flows in this test module."""
+    path = Path(ses_test_dir, request.module.__name__)
+    path.mkdir(exist_ok=True)
+    yield path
+    if _pytest_passed(request):
+        # test passed -> remove all files
+        rmtree(path, ignore_errors=False)
+    else:
+        # test failed -> remove the test dir if empty
+        _rm_if_empty(path)
+
+
+@pytest.fixture
+def test_dir(request, mod_test_dir):
+    """The root run dir for test flows in this test function."""
+    path = Path(mod_test_dir, request.function.__name__)
+    path.mkdir(parents=True, exist_ok=True)
+    yield path
+    if _pytest_passed(request):
+        # test passed -> remove all files
+        rmtree(path, ignore_errors=False)
+    else:
+        # test failed -> remove the test dir if empty
+        _rm_if_empty(path)

tests/functional/test_ROSE_ORIG_HOST.py

@@ -102,7 +102,7 @@ def fixture_provide_flow(tmp_path_factory, request):
 
 
 @pytest.fixture(scope='module')
-def fixture_install_flow(
+async def fixture_install_flow(
     fixture_provide_flow, monkeymodule, mod_cylc_install_cli
 ):
     """Run ``cylc install``.
@@ -113,9 +113,9 @@ def fixture_install_flow(
     If a test fails then using ``pytest --pdb`` and
     ``fixture_install_flow['result'].stderr`` may help with debugging.
     """
-    result = mod_cylc_install_cli(
+    result = await mod_cylc_install_cli(
         fixture_provide_flow['srcpath'],
-        {'workflow_name': fixture_provide_flow['test_flow_name']}
+        fixture_provide_flow['test_flow_name'],
     )
     install_conf_path = (
         fixture_provide_flow['flowpath'] /
@@ -130,20 +130,26 @@ def fixture_install_flow(
     }
 
 
-def test_cylc_validate_srcdir(fixture_install_flow, mod_cylc_validate_cli):
+async def test_cylc_validate_srcdir(
+    fixture_install_flow,
+    mod_cylc_validate_cli,
+):
     """Sanity check that workflow validates:
     """
     srcpath = fixture_install_flow['srcpath']
-    result = mod_cylc_validate_cli(srcpath)
+    result = await mod_cylc_validate_cli(srcpath)
     search = re.findall(r'ROSE_ORIG_HOST \(.*\) is: (.*)', result.logging)
     assert search == [HOST, HOST]
 
 
-def test_cylc_validate_rundir(fixture_install_flow, mod_cylc_validate_cli):
+async def test_cylc_validate_rundir(
+    fixture_install_flow,
+    mod_cylc_validate_cli,
+):
     """Sanity check that workflow validates:
     """
     flowpath = fixture_install_flow['flowpath']
-    result = mod_cylc_validate_cli(flowpath)
+    result = await mod_cylc_validate_cli(flowpath)
     assert 'ROSE_ORIG_HOST (env) is:' in result.logging