From ac614a619e1e599ec2341fb037d6e3f396a944ab Mon Sep 17 00:00:00 2001
From: yuri1969 <1969yuri1969@gmail.com>
Date: Wed, 11 Sep 2024 21:35:08 +0200
Subject: [PATCH] fix(core): make flow plugin defaults override global ones

This handles the OSS precedence.
---
 .../core/services/PluginDefaultService.java   |  8 +--
 .../services/PluginDefaultServiceTest.java    | 68 +++++++++++++++++++
 2 files changed, 72 insertions(+), 4 deletions(-)

diff --git a/core/src/main/java/io/kestra/core/services/PluginDefaultService.java b/core/src/main/java/io/kestra/core/services/PluginDefaultService.java
index cfd2657de3..4bbe85afc4 100644
--- a/core/src/main/java/io/kestra/core/services/PluginDefaultService.java
+++ b/core/src/main/java/io/kestra/core/services/PluginDefaultService.java
@@ -68,6 +68,10 @@ public class PluginDefaultService {
     protected List<PluginDefault> mergeAllDefaults(Flow flow) {
         List<PluginDefault> list = new ArrayList<>();
 
+        if (flow.getPluginDefaults() != null) {
+            list.addAll(flow.getPluginDefaults());
+        }
+
         if (taskGlobalDefault != null && taskGlobalDefault.getDefaults() != null) {
             if (warnOnce.compareAndSet(false, true)) {
                 log.warn("Global Task Defaults are deprecated, please use Global Plugin Defaults instead via the 'kestra.plugins.defaults' property.");
@@ -79,10 +83,6 @@ protected List<PluginDefault> mergeAllDefaults(Flow flow) {
             list.addAll(pluginGlobalDefault.getDefaults());
         }
 
-        if (flow.getPluginDefaults() != null) {
-            list.addAll(flow.getPluginDefaults());
-        }
-
         return list;
     }
 
diff --git a/core/src/test/java/io/kestra/core/services/PluginDefaultServiceTest.java b/core/src/test/java/io/kestra/core/services/PluginDefaultServiceTest.java
index 995b85ffb3..3060f4ec3d 100644
--- a/core/src/test/java/io/kestra/core/services/PluginDefaultServiceTest.java
+++ b/core/src/test/java/io/kestra/core/services/PluginDefaultServiceTest.java
@@ -25,12 +25,16 @@
 import lombok.experimental.SuperBuilder;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
 
 import java.time.Duration;
 import java.util.Collections;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
+import java.util.stream.Stream;
 
 import static org.hamcrest.MatcherAssert.assertThat;
 import static org.hamcrest.Matchers.containsInAnyOrder;
@@ -99,6 +103,51 @@ void shouldInjectGivenSimpleDefaults() {
         ), result);
     }
 
+    @ParameterizedTest
+    @MethodSource
+    void flowDefaultsOverrideGlobalDefaults(boolean flowDefaultForced, boolean globalDefaultForced, String fooValue, String barValue, String bazValue) {
+        final DefaultPrecedenceTester task = DefaultPrecedenceTester.builder()
+            .id("test")
+            .type(DefaultPrecedenceTester.class.getName())
+            .propBaz("taskValue")
+            .build();
+
+        final PluginDefault flowDefault = new PluginDefault(DefaultPrecedenceTester.class.getName(), flowDefaultForced, ImmutableMap.of(
+            "propBar", "flowValue",
+            "propBaz", "flowValue"
+        ));
+        final PluginDefault globalDefault = new PluginDefault(DefaultPrecedenceTester.class.getName(), globalDefaultForced, ImmutableMap.of(
+            "propFoo", "globalValue",
+            "propBar", "globalValue",
+            "propBaz", "globalValue"
+        ));
+
+        final Flow flowWithPluginDefault = Flow.builder()
+            .tasks(Collections.singletonList(task))
+            .pluginDefaults(List.of(flowDefault))
+            .build();
+
+        final PluginGlobalDefaultConfiguration pluginGlobalDefaultConfiguration = new PluginGlobalDefaultConfiguration();
+        pluginGlobalDefaultConfiguration.defaults = List.of(globalDefault);
+
+        pluginDefaultService.pluginGlobalDefault = pluginGlobalDefaultConfiguration;
+
+        final Flow injected = pluginDefaultService.injectDefaults(flowWithPluginDefault);
+
+        assertThat(((DefaultPrecedenceTester) injected.getTasks().getFirst()).getPropFoo(), is(fooValue));
+        assertThat(((DefaultPrecedenceTester) injected.getTasks().getFirst()).getPropBar(), is(barValue));
+        assertThat(((DefaultPrecedenceTester) injected.getTasks().getFirst()).getPropBaz(), is(bazValue));
+    }
+
+    private static Stream<Arguments> flowDefaultsOverrideGlobalDefaults() {
+        return Stream.of(
+            Arguments.of(false, false, "globalValue", "flowValue", "taskValue"),
+            Arguments.of(false, true, "globalValue", "globalValue", "globalValue"),
+            Arguments.of(true, false, "globalValue", "flowValue", "flowValue"),
+            Arguments.of(true, true, "globalValue", "flowValue", "flowValue")
+        );
+    }
+
     @Test
     public void injectFlowAndGlobals() {
         DefaultTester task = DefaultTester.builder()
@@ -297,4 +346,23 @@ public static class Lists {
             private Map<String, String> val;
         }
     }
+
+    @SuperBuilder
+    @ToString
+    @EqualsAndHashCode
+    @Getter
+    @NoArgsConstructor
+    @Plugin(aliases = "io.kestra.core.services.DefaultPrecedenceTesterAlias")
+    public static class DefaultPrecedenceTester extends Task implements RunnableTask<VoidOutput> {
+        private String propFoo;
+
+        private String propBar;
+
+        private String propBaz;
+
+        @Override
+        public VoidOutput run(RunContext runContext) throws Exception {
+            return null;
+        }
+    }
 }
\ No newline at end of file