Models for YAML test format.

lib/galaxy/tool_util/models.py

@@ -5,11 +5,19 @@
 """
 
 from typing import (
+    Any,
+    Dict,
     List,
     Optional,
+    Union,
 )
 
-from pydantic import BaseModel
+from pydantic import (
+    AnyUrl,
+    BaseModel,
+    ConfigDict,
+    RootModel,
+)
 
 from .parameters import (
     input_models_for_tool_source,
@@ -25,6 +33,7 @@
     from_tool_source,
     ToolOutput,
 )
+from .verify.assertion_models import assertions
 
 
 class ParsedTool(BaseModel):
@@ -73,3 +82,68 @@ 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
+    sorted: Optional[bool] = None
+    compare: Optional[str] = None
+    checksum: Optional[str] = None
+    metadata: Optional[Dict[str, Any]] = None
+    asserts: Optional[assertions] = None
+    compare: Optional[str] = None  # to do make more specific
+    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
+
+
+TestOutputAssertions = Union[TestCollectionOutputAssertions, TestDataOutputAssertions]
+
+
+class TestJob(StrictModel):
+    doc: Optional[str]
+    job: Dict[str, Any]
+    outputs: Dict[str, TestOutputAssertions]
+
+
+Tests = RootModel[List[TestJob]]

lib/galaxy/tool_util/verify/assertion_models.py

@@ -8,6 +8,7 @@
     BeforeValidator,
     ConfigDict,
     Field,
+    model_validator,
     RootModel,
     StrictFloat,
     StrictInt,

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

@@ -14,9 +14,7 @@
 
 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 @@
     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 in ["required", "allowed"] %}
+    children: typing.Optional["assertion_list"] = None
+    asserts: typing.Optional["assertion_list"] = None
+
 {% if assertion.children == "required" %}
-    children: "assertion_list"
+    @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 %}
-{% if assertion.children == "allowed" %}
-    children: typing.Optional["assertion_list"] = None
 {% 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 @@ def __schedule(self, workflow_scheduler_id, workflow_scheduler):
     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/workflow/flatten_collection.gxwf-tests.yml

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

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,42 @@
+import os
+from pathlib import Path
+
+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")
+USING_UNVERIFIED_SYNTAX = [
+    "Assembly-Hifi-Trio-phasing-VGP5-tests.yml",
+    "Assembly-Hifi-HiC-phasing-VGP4-tests.yml",
+    "Assembly-Hifi-only-VGP3-tests.yml",
+]
+
+
+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")
+    print(f"test_files:: {test_files}")
+    for test_file in test_files:
+        print(test_file)
+        skip_file = False
+        for unverified in 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)