Validate BioCompute Objects against schema in tests

lib/galaxy_test/base/json_schema_utils.py

@@ -0,0 +1,33 @@
+from typing import (
+    Any,
+    Dict,
+)
+
+import jsonschema
+import requests
+
+from galaxy_test.base import api_asserts
+
+schema_store: Dict[str, Any] = {}
+
+
+class JsonSchemaValidator:
+    @staticmethod
+    def validate_using_schema_url(instance: dict, schema_url: str):
+        schema = schema_store.get(schema_url, None)
+        if not schema:
+            response = requests.get(schema_url)
+            api_asserts.assert_status_code_is_ok(response)
+            schema = response.json()
+            schema_store[schema_url] = schema
+        JsonSchemaValidator.validate(instance, schema)
+
+    @staticmethod
+    def validate(instance: dict, schema: dict):
+        try:
+            schema_version = schema.get("$id", "Unknown schema version")
+            jsonschema.validate(instance=instance, schema=schema)
+        except jsonschema.exceptions.ValidationError as err:
+            raise AssertionError(
+                f"The instance does not validate against the schema: {schema_version}.\nReasons:\n{err}"
+            )

lib/galaxy_test/base/populators.py

@@ -97,6 +97,7 @@ from galaxy.util import (
     galaxy_root_path,
 )
 from galaxy.util.resources import resource_string
+from galaxy_test.base.json_schema_utils import JsonSchemaValidator
 from . import api_asserts
 from .api import ApiTestInteractor
 from .api_util import random_name
@@ -1555,33 +1556,7 @@ class BaseWorkflowPopulator(BasePopulator):
     def validate_biocompute_object(
         self, bco, expected_schema_version="https://w3id.org/ieee/ieee-2791-schema/2791object.json"
     ):
-        # TODO: actually use jsonref and jsonschema to validate this someday
-        api_asserts.assert_has_keys(
-            bco,
-            "object_id",
-            "spec_version",
-            "etag",
-            "provenance_domain",
-            "usability_domain",
-            "description_domain",
-            "execution_domain",
-            "parametric_domain",
-            "io_domain",
-            "error_domain",
-        )
-        assert bco["spec_version"] == expected_schema_version
-        api_asserts.assert_has_keys(bco["description_domain"], "keywords", "xref", "platform", "pipeline_steps")
-        api_asserts.assert_has_keys(
-            bco["execution_domain"],
-            "script",
-            "script_driver",
-            "software_prerequisites",
-            "external_data_endpoints",
-            "environment_variables",
-        )
-        for p in bco["parametric_domain"]:
-            api_asserts.assert_has_keys(p, "param", "value", "step")
-        api_asserts.assert_has_keys(bco["io_domain"], "input_subdomain", "output_subdomain")
+        JsonSchemaValidator.validate_using_schema_url(bco, expected_schema_version)
 
     def invoke_workflow_raw(self, workflow_id: str, request: dict, assert_ok: bool = False) -> Response:
         url = f"workflows/{workflow_id}/invocations"

test/integration/test_workflow_tasks.py

@@ -64,15 +64,8 @@ class TestWorkflowTasksIntegration(PosixFileSourceSetup, IntegrationTestCase, Us
     def test_export_bco_basic(self):
         bco_path = self._export_invocation_to_format(extension="bco.json", to_uri=False)
         with open(bco_path) as f:
-            bco_json = json.load(f)
-        assert bco_json["spec_version"] == "https://w3id.org/ieee/ieee-2791-schema/2791object.json"
-        assert bco_json["object_id"]
-        assert bco_json["description_domain"]
-        assert bco_json["execution_domain"]
-        assert bco_json["extension_domain"]
-        assert bco_json["io_domain"]
-        assert bco_json["parametric_domain"]
-        assert bco_json["provenance_domain"]
+            bco = json.load(f)
+        self.workflow_populator.validate_biocompute_object(bco)
 
     def _export_invocation_to_format(self, extension: str, to_uri: bool):
         with self.dataset_populator.test_history() as history_id: