Add egde-merge-mode property in input

examples/input-edge-merge-mode.yaml

@@ -0,0 +1,42 @@
+- step:
+    name: step1
+    command: echo "i will buzz"
+    image: alpine/alpine
+    inputs:
+      - name: dataset
+        default: /step1
+
+- step:
+    name: step2
+    command: echo "i will buzz"
+    image: alpine/alpine
+    inputs:
+      - name: dataset
+        default: /step2
+        edge-merge-mode: replace
+
+- step:
+    name: step3
+    command: echo "i will buzz"
+    image: alpine/alpine
+    inputs:
+      - name: dataset
+        default: /step3
+        edge-merge-mode: append
+
+- pipeline:
+    name: check-edge-merge-mode
+    nodes:
+      - name: step1
+        type: execution
+        step: step1
+      - name: step2
+        type: execution
+        step: step2
+      - name: step3
+        type: execution
+        step: step3
+
+    edges:
+      - [step1.output.*, step2.input.dataset]
+      - [step1.output.*, step3.input.dataset]

tests/__snapshots__/test_validation.ambr

@@ -155,6 +155,15 @@
 
   '''
 # ---
+# name: test_bad_examples_cli[wrong-input-edge-merge-mode.yaml]
+  '''
+  >>> wrong-input-edge-merge-mode.yaml
+  error:   Enum validation on step.inputs.edge-merge-mode.enum: 'boo-boo' is not one of ['replace', 'append'] (0.step.inputs.0.edge-merge-mode)
+  ------------------------------------------------------------
+  *** 1 errors, 0 warnings
+
+  '''
+# ---
 # name: test_warning_examples_cli[invalid-numeric-parameter-default.yaml]
   '''
   >>> invalid-numeric-parameter-default.yaml

tests/conftest.py

@@ -20,3 +20,4 @@
 multiple_param_config = config_fixture("multiple-param.yaml")
 input_extras_config = config_fixture("input-extras.yaml")
 timeouts_config = config_fixture("timeouts-example.yaml")
+input_edge_merge_mode_config = config_fixture("input-edge-merge-mode.yaml")

tests/error_examples/wrong-input-edge-merge-mode.yaml

@@ -0,0 +1,8 @@
+- step:
+    name: buzz
+    command: echo "i will buzz"
+    image: alpine/alpine
+    inputs:
+      - name: dataset
+        default: /data
+        edge-merge-mode: boo-boo

tests/test_input_edge_merge_mode.py

@@ -0,0 +1,20 @@
+from valohai_yaml.pipelines.conversion import PipelineConverter
+
+
+def test_input_override_mode(input_edge_merge_mode_config):
+    config = input_edge_merge_mode_config
+    pipeline = config.pipelines["check-edge-merge-mode"]
+    assert len(pipeline.nodes) == 3
+    converted_pipeline = PipelineConverter(config=config, commit_identifier="latest").convert_pipeline(pipeline)
+    step1 = converted_pipeline["nodes"][0]
+    step2 = converted_pipeline["nodes"][1]
+    step3 = converted_pipeline["nodes"][2]
+
+    # override-mode not defined
+    assert step1["template"]["inputs"]["dataset"] == []
+
+    # override-mode = replace
+    assert step2["template"]["inputs"]["dataset"] == []
+
+    # override-mode = append
+    assert step3["template"]["inputs"]["dataset"] == ["/step3"]

tests/test_linter.py

@@ -75,6 +75,10 @@ def test_expression_lint_ok(file_path):
             "task-stop-condition.yaml",
             "Task no-stop, `stop-condition` is not a valid expression:",
         ),
+        (
+            "wrong-input-edge-merge-mode.yaml",
+            "'boo-boo' is not one of ['replace', 'append']",
+        ),
     ],
 )
 def test_expression_lint_fail(file_path, expected_message):

tests/valid_examples/input-edge-merge-mode.yaml

@@ -0,0 +1,8 @@
+- step:
+    name: buzz
+    command: echo "i will buzz"
+    image: alpine/alpine
+    inputs:
+      - name: dataset
+        default: /data
+        edge-merge-mode: replace

valohai_yaml/objs/input.py

@@ -5,6 +5,22 @@
 from valohai_yaml.types import SerializedDict
 
 KeepDirectoriesValue = Union[bool, str, "KeepDirectories"]
+EdgeMergeModeValue = Union[str, "EdgeMergeMode"]
+
+
+class EdgeMergeMode(Enum):
+    """Input override mode."""
+
+    REPLACE = "replace"
+    APPEND = "append"
+
+    @classmethod
+    def cast(cls, value: EdgeMergeModeValue) -> "EdgeMergeMode":
+        if isinstance(value, EdgeMergeMode):
+            return value
+        if not value:
+            return EdgeMergeMode.REPLACE
+        return EdgeMergeMode(str(value).lower())
 
 
 class KeepDirectories(Enum):
@@ -37,13 +53,15 @@ def __init__(
         description: Optional[str] = None,
         keep_directories: KeepDirectoriesValue = False,
         filename: Optional[str] = None,
+        edge_merge_mode: EdgeMergeMode = EdgeMergeMode.REPLACE,
     ) -> None:
         self.name = name
         self.default = default  # may be None, a string or a list of strings
         self.optional = bool(optional)
         self.description = description
         self.keep_directories = KeepDirectories.cast(keep_directories)
         self.filename = filename
+        self.edge_merge_mode = EdgeMergeMode.cast(edge_merge_mode)
 
     def get_data(self) -> SerializedDict:
         data = super().get_data()

valohai_yaml/schema/input-item.yaml

@@ -38,3 +38,10 @@ properties:
           - "suffix"
     default: false
     description: Whether to retain directories when using wildcards for this input.
+  edge-merge-mode:
+    type: string
+    enum:
+      - replace
+      - append
+    default: replace
+    description: Whether to replace the values set for this input, or to append to them when overridden by e.g. a pipeline edge.