Models for YAML test format.

lib/galaxy/tool_util/models.py

@@ -5,11 +5,23 @@ for reasoning about tool state externally from Galaxy.
 """
 
 from typing import (
+    Any,
+    Dict,
     List,
     Optional,
+    Union,
 )
 
-from pydantic import BaseModel
+from pydantic import (
+    AnyUrl,
+    BaseModel,
+    ConfigDict,
+    RootModel,
+)
+from typing_extensions import (
+    NotRequired,
+    TypedDict,
+)
 
 from .parameters import (
     input_models_for_tool_source,
@@ -18,6 +30,7 @@ from .parameters import (
 from .parser.interface import (
     Citation,
     HelpContent,
+    OutputCompareType,
     ToolSource,
     XrefDict,
 )
@@ -25,6 +38,7 @@ from .parser.output_models import (
     from_tool_source,
     ToolOutput,
 )
+from .verify.assertion_models import assertions
 
 
 class ParsedTool(BaseModel):
@@ -73,3 +87,85 @@ def parse_tool(tool_source: ToolSource) -> ParsedTool:
         xrefs=xrefs,
         help=help,
     )
+
+
+class StrictModel(BaseModel):
+
+    model_config = ConfigDict(
+        extra="forbid",
+    )
+
+
+class BaseTestOutputModel(StrictModel):
+    file: Optional[str] = None
+    path: Optional[str] = None
+    location: Optional[AnyUrl] = None
+    ftype: Optional[str] = None
+    sort: Optional[bool] = None
+    compare: Optional[OutputCompareType] = None
+    checksum: Optional[str] = None
+    metadata: Optional[Dict[str, Any]] = None
+    asserts: Optional[assertions] = None
+    delta: Optional[int] = None
+    delta_frac: Optional[float] = None
+    lines_diff: Optional[int] = None
+    decompress: Optional[bool] = None
+
+
+class TestDataOutputAssertions(BaseTestOutputModel):
+    pass
+
+
+class TestCollectionCollectionElementAssertions(StrictModel):
+    elements: Optional[Dict[str, "TestCollectionElementAssertion"]] = None
+    element_tests: Optional[Dict[str, "TestCollectionElementAssertion"]] = None
+
+
+class TestCollectionDatasetElementAssertions(BaseTestOutputModel):
+    pass
+
+
+TestCollectionElementAssertion = Union[
+    TestCollectionDatasetElementAssertions, TestCollectionCollectionElementAssertions
+]
+TestCollectionCollectionElementAssertions.model_rebuild()
+
+
+class CollectionAttributes(StrictModel):
+    collection_type: Optional[str] = None
+
+
+class TestCollectionOutputAssertions(StrictModel):
+    elements: Optional[Dict[str, TestCollectionElementAssertion]] = None
+    element_tests: Optional[Dict[str, "TestCollectionElementAssertion"]] = None
+    attributes: Optional[CollectionAttributes] = None
+
+
+TestOutputLiteral = Union[bool, int, float, str]
+
+TestOutputAssertions = Union[TestCollectionOutputAssertions, TestDataOutputAssertions, TestOutputLiteral]
+
+JobDict = Dict[str, Any]
+
+
+class TestJob(StrictModel):
+    doc: Optional[str]
+    job: JobDict
+    outputs: Dict[str, TestOutputAssertions]
+
+
+Tests = RootModel[List[TestJob]]
+
+# TODO: typed dict versions of all thee above for verify code - make this Dict[str, Any] here more
+# specific.
+OutputChecks = Union[TestOutputLiteral, Dict[str, Any]]
+OutputsDict = Dict[str, OutputChecks]
+
+
+class TestJobDict(TypedDict):
+    doc: NotRequired[str]
+    job: NotRequired[JobDict]
+    outputs: OutputsDict
+
+
+TestDicts = List[TestJobDict]

lib/galaxy/tool_util/parser/interface.py

@@ -5,6 +5,7 @@ from abc import (
     ABCMeta,
     abstractmethod,
 )
+from enum import Enum
 from os.path import join
 from typing import (
     Any,
@@ -49,9 +50,18 @@ AssertionList = Optional[List[AssertionDict]]
 XmlInt = Union[str, int]
 
 
+class OutputCompareType(str, Enum):
+    diff = "diff"
+    re_match = "re_match"
+    sim_size = "sim_size"
+    re_match_multiline = "re_match_multiline"
+    contains = "contains"
+    image_diff = "image_diff"
+
+
 class ToolSourceTestOutputAttributes(TypedDict):
     object: NotRequired[Optional[Any]]
-    compare: str
+    compare: OutputCompareType
     lines_diff: int
     delta: int
     delta_frac: Optional[float]

lib/galaxy/tool_util/parser/xml.py

@@ -43,6 +43,7 @@ from .interface import (
     DynamicOptions,
     HelpContent,
     InputSource,
+    OutputCompareType,
     PageSource,
     PagesSource,
     RequiredFiles,
@@ -834,7 +835,7 @@ def __parse_test_attributes(
         value_object = json.loads(attrib.pop("value_json"))
 
     # Method of comparison
-    compare: str = attrib.pop("compare", "diff").lower()
+    compare: OutputCompareType = cast(OutputCompareType, attrib.pop("compare", "diff").lower())
     # Number of lines to allow to vary in logs (for dates, etc)
     lines_diff: int = int(attrib.pop("lines_diff", "0"))
     # Allow a file size to vary if sim_size compare

lib/galaxy/tool_util/validate_test_format.py

@@ -0,0 +1,40 @@
+#!/usr/bin/env python
+
+import argparse
+import sys
+
+import yaml
+
+from galaxy.tool_util.models import Tests
+
+DESCRIPTION = """
+A small utility to verify the Planemo test format.
+
+This script doesn't use semantic information about tools or workflows so only
+the structure of the file is checked and things like inputs matching up is not
+included.
+"""
+
+
+def validate_test_file(test_file: str) -> None:
+    with open(test_file) as f:
+        json = yaml.safe_load(f)
+    Tests.model_validate(json)
+
+
+def arg_parser() -> argparse.ArgumentParser:
+    parser = argparse.ArgumentParser(description=DESCRIPTION)
+    parser.add_argument("test_file")
+    return parser
+
+
+def main(argv=None) -> None:
+    if argv is None:
+        argv = sys.argv[1:]
+
+    args = arg_parser().parse_args(argv)
+    validate_test_file(args.test_file)
+
+
+if __name__ == "__main__":
+    main()

lib/galaxy/tool_util/verify/__init__.py

@@ -597,6 +597,8 @@ def files_image_diff(file1: str, file2: str, attributes: Optional[Dict[str, Any]
 # 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
+# TODO: Also migrate the logic for checking non-dictionaries out of Planemo - this function now
+# does that check also.
 def verify_file_path_against_dict(
     get_filename: GetFilenameT,
     get_location: GetLocationT,
@@ -621,30 +623,38 @@ def verify_file_contents_against_dict(
     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://")):
-                assert get_location
-                expected_file = get_location(location)
-            else:
-                expected_file = location.split("file://", 1)[-1]
+    expected_file: Optional[str] = None
+    if isinstance(test_properties, dict):
+        # 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://")):
+                    assert get_location
+                    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,
-    )
+        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,
+        )
+    else:
+        output_value = json.loads(output_content.decode("utf-8"))
+        if test_properties != output_value:
+            template = "Output [%s] value [%s] does not match expected value [%s]."
+            message = template % (item_label, output_value, test_properties)
+            raise AssertionError(message)
 
 
 __all__ = [

lib/galaxy/tool_util/verify/assertion_models.py

@@ -8,6 +8,7 @@ from pydantic import (
     BeforeValidator,
     ConfigDict,
     Field,
+    model_validator,
     RootModel,
     StrictFloat,
     StrictInt,

lib/galaxy/tool_util/verify/asserts/size.py

@@ -14,9 +14,7 @@ from ._util import _assert_number
 
 def assert_has_size(
     output_bytes: OutputBytes,
-    value: Annotated[
-        OptionalXmlInt, AssertionParameter("Deprecated alias for `size`", xml_type="Bytes", deprecated=True)
-    ] = None,
+    value: Annotated[OptionalXmlInt, AssertionParameter("Deprecated alias for `size`", xml_type="Bytes")] = None,
     size: Annotated[
         OptionalXmlInt,
         AssertionParameter(

lib/galaxy/tool_util/verify/codegen.py

@@ -55,6 +55,7 @@ from pydantic import (
     BeforeValidator,
     ConfigDict,
     Field,
+    model_validator,
     RootModel,
     StrictFloat,
     StrictInt,
@@ -113,9 +114,8 @@ def check_non_negative_if_int(v: typing.Any):
 {{assertion.name}}_{{ parameter.name }}_description = '''{{ parameter.description }}'''
 {% endfor %}
 
-class {{assertion.name}}_model(AssertionModel):
-    r\"\"\"{{ assertion.docstring }}\"\"\"
-    that: Literal["{{assertion.name}}"] = "{{assertion.name}}"
+class base_{{assertion.name}}_model(AssertionModel):
+    '''base model for {{assertion.name}} describing attributes.'''
 {% for parameter in assertion.parameters %}
 {% if not parameter.is_deprecated %}
     {{ parameter.name }}: {{ parameter.type_str }} = Field(
@@ -124,21 +124,52 @@ class {{assertion.name}}_model(AssertionModel):
     )
 {% endif %}
 {% endfor %}
-{% if assertion.children == "required" %}
-    children: "assertion_list"
-{% endif %}
-{% if assertion.children == "allowed" %}
+{% if assertion.children in ["required", "allowed"] %}
     children: typing.Optional["assertion_list"] = None
+    asserts: typing.Optional["assertion_list"] = None
+
+{% if assertion.children == "required" %}
+    @model_validator(mode='before')
+    @classmethod
+    def validate_children(self, data: typing.Any):
+        if isinstance(data, dict) and 'children' not in data and 'asserts' not in data:
+            raise ValueError("At least one of 'children' or 'asserts' must be specified for this assertion type.")
+        return data
 {% endif %}
+{% endif %}
+
+
+class {{assertion.name}}_model(base_{{assertion.name}}_model):
+    r\"\"\"{{ assertion.docstring }}\"\"\"
+    that: Literal["{{assertion.name}}"] = "{{assertion.name}}"
+
+class {{assertion.name}}_model_nested(AssertionModel):
+    r\"\"\"Nested version of this assertion model.\"\"\"
+    {{assertion.name}}: base_{{assertion.name}}_model
 {% endfor %}
 
-any_assertion_model = Annotated[typing.Union[
+any_assertion_model_flat = Annotated[typing.Union[
 {% for assertion in assertions %}
     {{assertion.name}}_model,
 {% endfor %}
 ], Field(discriminator="that")]
 
-assertion_list = RootModel[typing.List[any_assertion_model]]
+any_assertion_model_nested = typing.Union[
+{% for assertion in assertions %}
+    {{assertion.name}}_model_nested,
+{% endfor %}
+]
+
+assertion_list = RootModel[typing.List[typing.Union[any_assertion_model_flat, any_assertion_model_nested]]]
+
+
+class assertion_dict(AssertionModel):
+{% for assertion in assertions %}
+    {{assertion.name}}: typing.Optional[base_{{assertion.name}}_model] = None
+{% endfor %}
+
+
+assertions = typing.Union[assertion_list, assertion_dict]
 """
 
 

