Setup abstractions for Planemo & in-galaxy wf testing

lib/galaxy/tool_util/parser/yaml.py

@@ -310,6 +310,8 @@ def to_test_assert_list(assertions) -> AssertionList:
     return assert_list or None  # XML variant is None if no assertions made
 
 
+# Planemo depends on this and was never updated unfortunately.
+# https://github.com/galaxyproject/planemo/blob/master/planemo/test/_check_output.py
 __to_test_assert_list = to_test_assert_list
 
 

lib/galaxy/tool_util/verify/__init__.py

@@ -47,6 +47,7 @@ from galaxy.tool_util.parser.util import (
 )
 from galaxy.util import unicodify
 from galaxy.util.compression_utils import get_fileobj
+from galaxy.tool_util.parser.yaml import to_test_assert_list
 from .asserts import verify_assertions
 from .test_data import TestDataResolver
 
@@ -56,6 +57,8 @@ if TYPE_CHECKING:
 log = logging.getLogger(__name__)
 
 DEFAULT_TEST_DATA_RESOLVER = TestDataResolver()
+GetFilenameT = Optional[Callable[[str], str]]
+GetLocationT = Optional[Callable[[str], bytes]]
 
 
 def verify(
@@ -64,7 +67,7 @@ def verify(
     attributes: Optional[Dict[str, Any]],
     filename: Optional[str] = None,
     get_filecontent: Optional[Callable[[str], bytes]] = None,
-    get_filename: Optional[Callable[[str], str]] = None,
+    get_filename: GetFilenameT = None,
     keep_outputs_dir: Optional[str] = None,
     verify_extra_files: Optional[Callable] = None,
     mode="file",
@@ -585,3 +588,39 @@ def files_image_diff(file1: str, file2: str, attributes: Optional[Dict[str, Any]
     distance_eps = attributes.get("eps", DEFAULT_EPS)
     if distance > distance_eps:
         raise AssertionError(f"Image difference {distance} exceeds eps={distance_eps}.")
+
+
+# TODO: After tool-util with this included is published, fefactor planemo.test._check_output
+# to use this function. There is already a comment there about breaking fewer abstractions.
+# https://github.com/galaxyproject/planemo/blob/master/planemo/test/_check_output.py
+def verify_file_path_against_dict(get_filename: GetFilenameT, get_location: GetLocationT, path: str, output_content: bytes, test_properties, test_data_target_dir: Optional[str] = None) -> None:
+    with open(path, "rb") as f:
+        output_content = f.read()
+    item_label = "Output with path %s" % path
+    verify_file_contents_against_dict(get_filename, get_location, item_label, output_content, test_properties, test_data_target_dir)
+
+
+def verify_file_contents_against_dict(get_filename: GetFilenameT, get_location: GetLocationT, item_label: str, output_content: bytes, test_properties, test_data_target_dir: Optional[str] = None) -> None:
+    # Support Galaxy-like file location (using "file") or CWL-like ("path" or "location").
+    expected_file = test_properties.get("file", None)
+    if expected_file is None:
+        expected_file = test_properties.get("path", None)
+    if expected_file is None:
+        location = test_properties.get("location")
+        if location:
+            if location.startswith(("http://", "https://")):
+                expected_file = get_location(location)
+            else:
+                expected_file = location.split("file://", 1)[-1]
+
+    if "asserts" in test_properties:
+        test_properties["assert_list"] = to_test_assert_list(test_properties["asserts"])
+    verify(
+        item_label,
+        output_content,
+        attributes=test_properties,
+        filename=expected_file,
+        get_filename=get_filename,
+        keep_outputs_dir=test_data_target_dir,
+        verify_extra_files=None,
+    )