Add test for updating multiple workflows with overlapping tools

tests/test_cmd_autoupdate.py

@@ -2,6 +2,7 @@
 
 import json
 import os
+import shutil
 import tempfile
 from contextlib import contextmanager
 
@@ -33,7 +34,7 @@ def create_tmp_test_tool_file(tool_version):
     ) as t:
         t.write(xml_str)
         t.flush()
-        yield t.name
+        yield os.path.realpath(t.name)
 
 
 class CmdAutoupdateTestCase(CliTestCase):
@@ -80,31 +81,41 @@ def test_autoupdate_no_update_needed(self):
             result = self._runner.invoke(self._cli.planemo, autoupdate_command)
             assert f"No updates required or made to {xmlfile}." in result.output
 
-    def test_autoupdate_workflow(self):
+    def test_autoupdate_multiple_workflows(self):
         """Test autoupdate command for a workflow is needed."""
-        with self._isolate_with_test_data("wf_repos/autoupdate_tests") as f:
-            wf_file = os.path.realpath(os.path.join(f, "diff-refactor-test.ga"))
-            autoupdate_command = ["autoupdate", wf_file]
+        with self._isolate_with_test_data("wf_repos/autoupdate_tests") as f, tempfile.TemporaryDirectory(
+            dir=f, prefix="autoupdate_test"
+        ) as isolated_dir:
+            source_file = os.path.join(f, "diff-refactor-test.ga")
+            # We update identical workflows in the same autoupdate call,
+            # both workflows must be updated.
+            targets = [os.path.join(isolated_dir, wf) for wf in ("wf1.ga", "wf2.ga")]
+            for target in targets:
+                shutil.copy(source_file, target)
+            autoupdate_command = ["autoupdate", *targets]
             result = self._runner.invoke(self._cli.planemo, autoupdate_command)
-            assert f"Auto-updating workflow {wf_file}" in result.output
-            with open(wf_file) as g:
-                wf = json.load(g)
-            # check tool within parent wf has updated
-            assert wf["steps"]["1"]["tool_version"] == "3.7+galaxy0"
-            # check tool within subworkflow has updated
-            assert wf["steps"]["2"]["subworkflow"]["steps"]["1"]["tool_version"] == "3.7+galaxy0"
-            assert (
-                wf["steps"]["2"]["subworkflow"]["steps"]["1"]["tool_id"]
-                == "toolshed.g2.bx.psu.edu/repos/bgruening/diff/diff/3.7+galaxy0"
-            )
-            assert wf["version"] == 2
-            assert wf["release"] == "0.1.1"
-
-            result = self._runner.invoke(self._cli.planemo, autoupdate_command)  # rerun on already updated WF
-            assert "No newer tool versions were found, so the workflow was not updated." in result.output
-
+            assert "Auto-updating workflow" in result.output
+            for wf_file in targets:
+                with open(wf_file) as g:
+                    wf = json.load(g)
+                # check tool within parent wf has updated
+                assert wf["steps"]["1"]["tool_version"] == "3.7+galaxy0"
+                # check tool within subworkflow has updated
+                assert wf["steps"]["2"]["subworkflow"]["steps"]["1"]["tool_version"] == "3.7+galaxy0"
+                assert (
+                    wf["steps"]["2"]["subworkflow"]["steps"]["1"]["tool_id"]
+                    == "toolshed.g2.bx.psu.edu/repos/bgruening/diff/diff/3.7+galaxy0"
+                )
+                assert wf["version"] == 2
+                assert wf["release"] == "0.1.1"
+
+                result = self._runner.invoke(self._cli.planemo, autoupdate_command)  # rerun on already updated WF
+                assert "No newer tool versions were found, so the workflow was not updated." in result.output
+
+    def test_autoupdate_gxformat2_workflow(self):
+        with self._isolate_with_test_data("wf_repos/autoupdate_tests") as f:
             wf_file = os.path.join(f, "diff-refactor-test.gxwf.yml")
-            autoupdate_command[1] = wf_file
+            autoupdate_command = ["autoupdate", wf_file]
             result = self._runner.invoke(self._cli.planemo, autoupdate_command)
             assert f"Auto-updating workflow {wf_file}" in result.output
 
@@ -117,7 +128,7 @@ def test_autoupdate_workflow(self):
 
     def test_autoupdate_workflow_from_multiple_tool_sheds(self):
         with self._isolate_with_test_data("wf_repos/autoupdate_tests") as f:
-            wf_file = os.path.realpath(os.path.join(f, "wf_autoupdate_test_multiple_repos.ga"))
+            wf_file = os.path.join(f, "wf_autoupdate_test_multiple_repos.ga")
             autoupdate_command = ["autoupdate", wf_file]
             result = self._runner.invoke(self._cli.planemo, autoupdate_command)
             assert f"Auto-updating workflow {wf_file}" in result.output

tests/test_utils.py

@@ -128,7 +128,7 @@ def _isolate_with_test_data(self, relative_path):
         with self._isolate() as f:
             repo = os.path.join(TEST_DATA_DIR, relative_path)
             self._copy_directory(repo, f)
-            yield f
+            yield os.path.realpath(f)
 
     def _copy_repo(self, name, dest):
         repo = os.path.join(TEST_REPOS_DIR, name)