lib/galaxy/workflow/scheduling_manager.py

@@ -329,7 +329,6 @@ class WorkflowRequestMonitor(Monitors):
     def __attempt_schedule(self, invocation_id, workflow_scheduler):
         with self.app.model.context() as session:
             workflow_invocation = session.get(model.WorkflowInvocation, invocation_id)
-
             try:
                 if workflow_invocation.state == workflow_invocation.states.CANCELLING:
                     workflow_invocation.cancel_invocation_steps()

lib/galaxy_test/base/populators.py

@@ -3151,7 +3151,12 @@ def load_data_dict(
         elif is_dict and "type" in value:
             input_type = value.pop("type")
             if input_type == "File":
-                content = open_test_data(value)
+                if "value" in value:
+                    content = open_test_data(value)
+                elif "content" in value:
+                    content = value["content"]
+                else:
+                    raise ValueError(f"Invalid test_data def {test_data}")
                 new_dataset_kwds = {"content": content}
                 if "name" in value:
                     new_dataset_kwds["name"] = value["name"]

lib/galaxy_test/workflow/flatten_collection.gxwf-tests.yml

@@ -3,6 +3,7 @@
   job: {}
   outputs:
     out:
+      attributes: {collection_type: 'list'}
       elements:
         'oe1-ie1':
           asserts:

lib/galaxy_test/workflow/tests.py

@@ -8,6 +8,12 @@ import requests
 import yaml
 from gxformat2.yaml import ordered_load
 
+from galaxy.tool_util.models import (
+    OutputChecks,
+    OutputsDict,
+    TestDicts,
+    TestJobDict,
+)
 from galaxy.tool_util.parser.interface import TestCollectionOutputDef
 from galaxy.tool_util.verify import verify_file_contents_against_dict
 from galaxy.tool_util.verify.interactor import (
@@ -52,7 +58,7 @@ class TestWorkflow(ApiTestCase):
         self.dataset_collection_populator = DatasetCollectionPopulator(self.galaxy_interactor)
 
     @pytest.mark.workflow
-    def test_workflow(self, workflow_path: Path, test_job):
+    def test_workflow(self, workflow_path: Path, test_job: TestJobDict):
         with workflow_path.open() as f:
             yaml_content = ordered_load(f)
         with self.dataset_populator.test_history() as history_id:
@@ -63,30 +69,32 @@ class TestWorkflow(ApiTestCase):
             )
             self._verify(run_summary, test_job["outputs"])
 
-    def _verify(self, run_summary: RunJobsSummary, output_definitions):
+    def _verify(self, run_summary: RunJobsSummary, output_definitions: OutputsDict):
         for output_name, output_definition in output_definitions.items():
             self._verify_output(run_summary, output_name, output_definition)
 
-    def _verify_output(self, run_summary: RunJobsSummary, output_name, test_properties):
-        is_collection_test = "elements" in test_properties
+    def _verify_output(self, run_summary: RunJobsSummary, output_name, test_properties: OutputChecks):
+        is_collection_test = isinstance(test_properties, dict) and "elements" in test_properties
         item_label = f"Output named {output_name}"
 
         def get_filename(name):
             return tempfile.NamedTemporaryFile(prefix=f"gx_workflow_framework_test_file_{output_name}", delete=False)
 
-        def verify_dataset(dataset: dict, test_properties: dict):
+        def verify_dataset(dataset: dict, test_properties: OutputChecks):
             output_content = self.dataset_populator.get_history_dataset_content(
                 run_summary.history_id, dataset=dataset, type="bytes"
             )
             verify_file_contents_against_dict(get_filename, _get_location, item_label, output_content, test_properties)
-            metadata = get_metadata_to_test(test_properties)
-            if metadata:
-                dataset_details = self.dataset_populator.get_history_dataset_details(
-                    run_summary.history_id, content_id=dataset["id"]
-                )
-                compare_expected_metadata_to_api_response(metadata, dataset_details)
+            if isinstance(test_properties, dict):
+                metadata = get_metadata_to_test(test_properties)
+                if metadata:
+                    dataset_details = self.dataset_populator.get_history_dataset_details(
+                        run_summary.history_id, content_id=dataset["id"]
+                    )
+                    compare_expected_metadata_to_api_response(metadata, dataset_details)
 
         if is_collection_test:
+            assert isinstance(test_properties, dict)
             test_properties["name"] = output_name
             # setup preferred name "elements" in accordance with work in https://github.com/galaxyproject/planemo/pull/1417
             test_properties["element_tests"] = test_properties["elements"]
@@ -105,14 +113,15 @@ class TestWorkflow(ApiTestCase):
 
             verify_collection(output_def, output_collection, verify_dataset_element)
         else:
-            test_properties["name"] = output_name
+            if isinstance(test_properties, dict):
+                test_properties["name"] = output_name
             invocation_details = self.workflow_populator.get_invocation(run_summary.invocation_id, step_details=True)
             assert output_name in invocation_details["outputs"]
             test_output = invocation_details["outputs"][output_name]
             verify_dataset(test_output, test_properties)
 
 
-def _test_jobs(workflow_path: Path) -> list:
+def _test_jobs(workflow_path: Path) -> TestDicts:
     test_path = _workflow_test_path(workflow_path)
     with test_path.open() as f:
         jobs = yaml.safe_load(f)

packages/tool_util/setup.cfg

@@ -51,6 +51,7 @@ console_scripts =
         galaxy-tool-test = galaxy.tool_util.verify.script:main
         galaxy-tool-test-case-validation = galaxy.tool_util.parameters.scripts.validate_test_cases:main
         galaxy-tool-upgrade-advisor = galaxy.tool_util.upgrade.script:main
+        validate-test-format = galaxy.tool_util.validate_test_format:main
         mulled-build = galaxy.tool_util.deps.mulled.mulled_build:main
         mulled-build-channel = galaxy.tool_util.deps.mulled.mulled_build_channel:main
         mulled-build-files = galaxy.tool_util.deps.mulled.mulled_build_files:main

test/functional/tools/sample_tool_conf.xml

@@ -139,6 +139,7 @@
   <tool file="column_param_configfile.xml" />
   <tool file="column_param_list.xml" />
   <tool file="column_multi_param.xml" />
+  <tool file="select_optional_legacy.xml" />
   <tool file="select_optional.xml" />
   <tool file="select_dynamic.xml" />
   <tool file="hidden_param.xml" />

test/unit/tool_util/test_test_format_model.py

@@ -0,0 +1,39 @@
+import os
+from pathlib import Path
+from typing import List
+
+import yaml
+
+from galaxy.tool_util.models import Tests
+from galaxy.util import galaxy_directory
+from galaxy.util.unittest_utils import skip_unless_environ
+
+TEST_WORKFLOW_DIRECTORY = os.path.join(galaxy_directory(), "lib", "galaxy_test", "workflow")
+IWC_WORKFLOWS_USING_UNVERIFIED_SYNTAX: List[str] = []
+
+
+def test_validate_workflow_tests():
+    path = Path(TEST_WORKFLOW_DIRECTORY)
+    test_files = path.glob("*.gxwf-tests.yml")
+    for test_file in test_files:
+        with open(test_file) as f:
+            json = yaml.safe_load(f)
+        Tests.model_validate(json)
+
+
+@skip_unless_environ("GALAXY_TEST_IWC_DIRECTORY")
+def test_iwc_directory():
+    path = Path(os.environ["GALAXY_TEST_IWC_DIRECTORY"])
+    test_files = path.glob("workflows/**/*-test*.yml")
+
+    for test_file in test_files:
+        print(test_file)
+        skip_file = False
+        for unverified in IWC_WORKFLOWS_USING_UNVERIFIED_SYNTAX:
+            if str(test_file).endswith(unverified):
+                skip_file = True
+        if skip_file:
+            continue
+        with open(test_file) as f:
+            json = yaml.safe_load(f)
+        Tests.model_validate(json)