Add edge-merge-mode property in node

examples/edge-merge-mode.yaml

@@ -0,0 +1,24 @@
+- step:
+    name: step1
+    command: echo "i will buzz"
+    image: alpine/alpine
+    inputs:
+      - name: dataset
+        default: /step1
+
+- pipeline:
+    name: check-edge-merge-mode
+    nodes:
+      - name: node-with-default-edge-merge
+        type: execution
+        step: step1
+      - name: node-with-edge-merge-replace
+        type: execution
+        step: step1
+        edge-merge-mode: replace
+      - name: node-with-edge-merge-append
+        type: execution
+        step: step1
+        edge-merge-mode: append
+
+    edges: []

tests/__snapshots__/test_validation.ambr

@@ -155,6 +155,15 @@
 
   '''
 # ---
+# name: test_bad_examples_cli[wrong-edge-merge-mode.yaml]
+  '''
+  >>> wrong-edge-merge-mode.yaml
+  error:   Anyof validation on pipeline.nodes.anyOf: {'name': 'bad-pn', 'type': 'execution', 'step': 'buzz', 'edge-merge-mode': 'boo-boo'} is not valid under any of the given schemas (1.pipeline.nodes.0)
+  ------------------------------------------------------------
+  *** 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")
+edge_merge_mode_config = config_fixture("edge-merge-mode.yaml")

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

@@ -0,0 +1,16 @@
+- step:
+    name: buzz
+    command: echo "i will buzz"
+    image: alpine/alpine
+    inputs:
+      - name: dataset
+        default: /data
+
+- pipeline:
+    name: bad-pp
+    nodes:
+      - name: bad-pn
+        type: execution
+        step: buzz
+        edge-merge-mode: boo-boo
+    edges: []

tests/test_edge_merge_mode.py

@@ -0,0 +1,21 @@
+from valohai_yaml.objs.pipelines.edge_merge_mode import EdgeMergeMode
+from valohai_yaml.pipelines.conversion import PipelineConverter
+
+
+def test_edge_override_mode(edge_merge_mode_config):
+    config = 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)
+    node0 = converted_pipeline["nodes"][0]
+    node1 = converted_pipeline["nodes"][1]
+    node2 = converted_pipeline["nodes"][2]
+
+    # override-mode not defined
+    assert node0["edge-merge-mode"] == EdgeMergeMode.REPLACE.value
+
+    # override-mode = replace
+    assert node1["edge-merge-mode"] == EdgeMergeMode.REPLACE.value
+
+    # override-mode = append
+    assert node2["edge-merge-mode"] == EdgeMergeMode.APPEND.value

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-edge-merge-mode.yaml",
+            "'boo-boo'} is not valid under any of the given schemas",
+        ),
     ],
 )
 def test_expression_lint_fail(file_path, expected_message):

tests/valid_examples/edge-merge-mode.yaml

@@ -0,0 +1,16 @@
+- step:
+    name: buzz
+    command: echo "i will buzz"
+    image: alpine/alpine
+    inputs:
+      - name: dataset
+        default: /data
+
+- pipeline:
+    name: good
+    nodes:
+      - name: good_node
+        step: buzz
+        type: execution
+        edge-merge-mode: replace
+    edges: []

valohai_yaml/objs/pipelines/edge_merge_mode.py

@@ -0,0 +1,19 @@
+from enum import Enum
+from typing import Union
+
+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())

valohai_yaml/objs/pipelines/execution_node.py

@@ -1,6 +1,7 @@
 from typing import Any, Dict, List, Optional, Union
 
 from valohai_yaml.lint import LintResult
+from valohai_yaml.objs.pipelines.edge_merge_mode import EdgeMergeMode
 from valohai_yaml.objs.pipelines.node import ErrorAction, Node
 from valohai_yaml.objs.pipelines.node_action import NodeAction
 from valohai_yaml.objs.pipelines.override import Override
@@ -21,13 +22,15 @@ def __init__(
         actions: Optional[List[NodeAction]] = None,
         override: Optional[Union[Override, NodeOverrideDict]] = None,
         on_error: Union[str, ErrorAction] = ErrorAction.STOP_ALL,
+        edge_merge_mode: EdgeMergeMode = EdgeMergeMode.REPLACE,
     ) -> None:
         super().__init__(name=name, actions=actions, on_error=on_error)
         self.step = step
         if override is None or isinstance(override, Override):
             self.override = override
         else:
             self.override = Override.parse(override)
+        self.edge_merge_mode = EdgeMergeMode.cast(edge_merge_mode)
 
     def lint(self, lint_result: LintResult, context: LintContext) -> None:
         super().lint(lint_result, context)

valohai_yaml/schema/execution-node.yaml

@@ -17,6 +17,8 @@ properties:
       "$ref": "./node-action.json"
   override:
     "$ref": "./overridden-properties.json"
+  edge-merge-mode:
+    "$ref": "./node-edge-merge-mode.json"
 required:
   - name
   - step

valohai_yaml/schema/node-edge-merge-mode.yaml

@@ -0,0 +1,6 @@
+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.

valohai_yaml/schema/task-node.yaml

@@ -17,6 +17,8 @@ properties:
       "$ref": "./node-action.json"
   override:
     "$ref": "./overridden-properties.json"
+  edge-merge-mode:
+    "$ref": "./node-edge-merge-mode.json"
 required:
   - name
   - step