feat(system): add Unit test model and taskfixture

core/src/main/java/io/kestra/core/models/Label.java

@@ -20,6 +20,7 @@ public record Label(@NotNull String key, @NotNull String value) {
     public static final String REPLAY = SYSTEM_PREFIX + "replay";
     public static final String REPLAYED = SYSTEM_PREFIX + "replayed";
     public static final String SIMULATED_EXECUTION = SYSTEM_PREFIX + "simulatedExecution";
+    public static final String TEST = SYSTEM_PREFIX + "test";
 
     /**
      * Static helper method for converting a list of labels to a nested map.

core/src/main/java/io/kestra/core/models/executions/Execution.java

@@ -22,6 +22,7 @@
 import io.kestra.core.serializers.ListOrMapOfLabelDeserializer;
 import io.kestra.core.serializers.ListOrMapOfLabelSerializer;
 import io.kestra.core.services.LabelService;
+import io.kestra.core.test.flow.TaskFixture;
 import io.kestra.core.utils.IdUtils;
 import io.kestra.core.utils.MapUtils;
 import io.micronaut.core.annotation.Nullable;
@@ -112,6 +113,10 @@ public class Execution implements DeletedInterface, TenantInterface {
     @Setter
     String traceParent;
 
+    @With
+    @Nullable
+    List<TaskFixture> fixtures;
+
     /**
      * Factory method for constructing a new {@link Execution} object for the given {@link Flow}.
      *
@@ -210,7 +215,8 @@ public Execution withState(State.Type state) {
             this.deleted,
             this.metadata,
             this.scheduleDate,
-            this.traceParent
+            this.traceParent,
+            this.fixtures
         );
     }
 
@@ -234,7 +240,8 @@ public Execution withLabels(List<Label> labels) {
             this.deleted,
             this.metadata,
             this.scheduleDate,
-            this.traceParent
+            this.traceParent,
+            this.fixtures
         );
     }
 
@@ -271,7 +278,8 @@ public Execution withTaskRun(TaskRun taskRun) throws InternalException {
             this.deleted,
             this.metadata,
             this.scheduleDate,
-            this.traceParent
+            this.traceParent,
+            this.fixtures
         );
     }
 
@@ -295,7 +303,8 @@ public Execution childExecution(String childExecutionId, List<TaskRun> taskRunLi
             this.deleted,
             this.metadata,
             this.scheduleDate,
-            this.traceParent
+            this.traceParent,
+            this.fixtures
         );
     }
 
@@ -728,6 +737,16 @@ public FailedExecutionWithLog failedExecutionFromExecutor(Exception e) {
             );
     }
 
+    public Optional<TaskFixture> getFixtureForTaskRun(TaskRun taskRun) {
+        if (this.fixtures == null) {
+            return Optional.empty();
+        }
+
+        return this.fixtures.stream()
+            .filter(fixture -> Objects.equals(fixture.getId(), taskRun.getTaskId()) && Objects.equals(fixture.getValue(), taskRun.getValue()))
+            .findFirst();
+    }
+
     /**
      * Create a new attempt for failed worker execution
      *

core/src/main/java/io/kestra/core/runners/ExecutorService.java

@@ -15,6 +15,7 @@
 import io.kestra.core.queues.QueueFactoryInterface;
 import io.kestra.core.queues.QueueInterface;
 import io.kestra.core.services.*;
+import io.kestra.core.test.flow.TaskFixture;
 import io.kestra.core.storages.StorageContext;
 import io.kestra.core.trace.propagation.RunContextTextMapSetter;
 import io.kestra.core.utils.ListUtils;
@@ -770,7 +771,7 @@ private Executor handleWorkerTask(final Executor executor) throws InternalExcept
         Map<Boolean, List<WorkerTask>> workerTasks = executor.getExecution()
             .getTaskRunList()
             .stream()
-            .filter(taskRun -> taskRun.getState().getCurrent().isCreated())
+            .filter(taskRun -> taskRun.getState().getCurrent().isCreated() && executor.getExecution().getFixtureForTaskRun(taskRun).isEmpty())
             .map(throwFunction(taskRun -> {
                     Task task = executor.getFlow().findTaskByTaskId(taskRun.getTaskId());
                     RunContext runContext = runContextFactory.of(executor.getFlow(), task, executor.getExecution(), taskRun);
@@ -826,7 +827,31 @@ private Executor handleWorkerTask(final Executor executor) throws InternalExcept
             )
             .collect(Collectors.groupingBy(workerTask -> workerTask.getTaskRun().getState().isFailed() || workerTask.getTaskRun().getState().getCurrent() == State.Type.CANCELLED));
 
-        if (workerTasks.isEmpty()) {
+        // mock WorkerTaskResult for mocked execution
+        // submit TaskRun when receiving created, must be done after the state execution store
+        boolean hasMockedWorkerTask = false;
+        record FixtureAndTaskRun(TaskFixture fixture, TaskRun taskRun) {}
+        if (executor.getExecution().getFixtures() != null) {
+            List<WorkerTaskResult> workerTaskResults = executor.getExecution()
+                .getTaskRunList()
+                .stream()
+                .filter(taskRun -> taskRun.getState().getCurrent().isCreated())
+                .flatMap(taskRun -> executor.getExecution().getFixtureForTaskRun(taskRun).stream().map(fixture -> new FixtureAndTaskRun(fixture, taskRun)))
+                .map(fixtureAndTaskRun -> WorkerTaskResult.builder()
+                    .taskRun(fixtureAndTaskRun.taskRun()
+                        .withState(Optional.ofNullable(fixtureAndTaskRun.fixture().getState()).orElse(State.Type.SUCCESS))
+                        .withOutputs(variablesService.of(StorageContext.forTask(fixtureAndTaskRun.taskRun), fixtureAndTaskRun.fixture().getOutputs()))
+                    )
+                    .build()
+                )
+                .toList();
+
+            hasMockedWorkerTask = !workerTaskResults.isEmpty();
+            this.addWorkerTaskResults(executor, executor.getFlow(), workerTaskResults);
+        }
+
+
+        if (workerTasks.isEmpty() || hasMockedWorkerTask) {
             return executor;
         }
 

core/src/main/java/io/kestra/core/runners/RunnerUtils.java

@@ -85,6 +85,10 @@ public Execution runOne(Flow flow, BiFunction<FlowInterface, Execution, Map<Stri
 
         Execution execution = Execution.newExecution(flow, inputs, labels, Optional.empty());
 
+        return runOne(execution, flow, duration);
+    }
+
+    public Execution runOne(Execution execution, Flow flow, Duration duration) throws TimeoutException, QueueException {
         return this.awaitExecution(isTerminatedExecution(execution, flow), throwRunnable(() -> {
             this.executionQueue.emit(execution);
         }), duration);

core/src/main/java/io/kestra/core/test/TestSuite.java

@@ -0,0 +1,91 @@
+package io.kestra.core.test;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+import io.kestra.core.models.DeletedInterface;
+import io.kestra.core.models.HasSource;
+import io.kestra.core.models.HasUID;
+import io.kestra.core.models.TenantInterface;
+import io.kestra.core.test.flow.UnitTest;
+import io.kestra.core.utils.IdUtils;
+import io.micronaut.core.annotation.Introspected;
+import io.swagger.v3.oas.annotations.Hidden;
+import jakarta.validation.constraints.*;
+import lombok.*;
+import lombok.experimental.SuperBuilder;
+
+import java.util.List;
+
+@SuperBuilder(toBuilder = true)
+@Getter
+@NoArgsConstructor
+@AllArgsConstructor
+@Introspected
+@ToString
+@EqualsAndHashCode
+public class TestSuite implements HasUID, TenantInterface, DeletedInterface, HasSource {
+
+    @NotNull
+    @NotBlank
+    @Pattern(regexp = "^[a-zA-Z0-9][a-zA-Z0-9_-]*")
+    private String id;
+
+    @Hidden
+    @Pattern(regexp = "^[a-z0-9][a-z0-9_-]*")
+    private String tenantId;
+
+    private String description;
+
+    @NotNull
+    @Pattern(regexp = "^[a-z0-9][a-z0-9._-]*")
+    @Size(min = 1, max = 150)
+    private String namespace;
+
+    @NotNull
+    @Pattern(regexp = "^[a-zA-Z0-9][a-zA-Z0-9._-]*")
+    @Size(min = 1, max = 100)
+    private String flowId;
+
+    private String source;
+
+    @NotNull
+    @NotEmpty
+    private List<UnitTest> testCases;
+
+    @JsonProperty("deleted")
+    boolean isDeleted = Boolean.FALSE;
+
+    @Builder.Default
+    private Boolean disabled = Boolean.FALSE;
+
+    @Override
+    public String uid() {
+        return IdUtils.fromParts(
+            tenantId,
+            namespace,
+            id
+        );
+    }
+
+    public TestSuite update(final String newSource, final TestSuite newTestSuite) {
+        return new TestSuite(
+            newTestSuite.getId(),
+            this.tenantId,
+            newTestSuite.getDescription(),
+            newTestSuite.getNamespace(),
+            newTestSuite.getFlowId(),
+            newSource,
+            newTestSuite.getTestCases(),
+            newTestSuite.isDeleted(),
+            newTestSuite.getDisabled()
+            );
+    }
+
+    public TestSuite delete() {
+        return this.toBuilder().isDeleted(true).build();
+    }
+
+    @Override
+    public String source() {
+        return this.getSource();
+    }
+}

core/src/main/java/io/kestra/core/test/flow/Assertion.java

@@ -0,0 +1,35 @@
+package io.kestra.core.test.flow;
+
+import io.kestra.core.models.property.Property;
+import jakarta.validation.constraints.NotNull;
+import lombok.Builder;
+import lombok.Getter;
+
+import java.util.List;
+
+@Getter
+@Builder
+public class Assertion {
+    @NotNull
+    private Property<Object> value;
+
+    private String taskId;
+
+    private Property<String> errorMessage;
+
+    private Property<String> description;
+
+    private Property<String> endsWith;
+    private Property<String> startsWith;
+    private Property<String> contains;
+    private Property<String> equalsTo;
+    private Property<String> notEqualsTo;
+    private Property<Double> greaterThan;
+    private Property<Double> greaterThanOrEqualTo;
+    private Property<Double> lessThan;
+    private Property<Double> lesThanOrEqualTo;
+    private Property<List<String>> in;
+    private Property<List<String>> notIn;
+    private Property<Boolean> isNull;
+    private Property<Boolean> isNotNull;
+}

core/src/main/java/io/kestra/core/test/flow/Fixtures.java

@@ -0,0 +1,17 @@
+package io.kestra.core.test.flow;
+
+import lombok.Builder;
+import lombok.Getter;
+
+import java.util.List;
+import java.util.Map;
+
+@Getter
+@Builder
+public class Fixtures {
+    private Map<String, Object> inputs;
+
+    private List<TaskFixture> tasks;
+
+    private TriggerFixture trigger;
+}

core/src/main/java/io/kestra/core/test/flow/TaskFixture.java

@@ -0,0 +1,25 @@
+package io.kestra.core.test.flow;
+
+import io.kestra.core.models.flows.State;
+import io.kestra.core.models.property.Property;
+import jakarta.validation.constraints.NotNull;
+import lombok.Builder;
+import lombok.Getter;
+
+import java.util.Map;
+
+@Getter
+@Builder
+public class TaskFixture {
+    @NotNull
+    private String id;
+
+    private String value;
+
+    @Builder.Default
+    private State.Type state = State.Type.SUCCESS;
+
+    private Map<String, Object> outputs;
+
+    private Property<String> description;
+}

core/src/main/java/io/kestra/core/test/flow/TriggerFixture.java

@@ -0,0 +1,19 @@
+package io.kestra.core.test.flow;
+
+import jakarta.validation.constraints.NotNull;
+import lombok.Builder;
+import lombok.Getter;
+
+import java.util.Map;
+
+@Getter
+@Builder
+public class TriggerFixture {
+    @NotNull
+    private String id;
+
+    @NotNull
+    private String type;
+
+    private Map<String, Object> variables;
+}

core/src/main/java/io/kestra/core/test/flow/UnitTest.java

@@ -0,0 +1,27 @@
+package io.kestra.core.test.flow;
+
+import jakarta.validation.constraints.NotNull;
+import lombok.Builder;
+import lombok.Getter;
+
+import java.util.List;
+
+@Getter
+@Builder
+public class UnitTest {
+    @NotNull
+    private String id;
+
+    @NotNull
+    private String type;
+
+    @Builder.Default
+    private boolean disabled = false;
+
+    private String description;
+
+    private Fixtures fixtures;
+
+    @NotNull
+    private List<Assertion> assertions;
+}