fix(create_inputs): skip empty optional File/Directory inputs

create_inputs.py

@@ -195,6 +195,15 @@ def map_inputs_from_context(job_inputs: dict, cwl_inputs: [CommandInputParameter
         input_name = get_id_from_uri(input.id)
         base_type, is_optional = parse_cwl_type(input_type)
 
+        if base_type in ("Directory", "File"):
+            raw_value = job_inputs.get(input_name)
+            if raw_value is None or raw_value == "None" or raw_value == "":
+                if is_optional:
+                    # Omit so cwltool does not pass the flag to the algorithm
+                    print(f"No value provided for optional {base_type} input '{input_name}', skipping.")
+                    continue
+                raise ValueError(f"No value provided for required {base_type} input '{input_name}'.")
+
         if base_type == "Directory":
             print("Encountered STAC input, localizing assets.")
             # Load STAC JSON directly from the URI rather than using the HySDS-localized copy.

tests/test_create_inputs.py

@@ -0,0 +1,78 @@
+import pytest
+import sys
+from pathlib import Path
+from unittest.mock import Mock, patch
+
+# Add parent directory to path to import create_inputs
+sys.path.insert(0, str(Path(__file__).parent.parent))
+
+from create_inputs import map_inputs_from_context
+
+
+def _cwl_input(name, type_):
+    """Build a minimal stand-in for a cwl_utils CommandInputParameter."""
+    return Mock(id=f"file:///process.cwl#main/{name}", type_=type_)
+
+
+class TestOptionalFileDirectoryInputs:
+    """Test handling of empty File/Directory inputs in map_inputs_from_context()."""
+
+    @pytest.mark.parametrize("type_", [["null", "File"], "File?"])
+    @pytest.mark.parametrize("value", [None, "", "None"])
+    def test_empty_optional_file_is_omitted(self, type_, value):
+        cwl_inputs = [_cwl_input("opt_file", type_)]
+        job_inputs = {} if value is None else {"opt_file": value}
+
+        assert map_inputs_from_context(job_inputs, cwl_inputs) == {}
+
+    @pytest.mark.parametrize("type_", [["null", "Directory"], "Directory?"])
+    @pytest.mark.parametrize("value", [None, "", "None"])
+    @patch('create_inputs.stage_in.download_stac_assets_from_json')
+    @patch('create_inputs.load_json_from_uri')
+    def test_empty_optional_directory_is_omitted_without_staging(self, mock_load, mock_download, type_, value):
+        cwl_inputs = [_cwl_input("opt_dir", type_)]
+        job_inputs = {} if value is None else {"opt_dir": value}
+
+        assert map_inputs_from_context(job_inputs, cwl_inputs) == {}
+        mock_load.assert_not_called()
+        mock_download.assert_not_called()
+
+    @pytest.mark.parametrize("type_", ["File", "Directory"])
+    @pytest.mark.parametrize("value", [None, ""])
+    def test_empty_required_file_or_directory_raises(self, type_, value):
+        cwl_inputs = [_cwl_input("req_input", type_)]
+        job_inputs = {} if value is None else {"req_input": value}
+
+        with pytest.raises(ValueError, match="req_input"):
+            map_inputs_from_context(job_inputs, cwl_inputs)
+
+    def test_provided_optional_file_is_included(self):
+        cwl_inputs = [_cwl_input("opt_file", ["null", "File"])]
+        job_inputs = {"opt_file": "s3://bucket/path/data.tif"}
+
+        result = map_inputs_from_context(job_inputs, cwl_inputs)
+
+        assert result["opt_file"]["class"] == "File"
+        assert result["opt_file"]["path"] == str(Path("data.tif").resolve())
+
+    @patch('create_inputs.stage_in.download_stac_assets_from_json')
+    @patch('create_inputs.load_json_from_uri', return_value={"type": "Feature"})
+    def test_provided_optional_directory_is_staged(self, mock_load, mock_download):
+        cwl_inputs = [_cwl_input("opt_dir", ["null", "Directory"])]
+        job_inputs = {"opt_dir": "https://example.com/item.json"}
+
+        result = map_inputs_from_context(job_inputs, cwl_inputs)
+
+        mock_load.assert_called_once_with("https://example.com/item.json")
+        mock_download.assert_called_once()
+        assert result["opt_dir"] == {"class": "Directory", "path": str(Path("opt_dir").resolve())}
+
+    def test_empty_optional_file_does_not_affect_other_inputs(self):
+        cwl_inputs = [
+            _cwl_input("opt_file", ["null", "File"]),
+            _cwl_input("req_str", "string"),
+            _cwl_input("opt_str", ["null", "string"]),
+        ]
+        job_inputs = {"req_str": "x", "opt_str": ""}
+
+        assert map_inputs_from_context(job_inputs, cwl_inputs) == {"req_str": "x"}