feat: HEXA-1146 Propagate pipeline default parameters in new versions

hexa/pipelines/models.py

@@ -341,6 +341,40 @@ def is_schedulable(self):
         elif self.type == PipelineType.ZIPFILE:
             return self.last_version and self.last_version.is_schedulable
 
+    def get_config_from_previous_version(self, new_parameters: dict):
+        """
+        Get the config from the previous version of the pipeline considering only overlapping parameters between the new and the previous version.
+        """
+        previous_config_from_overlapping_parameters = {}
+        if self.last_version:
+            previous_parameters = self.last_version.parameters
+            overlapping_parameters = [
+                new_parameter
+                for new_parameter in new_parameters
+                if new_parameter in previous_parameters
+            ]
+            previous_config_from_overlapping_parameters = {
+                overlapping_parameter["code"]: value
+                for overlapping_parameter in overlapping_parameters
+                if (
+                    value := self.last_version.config.get(
+                        overlapping_parameter["code"],
+                        overlapping_parameter.get("default"),
+                    )
+                )
+                is not None
+            }
+        return {
+            new_parameter["code"]: value
+            for new_parameter in new_parameters
+            if (
+                value := previous_config_from_overlapping_parameters.get(
+                    new_parameter["code"], new_parameter.get("default")
+                )
+            )
+            is not None
+        }
+
     def upload_new_version(
         self,
         user: User,
@@ -355,14 +389,8 @@ def upload_new_version(
         if not user.has_perm("pipelines.update_pipeline", self):
             raise PermissionDenied
 
-        if config is None:
-            # No default configuration has been provided, let's take the default values from the parameters
-            # In the future, we'll use the one from the last version
-            config = {
-                parameter["code"]: parameter["default"]
-                for parameter in parameters
-                if parameter.get("default") is not None
-            }
+        config = config or self.get_config_from_previous_version(parameters)
+
         version = PipelineVersion(
             user=user,
             pipeline=self,

hexa/pipelines/tests/test_models.py

@@ -185,6 +185,110 @@ def test_mail_run_recipients_mail_all_recipients(self):
         mail_run_recipients(run)
         self.assertEqual(len(mail.outbox), 3)
 
+    def test_get_config_from_previous_version(self):
+        pipeline = Pipeline.objects.create(
+            name="Test pipeline",
+        )
+        pipeline.upload_new_version(
+            user=self.USER_ADMIN,
+            zipfile=b"",
+            parameters=[
+                {
+                    "choices": None,
+                    "code": "param_1",
+                    "default": None,
+                    "help": None,
+                    "multiple": False,
+                    "name": "Param 1",
+                    "required": True,
+                    "type": "int",
+                },
+                {
+                    "choices": None,
+                    "code": "param_2",
+                    "default": None,
+                    "help": None,
+                    "multiple": False,
+                    "name": "Param 2",
+                    "required": True,
+                    "type": "int",
+                },
+            ],
+            name="Version 1",
+            config={"param_1": 43, "param_2": 42},
+        )
+        self.assertEqual(
+            {"param_1": 43, "param_2": 42},
+            pipeline.last_version.config,
+            "Initial config",
+        )
+        pipeline.upload_new_version(
+            user=self.USER_ADMIN,
+            zipfile=b"",
+            parameters=[
+                {
+                    "choices": None,
+                    "code": "param_1",
+                    "default": None,
+                    "help": None,
+                    "multiple": False,
+                    "name": "Param 1",
+                    "required": True,
+                    "type": "int",
+                },
+                {
+                    "choices": None,
+                    "code": "param_3",
+                    "default": None,
+                    "help": None,
+                    "multiple": False,
+                    "name": "Param 3",
+                    "required": True,
+                    "type": "int",
+                },
+            ],
+            name="Version 2",
+            config=None,
+        )
+        self.assertEqual(
+            {"param_1": 43},
+            pipeline.last_version.config,
+            "Config from previous version with a partial change of parameters",
+        )
+        pipeline.upload_new_version(
+            user=self.USER_ADMIN,
+            zipfile=b"",
+            parameters=[
+                {
+                    "choices": None,
+                    "code": "param_1",
+                    "default": 45,
+                    "help": None,
+                    "multiple": False,
+                    "name": "Param 1",
+                    "required": True,
+                    "type": "int",
+                },
+                {
+                    "choices": None,
+                    "code": "param_2",
+                    "default": 46,
+                    "help": None,
+                    "multiple": False,
+                    "name": "Param 2",
+                    "required": True,
+                    "type": "int",
+                },
+            ],
+            name="Version 3",
+            config=None,
+        )
+        self.assertEqual(
+            {"param_1": 45, "param_2": 46},
+            pipeline.last_version.config,
+            "Config from previous version with a change of default values",
+        )
+
     def test_get_or_create_template(self):
         template_name = "Test Template"
         template = self.PIPELINE.get_or_create_template(