Merge pull request #3386 from mirpedrol/schema-allow-oneof-anyof-allof

mirpedrol · web-flow · commit b68580097adc · 2025-01-09T18:00:12.000+01:00
Parameters schema validation: allow oneOf, anyOf and allOf with `required`
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -14,6 +14,8 @@
 
 ### General
 
+- Parameters schema validation: allow oneOf, anyOf and allOf with `required` ([#3386](https://github.com/nf-core/tools/pull/3386))
+
 ### Version updates
 
 ## [v3.1.1 - Brass Boxfish Patch](https://github.com/nf-core/tools/releases/tag/3.1.1) - [2024-12-20]
diff --git a/nf_core/pipelines/schema.py b/nf_core/pipelines/schema.py
@@ -327,11 +327,32 @@ def validate_default_params(self):
             schema_no_required = copy.deepcopy(self.schema)
             if "required" in schema_no_required:
                 schema_no_required.pop("required")
+            for keyword in ["allOf", "anyOf", "oneOf"]:
+                if keyword in schema_no_required:
+                    for i, kw_content in enumerate(schema_no_required[keyword]):
+                        if "required" in kw_content:
+                            schema_no_required[keyword][i].pop("required")
+                    schema_no_required[keyword] = [
+                        kw_content for kw_content in schema_no_required[keyword] if kw_content
+                    ]
+                    if not schema_no_required[keyword]:
+                        schema_no_required.pop(keyword)
             for group_key, group in schema_no_required.get(self.defs_notation, {}).items():
                 if "required" in group:
                     schema_no_required[self.defs_notation][group_key].pop("required")
+                for keyword in ["allOf", "anyOf", "oneOf"]:
+                    if keyword in group:
+                        for i, kw_content in enumerate(group[keyword]):
+                            if "required" in kw_content:
+                                schema_no_required[self.defs_notation][group_key][keyword][i].pop("required")
+                        schema_no_required[self.defs_notation][group_key][keyword] = [
+                            kw_content for kw_content in group[keyword] if kw_content
+                        ]
+                        if not group[keyword]:
+                            schema_no_required[self.defs_notation][group_key].pop(keyword)
             jsonschema.validate(self.schema_defaults, schema_no_required)
         except jsonschema.exceptions.ValidationError as e:
+            log.debug(f"Complete error message:\n{e}")
             raise AssertionError(f"Default parameters are invalid: {e.message}")
         for param, default in self.schema_defaults.items():
             if default in ("null", "", None, "None") or default is False:
diff --git a/tests/pipelines/test_schema.py b/tests/pipelines/test_schema.py
@@ -285,6 +285,89 @@ def test_remove_schema_notfound_configs_childschema(self):
         assert len(params_removed) == 1
         assert "foo" in params_removed
 
+    def test_validate_defaults(self):
+        """Test validating default values"""
+        self.schema_obj.schema = {
+            "properties": {"foo": {"type": "string"}, "bar": {"type": "string"}},
+            "required": ["foo"],
+        }
+        self.schema_obj.schema_defaults = {"foo": "foo", "bar": "bar"}
+        self.schema_obj.no_prompts = True
+        try:
+            self.schema_obj.validate_default_params()
+        except AssertionError:
+            self.fail("Error validating schema defaults")
+
+    def test_validate_defaults_required(self):
+        """Test validating default values when required params don't have a default"""
+        self.schema_obj.schema = {
+            "properties": {"foo": {"type": "string"}, "bar": {"type": "string"}},
+            "required": ["foo"],
+        }
+        self.schema_obj.schema_defaults = {}
+        self.schema_obj.no_prompts = True
+        try:
+            self.schema_obj.validate_default_params()
+        except AssertionError:
+            self.fail("Error validating schema defaults")
+
+    def test_validate_defaults_required_inside_group(self):
+        """Test validating default values when required params don't have a default, inside a group"""
+        self.schema_obj.schema = {
+            "$defs": {
+                "subSchemaId": {
+                    "properties": {"foo": {"type": "string"}, "bar": {"type": "string"}},
+                    "required": ["foo"],
+                },
+            }
+        }
+        self.schema_obj.schema_defaults = {}
+        self.schema_obj.no_prompts = True
+        try:
+            self.schema_obj.validate_default_params()
+        except AssertionError:
+            self.fail("Error validating schema defaults")
+
+    def test_validate_defaults_required_inside_group_with_anyof(self):
+        """Test validating default values when required params don't have a default, inside a group with anyOf"""
+        self.schema_obj.schema = {
+            "$defs": {
+                "subSchemaId": {
+                    "anyOf": [{"required": ["foo"]}, {"required": ["bar"]}],
+                    "properties": {"foo": {"type": "string"}, "bar": {"type": "string"}},
+                },
+            }
+        }
+        self.schema_obj.schema_defaults = {}
+        self.schema_obj.no_prompts = True
+        try:
+            self.schema_obj.validate_default_params()
+        except AssertionError:
+            self.fail("Error validating schema defaults")
+
+    def test_validate_defaults_required_with_anyof(self):
+        """Test validating default values when required params don't have a default, with anyOf"""
+        self.schema_obj.schema = {
+            "properties": {"foo": {"type": "string"}, "bar": {"type": "string"}, "baz": {"type": "string"}},
+            "anyOf": [{"required": ["foo"]}, {"required": ["bar"]}],
+        }
+        self.schema_obj.schema_defaults = {"baz": "baz"}
+        self.schema_obj.no_prompts = True
+        try:
+            self.schema_obj.validate_default_params()
+        except AssertionError:
+            self.fail("Error validating schema defaults")
+
+    def test_validate_defaults_error(self):
+        """Test validating default raises an exception when a default is not valid"""
+        self.schema_obj.schema = {
+            "properties": {"foo": {"type": "string"}},
+        }
+        self.schema_obj.schema_defaults = {"foo": 1}
+        self.schema_obj.no_prompts = True
+        with self.assertRaises(AssertionError):
+            self.schema_obj.validate_default_params()
+
     def test_add_schema_found_configs(self):
         """Try adding a new parameter to the schema from the config"""
         self.schema_obj.pipeline_params = {"foo": "bar"}
