Add unit tests for workflow-state.

cylc/flow/scripts/workflow_state.py

@@ -74,6 +74,7 @@
 
 if TYPE_CHECKING:
     from optparse import Values
+    from typing import Tuple, Optional
 
 
 class WorkflowPoller(Poller):
@@ -194,6 +195,19 @@ def get_option_parser() -> COP:
     return parser
 
 
+def get_workflow(w_id: str, alt_run_dir: Optional[str] = None) -> Tuple[str, str]:
+    """Infer run number for a workflow ID, for alternate run dir if nec."""
+    if alt_run_dir:
+        run_dir = alt_run_dir = expand_path(alt_run_dir)
+    else:
+        run_dir = get_cylc_run_dir()
+        alt_run_dir = None
+    workflow_id, *_ = parse_id(
+        w_id, constraint='workflows', alt_run_dir=alt_run_dir
+    )
+    return workflow_id, run_dir
+
+
 @cli_function(get_option_parser, remove_opts=["--db"])
 def main(parser: COP, options: 'Values', workflow_id: str) -> None:
 
@@ -227,18 +241,8 @@ def main(parser: COP, options: 'Values', workflow_id: str) -> None:
             options.status not in CylcWorkflowDBChecker.STATE_ALIASES):
         raise InputError(f"invalid status '{options.status}'")
 
-    # this only runs locally
-    if options.alt_run_dir:
-        run_dir = alt_run_dir = expand_path(options.alt_run_dir)
-    else:
-        run_dir = get_cylc_run_dir()
-        alt_run_dir = None
-
-    workflow_id, *_ = parse_id(
-        workflow_id,
-        constraint='workflows',
-        alt_run_dir=alt_run_dir
-    )
+    # Infer workflow run number if necessary.
+    workflow_id, run_dir = get_workflow(workflow_id, options.alt_run_dir)
 
     pollargs = {
         'workflow_id': workflow_id,

tests/unit/scripts/test_workflow_state.py

@@ -0,0 +1,49 @@
+# THIS FILE IS PART OF THE CYLC WORKFLOW ENGINE.
+# Copyright (C) NIWA & British Crown (Met Office) & Contributors.
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+# Test aspects of the `cylc workflow-state` command.
+
+import pytest
+from shutil import copytree, rmtree
+
+from cylc.flow.exceptions import InputError
+from cylc.flow.pathutil import get_cylc_run_dir
+from cylc.flow.scripts.workflow_state import get_workflow
+
+
+def test_get_workflow(tmp_run_dir):
+    """It should infer the run name for auto-numbered installations."""
+    # it doesn't do anything for a named run
+    tmp_run_dir('foo/run1', installed=True)
+    cylc_run_dir = get_cylc_run_dir()
+
+    assert get_workflow('foo') == ('foo/run1', cylc_run_dir)
+
+    # Now test we can see workflows in alternate cylc-run directories
+    # e.g. for `cylc workflow-state` or xtriggers targetting another user.
+    alt_cylc_run_dir = cylc_run_dir + "_alt"
+
+    # copy the cylc-run dir to alt location and delete the original.
+    copytree(cylc_run_dir, alt_cylc_run_dir, symlinks=True)
+    rmtree(cylc_run_dir)
+
+    # It can no longer parse IDs in the original cylc-run location.
+    with pytest.raises(InputError):
+        get_workflow('foo')
+
+    # But it can if we specify the alternate location.
+    assert get_workflow('foo', alt_cylc_run_dir) == (
+        'foo/run1', alt_cylc_run_dir)

tests/unit/test_id_cli.py

@@ -18,6 +18,7 @@
 import os
 from pathlib import Path
 import pytest
+from shutil import copytree, rmtree
 
 from cylc.flow import CYLC_LOG
 from cylc.flow.async_util import pipe
@@ -248,6 +249,32 @@ async def test_parse_ids_infer_run_name(tmp_run_dir):
     )
     assert list(workflows) == ['bar']
 
+    # Now test we can see workflows in alternate cylc-run directories
+    # e.g. for `cylc workflow-state` or xtriggers targetting another user.
+    cylc_run_dir = get_cylc_run_dir()
+    alt_cylc_run_dir = cylc_run_dir + "_alt"
+
+    # copy the cylc-run dir to alt location and delete the original.
+    copytree(cylc_run_dir, alt_cylc_run_dir, symlinks=True)
+    rmtree(cylc_run_dir)
+
+    # It can no longer parse IDs in the original cylc-run location.
+    with pytest.raises(InputError):
+        workflows, *_ = await parse_ids_async(
+            'bar//',
+            constraint='workflows',
+            infer_latest_runs=True,
+        )
+
+    # But it can if we specify the alternate location.
+    workflows, *_ = await parse_ids_async(
+        'bar//',
+        constraint='workflows',
+        infer_latest_runs=True,
+        alt_run_dir=alt_cylc_run_dir
+    )
+    assert list(workflows) == ['bar/run2']
+
 
 @pytest.fixture
 def patch_expand_workflow_tokens(monkeypatch):

tests/unit/xtriggers/test_workflow_state.py

@@ -15,15 +15,18 @@
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
 from pathlib import Path
+import pytest
 import sqlite3
 from typing import Callable
 from unittest.mock import Mock
+from shutil import copytree, rmtree
 
+from cylc.flow.exceptions import InputError
+from cylc.flow.pathutil import get_cylc_run_dir
 from cylc.flow.workflow_files import WorkflowFiles
 from cylc.flow.xtriggers.workflow_state import workflow_state
 from ..conftest import MonkeyMock
 
-
 def test_inferred_run(tmp_run_dir: Callable, monkeymock: MonkeyMock):
     """Test that the workflow_state xtrigger infers the run number"""
     id_ = 'isildur'
@@ -42,6 +45,24 @@ def test_inferred_run(tmp_run_dir: Callable, monkeymock: MonkeyMock):
     assert results['workflow'] == expected_workflow_id
 
 
+    # Now test we can see workflows in alternate cylc-run directories
+    # e.g. for `cylc workflow-state` or xtriggers targetting another user.
+    alt_cylc_run_dir = cylc_run_dir + "_alt"
+
+    # copy the cylc-run dir to alt location and delete the original.
+    copytree(cylc_run_dir, alt_cylc_run_dir, symlinks=True)
+    rmtree(cylc_run_dir)
+
+    # It can no longer parse IDs in the original cylc-run location.
+    with pytest.raises(InputError):
+        _, results = workflow_state(id_, task='precious', point='3000')
+
+    # But it can via an explicit alternate run directory.
+    mock_db_checker.reset_mock()
+    _, results = workflow_state(id_, task='precious', point='3000', cylc_run_dir=alt_cylc_run_dir)
+    mock_db_checker.assert_called_once_with(alt_cylc_run_dir, expected_workflow_id)
+    assert results['workflow'] == expected_workflow_id
+
 def test_back_compat(tmp_run_dir):
     """Test workflow_state xtrigger backwards compatibility with Cylc 7
     database."""