diff --git a/AGENTS.md b/AGENTS.md index 4601ab03..8e4de1c4 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -62,9 +62,11 @@ This is a **multi-module** plugin with 19 submodules: **plugin-script-groovy:** - `io.kestra.plugin.scripts.groovy.Commands` +- `io.kestra.plugin.scripts.groovy.CommandsTrigger` - `io.kestra.plugin.scripts.groovy.Eval` - `io.kestra.plugin.scripts.groovy.FileTransform` - `io.kestra.plugin.scripts.groovy.Script` +- `io.kestra.plugin.scripts.groovy.ScriptTrigger` **plugin-script-jbang:** - `io.kestra.plugin.scripts.jbang.Commands` diff --git a/plugin-script-groovy/src/main/java/io/kestra/plugin/scripts/groovy/AbstractGroovyTrigger.java b/plugin-script-groovy/src/main/java/io/kestra/plugin/scripts/groovy/AbstractGroovyTrigger.java new file mode 100644 index 00000000..e8f02293 --- /dev/null +++ b/plugin-script-groovy/src/main/java/io/kestra/plugin/scripts/groovy/AbstractGroovyTrigger.java @@ -0,0 +1,177 @@ +package io.kestra.plugin.scripts.groovy; + +import java.time.Duration; +import java.time.Instant; +import java.util.Map; +import java.util.Optional; +import java.util.regex.Pattern; + +import io.kestra.core.models.annotations.PluginProperty; +import io.kestra.core.models.conditions.ConditionContext; +import io.kestra.core.models.executions.Execution; +import io.kestra.core.models.property.Property; +import io.kestra.core.models.tasks.RunnableTaskException; +import io.kestra.core.models.tasks.runners.TaskException; +import io.kestra.core.models.triggers.*; +import io.kestra.core.runners.RunContext; +import io.kestra.core.storages.kv.KVValueAndMetadata; +import io.kestra.plugin.scripts.exec.ExitConditionRegex; +import io.kestra.plugin.scripts.exec.scripts.models.ScriptOutput; + +import io.swagger.v3.oas.annotations.media.Schema; +import jakarta.validation.constraints.NotNull; +import lombok.*; +import lombok.experimental.SuperBuilder; + +@SuperBuilder +@ToString +@EqualsAndHashCode +@Getter +@NoArgsConstructor +public abstract class AbstractGroovyTrigger extends AbstractTrigger + implements PollingTriggerInterface, TriggerOutput { + + protected static final String DEFAULT_IMAGE = "groovy"; + private static final Pattern EXIT_CONDITION = Pattern.compile("^exit\\s+(\\d+)$", Pattern.CASE_INSENSITIVE); + + @Schema(title = "Container image", description = "Image used by the underlying Groovy task on each poll. Defaults to groovy.") + @PluginProperty(group = "execution") + @Builder.Default + protected Property containerImage = Property.ofValue(DEFAULT_IMAGE); + + @Schema(title = "Condition to match", description = """ + Evaluated after each run. 'exit N' compares a known process exit code (case insensitive). + Other nonempty values are regexes matched against structured vars emitted using + ::{"outputs":{...}}::. Invalid or excessive regexes fall back to substring matching. + Raw stdout and stderr are not matched. + """) + @NotNull + @PluginProperty(group = "main") + protected Property exitCondition; + + @Schema(title = "Check interval", description = "Interval between fresh executions of the configured script or commands. Defaults to PT60S.") + @PluginProperty(group = "execution") + @Builder.Default + private final Duration interval = Duration.ofSeconds(60); + + @Schema(title = "Edge trigger mode", description = """ + When true (default), the first matching poll emits and consecutive matches are suppressed + until a nonmatching poll. The last match state is stored in the flow namespace KV store, + scoped to this flow and trigger. When false, each fresh matching run emits without accessing + or updating KV state. Re-enabling edge mode resumes the last stored edge-mode state. + This does not provide exactly-once delivery if an evaluation crashes. + """) + @PluginProperty(group = "advanced") + @Builder.Default + protected Property edge = Property.ofValue(true); + + protected abstract ScriptOutput executeTask(RunContext runContext) throws Exception; + + @Override + public Optional evaluate(ConditionContext conditionContext, TriggerContext context) throws Exception { + var runContext = conditionContext.getRunContext(); + var rEdge = runContext.render(this.edge).as(Boolean.class).orElse(true); + var rCondition = runContext.render(this.exitCondition).as(String.class) + .orElseThrow(() -> new IllegalArgumentException("exitCondition is required; set 'exit N' or a regex against vars")); + validateCondition(rCondition); + + Output output; + try { + output = runOnce(runContext, rCondition); + } catch (Exception e) { + runContext.logger().warn("Groovy trigger could not execute its task; check the container image and task runner", e); + return Optional.empty(); + } + + var matched = matchesCondition(output); + if (!rEdge) { + return matched + ? Optional.of(TriggerService.generateExecution(this, conditionContext, context, output)) + : Optional.empty(); + } + + var store = runContext.namespaceKv(context.getNamespace()); + var key = edgeStateKey(context.getFlowId(), context.getTriggerId()); + var lastMatched = store.getValue(key).map(value -> Boolean.parseBoolean(String.valueOf(value.value()))).orElse(false); + if (matched != lastMatched) { + store.put(key, new KVValueAndMetadata(null, matched)); + } + + if (!matched || lastMatched) { + return Optional.empty(); + } + return Optional.of(TriggerService.generateExecution(this, conditionContext, context, output)); + } + + static String edgeStateKey(String flowId, String triggerId) { + return "groovy_edge_" + flowId.length() + "_" + flowId + "_" + triggerId.length() + "_" + triggerId; + } + + private Output runOnce(RunContext runContext, String rCondition) throws Exception { + try { + var taskOutput = executeTask(runContext); + return new Output( + Instant.now(), rCondition, + taskOutput == null ? null : taskOutput.getExitCode(), + taskOutput == null ? null : taskOutput.getVars() + ); + } catch (RunnableTaskException e) { + Integer exitCode = null; + for (var cause = e.getCause(); cause != null; cause = cause.getCause()) { + if (cause instanceof TaskException taskException) { + exitCode = taskException.getExitCode(); + break; + } + } + if (exitCode == null) { + throw e; + } + var vars = e.getOutput() instanceof ScriptOutput taskOutput ? taskOutput.getVars() : null; + return new Output(Instant.now(), rCondition, exitCode, vars); + } + } + + static void validateCondition(String condition) { + if (condition == null || condition.isBlank()) { + throw new IllegalArgumentException("exitCondition must not be empty; set 'exit N' or a regex against vars"); + } + var matcher = EXIT_CONDITION.matcher(condition.trim()); + if (matcher.matches()) { + try { + Integer.parseInt(matcher.group(1)); + } catch (NumberFormatException e) { + throw new IllegalArgumentException("exitCondition exit code must be between 0 and " + Integer.MAX_VALUE, e); + } + } + } + + boolean matchesCondition(Output output) { + if (output == null || output.getCondition() == null || output.getCondition().isBlank()) { + return false; + } + var condition = output.getCondition().trim(); + validateCondition(condition); + var matcher = EXIT_CONDITION.matcher(condition); + if (matcher.matches()) { + return output.getExitCode() != null && output.getExitCode() == Integer.parseInt(matcher.group(1)); + } + return output.getVars() != null && !output.getVars().isEmpty() + && ExitConditionRegex.find(condition, output.getVars().toString()); + } + + @Data + @AllArgsConstructor + public static class Output implements io.kestra.core.models.tasks.Output { + @Schema(title = "Poll timestamp", description = "Time the Groovy task finished for this poll.") + private Instant timestamp; + + @Schema(title = "Rendered condition", description = "Rendered exitCondition for this poll.") + private String condition; + + @Schema(title = "Process exit code", description = "Groovy task exit code, or null when unavailable. Unknown codes never match 'exit N'.") + private Integer exitCode; + + @Schema(title = "Structured script vars", description = "Values emitted by the task using ::{\"outputs\":{...}}::, including values emitted before a process failure.") + private Map vars; + } +} diff --git a/plugin-script-groovy/src/main/java/io/kestra/plugin/scripts/groovy/CommandsTrigger.java b/plugin-script-groovy/src/main/java/io/kestra/plugin/scripts/groovy/CommandsTrigger.java new file mode 100644 index 00000000..0dccc427 --- /dev/null +++ b/plugin-script-groovy/src/main/java/io/kestra/plugin/scripts/groovy/CommandsTrigger.java @@ -0,0 +1,72 @@ +package io.kestra.plugin.scripts.groovy; + +import java.util.List; + +import io.kestra.core.models.annotations.Example; +import io.kestra.core.models.annotations.Plugin; +import io.kestra.core.models.annotations.PluginProperty; +import io.kestra.core.models.property.Property; +import io.kestra.core.runners.RunContext; +import io.kestra.plugin.scripts.exec.TriggerRunContext; +import io.kestra.plugin.scripts.exec.scripts.models.ScriptOutput; + +import io.swagger.v3.oas.annotations.media.Schema; +import jakarta.validation.constraints.NotNull; +import lombok.*; +import lombok.experimental.SuperBuilder; + +@SuperBuilder +@ToString +@EqualsAndHashCode +@Getter +@NoArgsConstructor +@Schema( + title = "Trigger a flow when Groovy commands match a condition", + description = "Runs its own Groovy commands on an interval through the Commands task and starts a flow when exitCondition matches. Defaults to the groovy container image and a 60-second interval." +) +@Plugin( + examples = { + @Example(title = "Trigger when a Groovy command exits with code 1.", full = true, code = """ + id: groovy_commands_trigger + namespace: company.team + + tasks: + - id: log + type: io.kestra.plugin.core.log.Log + message: "Triggered with exitCode={{ trigger.exitCode }}" + + triggers: + - id: poll + type: io.kestra.plugin.scripts.groovy.CommandsTrigger + interval: PT60S + exitCondition: "exit 1" + commands: + - groovy -e "System.exit(1)" + """) + } +) +public class CommandsTrigger extends AbstractGroovyTrigger { + + @Schema( + title = "Groovy commands", + description = "Executed on each poll with the same command rendering semantics as the Groovy Commands task, including runner variables such as workingDir. beforeCommands, env and taskRunner are not exposed by this trigger. Supply dependencies and configuration in the commands (for example @Grab in Groovy code, which may require network access) or container image." + ) + @NotNull + @PluginProperty(group = "main") + protected Property> commands; + + @Override + protected ScriptOutput executeTask(RunContext runContext) throws Exception { + if (this.commands == null) { + throw new IllegalArgumentException("commands is required; supply the Groovy commands to run on each poll"); + } + var rImage = runContext.render(this.containerImage).as(String.class).orElse(DEFAULT_IMAGE); + var task = Commands.builder() + .id(this.getId()) + .type(Commands.class.getName()) + .containerImage(Property.ofValue(rImage)) + .commands(this.commands) + .build(); + return task.run(TriggerRunContext.forEmbeddedTask(runContext, task)); + } +} diff --git a/plugin-script-groovy/src/main/java/io/kestra/plugin/scripts/groovy/ScriptTrigger.java b/plugin-script-groovy/src/main/java/io/kestra/plugin/scripts/groovy/ScriptTrigger.java new file mode 100644 index 00000000..cd2d220f --- /dev/null +++ b/plugin-script-groovy/src/main/java/io/kestra/plugin/scripts/groovy/ScriptTrigger.java @@ -0,0 +1,70 @@ +package io.kestra.plugin.scripts.groovy; + +import io.kestra.core.models.annotations.Example; +import io.kestra.core.models.annotations.Plugin; +import io.kestra.core.models.annotations.PluginProperty; +import io.kestra.core.models.property.Property; +import io.kestra.core.runners.RunContext; +import io.kestra.plugin.scripts.exec.TriggerRunContext; +import io.kestra.plugin.scripts.exec.scripts.models.ScriptOutput; + +import io.swagger.v3.oas.annotations.media.Schema; +import jakarta.validation.constraints.NotNull; +import lombok.*; +import lombok.experimental.SuperBuilder; + +@SuperBuilder +@ToString +@EqualsAndHashCode +@Getter +@NoArgsConstructor +@Schema( + title = "Trigger a flow when a Groovy script matches a condition", + description = "Runs its own Groovy script on an interval through the Script task and starts a flow when exitCondition matches. Defaults to the groovy container image and a 60-second interval." +) +@Plugin( + examples = { + @Example(title = "Trigger when structured vars report ready.", full = true, code = """ + id: groovy_script_trigger + namespace: company.team + + tasks: + - id: log + type: io.kestra.plugin.core.log.Log + message: "Triggered with exitCode={{ trigger.exitCode }} (vars={{ trigger.vars }})" + + triggers: + - id: poll + type: io.kestra.plugin.scripts.groovy.ScriptTrigger + interval: PT60S + exitCondition: "ready" + script: | + println '::{"outputs":{"status":"ready"}}::' + """) + } +) +public class ScriptTrigger extends AbstractGroovyTrigger { + + @Schema( + title = "Groovy script", + description = "Executed on each poll with the same script rendering semantics as the Groovy Script task, including runner variables such as workingDir. beforeCommands, env and taskRunner are not exposed by this trigger. Supply dependencies and configuration in the script (for example @Grab, which may require network access) or container image." + ) + @NotNull + @PluginProperty(group = "main") + protected Property script; + + @Override + protected ScriptOutput executeTask(RunContext runContext) throws Exception { + if (this.script == null) { + throw new IllegalArgumentException("script is required; supply the Groovy script to run on each poll"); + } + var rImage = runContext.render(this.containerImage).as(String.class).orElse(DEFAULT_IMAGE); + var task = Script.builder() + .id(this.getId()) + .type(Script.class.getName()) + .containerImage(Property.ofValue(rImage)) + .script(this.script) + .build(); + return task.run(TriggerRunContext.forEmbeddedTask(runContext, task)); + } +} diff --git a/plugin-script-groovy/src/main/resources/doc/io.kestra.plugin.scripts.groovy.md b/plugin-script-groovy/src/main/resources/doc/io.kestra.plugin.scripts.groovy.md index 62dcb585..99451466 100644 --- a/plugin-script-groovy/src/main/resources/doc/io.kestra.plugin.scripts.groovy.md +++ b/plugin-script-groovy/src/main/resources/doc/io.kestra.plugin.scripts.groovy.md @@ -14,4 +14,16 @@ Add Maven dependencies inline using Grape annotations: `@Grab('group:artifact:ve ## Container user -The working directory Kestra mounts is not necessarily owned by the image's default user (`groovy`, uid 1000, on `groovy:jdk21`), so a non-root process cannot always write the files that `outputFiles` collects. On the Docker task runner, `Commands` therefore runs the container as `root` unless you set `taskRunner.user` explicitly; set it if you need the container to keep its own default user, and make sure that user can write to the working directory. `Script` always runs as `root` on Docker and ignores `taskRunner.user`. +The working directory Kestra mounts is not necessarily owned by the image's default user (`groovy`, uid 1000, on `groovy:jdk21`), so a non-root process cannot always write the files that `outputFiles` collects. On the Docker task runner, `Commands` therefore runs the container as `root` unless you set `taskRunner.user` explicitly; set it if you need the container to keep its own default user, and make sure that user can write to the working directory. `Script` uses the same root default and respects an explicit `taskRunner.user`. + +## Polling triggers + +`ScriptTrigger` runs its own inline Groovy script on an interval; `CommandsTrigger` runs its own list of commands. Both delegate to the existing task runner tasks, with the `groovy` image and a 60-second interval by default. + +The triggers expose `script` or `commands`, `containerImage`, `interval`, `exitCondition`, and `edge`; they do not expose `beforeCommands`, `env`, or `taskRunner`. Supply dependencies and configuration through the script/commands or the container image. Groovy `@Grab` is one way to load dependencies at runtime and may require network access; it does not guarantee offline availability. The underlying task runner renders script/command expressions such as `{{ workingDir }}` with its container working directory. This does not expose every option of the underlying tasks. + +Set the required `exitCondition` to `exit N` to match a known process exit code, including nonzero exits. Alternatively, use a regex against structured vars emitted with `::{"outputs":{...}}::`. Invalid or excessive regexes fall back to substring matching. Raw stdout/stderr and comparison expressions such as `exitCode != 0` are not condition inputs. + +Downstream tasks can read `trigger.timestamp`, `trigger.condition`, `trigger.exitCode`, and `trigger.vars`. Vars emitted before a process failure are preserved. Container startup, image, rendering, and other setup failures cannot satisfy a process exit condition. + +With `edge: true` (the default), the first matching evaluation emits; further matches are suppressed until a nonmatch. The previous match is stored in the namespace KV store under `groovy_edge____`, so it survives recreation of the trigger object. Only edge-mode state transitions write the key. With `edge: false`, evaluations do not access or update KV state; re-enabling edge mode resumes the last persisted edge-mode state. Setup/rendering failures preserve the previous state. Deleting that key resets the edge state. With `edge: false`, each evaluation runs the script/commands again and may emit. A fresh process run is a new observation; this is not an event listener or an exactly-once delivery guarantee across crashes. diff --git a/plugin-script-groovy/src/test/java/io/kestra/plugin/scripts/groovy/GroovyTriggerConditionTest.java b/plugin-script-groovy/src/test/java/io/kestra/plugin/scripts/groovy/GroovyTriggerConditionTest.java new file mode 100644 index 00000000..4a24ff17 --- /dev/null +++ b/plugin-script-groovy/src/test/java/io/kestra/plugin/scripts/groovy/GroovyTriggerConditionTest.java @@ -0,0 +1,63 @@ +package io.kestra.plugin.scripts.groovy; + +import java.time.Instant; +import java.util.Map; +import java.util.stream.Stream; + +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 org.junit.jupiter.params.provider.ValueSource; + +import static org.junit.jupiter.api.Assertions.*; + +class GroovyTriggerConditionTest { + private AbstractGroovyTrigger trigger(boolean commands) { + return commands ? CommandsTrigger.builder().build() : ScriptTrigger.builder().build(); + } + + static Stream conditions() { + return Stream.of(false, true).flatMap( + commands -> Stream.of( + Arguments.of(commands, "exit 0", 0, null, true), + Arguments.of(commands, " EXIT 1 ", 1, null, true), + Arguments.of(commands, "exit 42", 42, Map.of(), true), + Arguments.of(commands, "exit 0", 1, null, false), + Arguments.of(commands, "exit 0", null, null, false), + Arguments.of(commands, null, 0, Map.of("status", "ready"), false), + Arguments.of(commands, "", 0, Map.of("status", "ready"), false), + Arguments.of(commands, " ", 0, Map.of("status", "ready"), false), + Arguments.of(commands, "ready", 0, null, false), + Arguments.of(commands, "ready", 0, Map.of(), false), + Arguments.of(commands, "status=r.*", 0, Map.of("status", "ready"), true), + Arguments.of(commands, "status=waiting", 0, Map.of("status", "ready"), false), + Arguments.of(commands, "[", 0, Map.of("status", "[ready"), true), + Arguments.of(commands, "[", 0, Map.of("status", "ready"), false) + ) + ); + } + + @ParameterizedTest + @MethodSource("conditions") + void matchesSupportedConditions(boolean commands, String condition, Integer code, Map vars, boolean expected) { + assertEquals(expected, trigger(commands).matchesCondition(new AbstractGroovyTrigger.Output(Instant.now(), condition, code, vars))); + } + + @ParameterizedTest + @ValueSource(booleans = { false, true }) + void nullOutputAndOverflow(boolean commands) { + assertFalse(trigger(commands).matchesCondition(null)); + assertThrows( + IllegalArgumentException.class, () -> trigger(commands).matchesCondition( + new AbstractGroovyTrigger.Output(Instant.now(), "exit 2147483648", 0, null) + ) + ); + } + + @Test + void scopedKeysDoNotCollide() { + assertNotEquals(AbstractGroovyTrigger.edgeStateKey("ab", "c"), AbstractGroovyTrigger.edgeStateKey("a", "bc")); + assertNotEquals(AbstractGroovyTrigger.edgeStateKey("a_b", "c"), AbstractGroovyTrigger.edgeStateKey("a", "b_c")); + } +} diff --git a/plugin-script-groovy/src/test/java/io/kestra/plugin/scripts/groovy/GroovyTriggerTest.java b/plugin-script-groovy/src/test/java/io/kestra/plugin/scripts/groovy/GroovyTriggerTest.java new file mode 100644 index 00000000..acdc7134 --- /dev/null +++ b/plugin-script-groovy/src/test/java/io/kestra/plugin/scripts/groovy/GroovyTriggerTest.java @@ -0,0 +1,553 @@ +package io.kestra.plugin.scripts.groovy; + +import java.io.IOException; +import java.nio.file.Path; +import java.time.Duration; +import java.time.ZonedDateTime; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.UUID; +import java.util.stream.Stream; + +import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.MethodSource; +import org.junit.jupiter.params.provider.ValueSource; + +import io.kestra.core.exceptions.IllegalVariableEvaluationException; +import io.kestra.core.junit.annotations.KestraTest; +import io.kestra.core.models.property.Property; +import io.kestra.core.models.tasks.RunnableTaskException; +import io.kestra.core.models.triggers.Trigger; +import io.kestra.core.runners.RunContextFactory; +import io.kestra.core.serializers.JacksonMapper; +import io.kestra.core.storages.kv.KVStore; +import io.kestra.core.storages.kv.KVValue; +import io.kestra.core.storages.kv.KVValueAndMetadata; +import io.kestra.core.utils.TestsUtils; +import io.kestra.plugin.scripts.exec.scripts.models.ScriptOutput; + +import jakarta.inject.Inject; +import jakarta.validation.Validator; + +import static org.junit.jupiter.api.Assertions.*; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.*; + +@KestraTest +class GroovyTriggerTest { + @Inject + RunContextFactory runContextFactory; + + @Inject + Validator validator; + + private AbstractGroovyTrigger trigger(boolean commands, String id, String script, String condition, boolean edge) { + if (commands) { + return CommandsTrigger.builder() + .id(id).type(CommandsTrigger.class.getName()) + .containerImage(Property.ofValue("groovy:jdk21")) + .commands(Property.ofValue(List.of("groovy -e '" + script.replace("'", "'\"'\"'") + "'"))) + .exitCondition(Property.ofValue(condition)).edge(Property.ofValue(edge)).build(); + } + return ScriptTrigger.builder() + .id(id).type(ScriptTrigger.class.getName()) + .containerImage(Property.ofValue("groovy:jdk21")) + .script(Property.ofValue(script)) + .exitCondition(Property.ofValue(condition)).edge(Property.ofValue(edge)).build(); + } + + private String id() { + return "poll-" + UUID.randomUUID(); + } + + @ParameterizedTest + @ValueSource(booleans = { false, true }) + void successfulExitCreatesExecutionFromTriggerContext(boolean commands) throws Exception { + var trigger = trigger(commands, id(), "println 'ok'", "exit 0", true); + var context = TestsUtils.mockTrigger(runContextFactory, trigger); + assertFalse(context.getKey().getRunContext().getVariables().containsKey("task")); + assertFalse(context.getKey().getRunContext().getVariables().containsKey("execution")); + var execution = trigger.evaluate(context.getKey(), context.getValue()).orElseThrow(); + var variables = execution.getTrigger().getVariables(); + assertEquals("exit 0", variables.get("condition")); + assertEquals(0, variables.get("exitCode")); + assertNotNull(variables.get("timestamp")); + assertEquals(trigger.getId(), execution.getTrigger().getId()); + } + + @ParameterizedTest + @ValueSource(booleans = { false, true }) + void failingProcessCanTriggerAndPreservesVars(boolean commands) throws Exception { + var trigger = trigger(commands, id(), "println '::{\"outputs\":{\"status\":\"ready\",\"count\":3}}::'; System.exit(1)", "exit 1", true); + var context = TestsUtils.mockTrigger(runContextFactory, trigger); + var variables = trigger.evaluate(context.getKey(), context.getValue()).orElseThrow().getTrigger().getVariables(); + assertEquals(1, variables.get("exitCode")); + assertEquals("exit 1", variables.get("condition")); + assertEquals("ready", ((Map) variables.get("vars")).get("status")); + assertEquals(3, ((Number) ((Map) variables.get("vars")).get("count")).intValue()); + } + + @ParameterizedTest + @ValueSource(booleans = { false, true }) + void exitMismatchDoesNotCreateExecution(boolean commands) throws Exception { + var trigger = trigger(commands, id(), "System.exit(1)", "exit 0", true); + var context = TestsUtils.mockTrigger(runContextFactory, trigger); + assertTrue(trigger.evaluate(context.getKey(), context.getValue()).isEmpty()); + } + + @ParameterizedTest + @ValueSource(booleans = { false, true }) + void structuredVarsMatchAndNonmatch(boolean commands) throws Exception { + var trigger = trigger(commands, id(), "println '::{\"outputs\":{\"status\":\"ready\"}}::'", "status=ready", true); + var context = TestsUtils.mockTrigger(runContextFactory, trigger); + var variables = trigger.evaluate(context.getKey(), context.getValue()).orElseThrow().getTrigger().getVariables(); + assertEquals(Map.of("status", "ready"), variables.get("vars")); + var nonmatch = trigger(commands, id(), "println '::{\"outputs\":{\"status\":\"waiting\"}}::'", "status=ready", true); + assertTrue(nonmatch.evaluate(context.getKey(), context.getValue().toBuilder().triggerId(nonmatch.getId()).build()).isEmpty()); + } + + @ParameterizedTest + @ValueSource(booleans = { false, true }) + void edgeStateSurvivesNewTriggerInstancesAndResetsOnNonmatch(boolean commands) throws Exception { + var id = id(); + var first = trigger(commands, id, "println 'ok'", "exit 0", true); + var context = TestsUtils.mockTrigger(runContextFactory, first); + assertTrue(first.evaluate(context.getKey(), context.getValue()).isPresent()); + var rehydrated = trigger(commands, id, "println 'ok'", "exit 0", true); + assertTrue(rehydrated.evaluate(context.getKey(), context.getValue()).isEmpty()); + var nonmatch = trigger(commands, id, "System.exit(1)", "exit 0", true); + assertTrue(nonmatch.evaluate(context.getKey(), context.getValue()).isEmpty()); + assertTrue(rehydrated.evaluate(context.getKey(), context.getValue()).isPresent()); + assertTrue(rehydrated.evaluate(context.getKey(), context.getValue()).isEmpty()); + } + + @ParameterizedTest + @ValueSource(booleans = { false, true }) + void edgeFalseEmitsOnEachFreshRun(boolean commands) throws Exception { + var trigger = trigger(commands, id(), "println 'ok'", "exit 0", false); + var context = TestsUtils.mockTrigger(runContextFactory, trigger); + assertTrue(trigger.evaluate(context.getKey(), context.getValue()).isPresent()); + assertTrue(trigger.evaluate(context.getKey(), context.getValue()).isPresent()); + } + + @ParameterizedTest + @ValueSource(booleans = { false, true }) + void rendersPropertiesFromSchedulerVariables(boolean commands) throws Exception { + var id = id(); + AbstractGroovyTrigger trigger; + if (commands) { + trigger = CommandsTrigger.builder().id(id).type(CommandsTrigger.class.getName()) + .containerImage(Property.ofExpression("{{ 'groovy:jdk21' }}")) + .commands(Property.ofExpression("{{ [\"groovy -e 'println \\\"\" ~ flow.id ~ \"\\\"'\"] }}")) + .exitCondition(Property.ofExpression("exit {{ 0 }}")) + .edge(Property.ofExpression("{{ false }}")).build(); + } else { + trigger = ScriptTrigger.builder().id(id).type(ScriptTrigger.class.getName()) + .containerImage(Property.ofExpression("{{ 'groovy:jdk21' }}")) + .script(Property.ofExpression("println '{{ flow.id }}'")) + .exitCondition(Property.ofExpression("exit {{ 0 }}")) + .edge(Property.ofExpression("{{ false }}")).build(); + } + var context = TestsUtils.mockTrigger(runContextFactory, trigger); + assertEquals("exit 0", trigger.evaluate(context.getKey(), context.getValue()).orElseThrow().getTrigger().getVariables().get("condition")); + assertTrue(trigger.evaluate(context.getKey(), context.getValue()).isPresent()); + } + + @ParameterizedTest + @ValueSource(booleans = { false, true }) + void invalidImageCannotMatchSuccessfulExit(boolean commands) throws Exception { + var id = id(); + AbstractGroovyTrigger trigger = commands + ? CommandsTrigger.builder().id(id).type(CommandsTrigger.class.getName()) + .containerImage(Property.ofValue("INVALID IMAGE")) + .commands(Property.ofValue(List.of("groovy -e 'println 1'"))).exitCondition(Property.ofValue("exit 0")).build() + : ScriptTrigger.builder().id(id).type(ScriptTrigger.class.getName()) + .containerImage(Property.ofValue("INVALID IMAGE")) + .script(Property.ofValue("println 1")).exitCondition(Property.ofValue("exit 0")).build(); + var context = TestsUtils.mockTrigger(runContextFactory, trigger); + assertTrue(trigger.evaluate(context.getKey(), context.getValue()).isEmpty()); + } + + @ParameterizedTest + @ValueSource(booleans = { false, true }) + void defaultsAndRequiredInputValidation(boolean commands) throws Exception { + AbstractGroovyTrigger trigger = commands ? CommandsTrigger.builder().build() : ScriptTrigger.builder().build(); + assertEquals(Duration.ofSeconds(60), trigger.getInterval()); + var runContext = runContextFactory.of(); + assertEquals("groovy", runContext.render(trigger.getContainerImage()).as(String.class).orElseThrow()); + assertTrue(runContext.render(trigger.getEdge()).as(Boolean.class).orElseThrow()); + var missingFields = validator.validate(trigger).stream().map(violation -> violation.getPropertyPath().toString()).toList(); + assertTrue(missingFields.contains("exitCondition")); + assertTrue(missingFields.contains(commands ? "commands" : "script")); + } + + @ParameterizedTest + @ValueSource(booleans = { false, true }) + void rejectsEmptyAndOverflowingConditionsBeforeRunning(boolean commands) { + for (var condition : List.of(" ", "exit 999999999999999999999999")) { + var trigger = trigger(commands, id(), "println 'ok'", condition, true); + var context = TestsUtils.mockTrigger(runContextFactory, trigger); + assertThrows(IllegalArgumentException.class, () -> trigger.evaluate(context.getKey(), context.getValue())); + } + } + + @ParameterizedTest + @ValueSource(booleans = { false, true }) + void unknownExitAndSetupExceptionsDoNotMatch(boolean commands) throws Exception { + var trigger = spy(trigger(commands, id(), "println 'ok'", "exit 0", true)); + var context = TestsUtils.mockTrigger(runContextFactory, trigger); + doReturn(null).when(trigger).executeTask(any()); + assertTrue(trigger.evaluate(context.getKey(), context.getValue()).isEmpty()); + var misleadingOutput = ScriptOutput.builder().exitCode(0).build(); + doThrow(new RunnableTaskException("setup failed", misleadingOutput)) + .when(trigger).executeTask(any()); + assertTrue(trigger.evaluate(context.getKey(), context.getValue()).isEmpty()); + doThrow(new IllegalStateException("Docker unavailable")) + .when(trigger).executeTask(any()); + assertTrue(trigger.evaluate(context.getKey(), context.getValue()).isEmpty()); + } + + @ParameterizedTest + @ValueSource(strings = { "script", "list", "expression" }) + void workingDirRendersInRunnerContext(String input) throws Exception { + String script = """ + def dir = new File('{{ workingDir }}') + assert dir.isAbsolute() && dir.isDirectory() + def probe = new File(dir, 'probe.txt') + probe.text = 'usable' + assert probe.text == 'usable' + println '::{"outputs":{"path":"{{ workingDir }}","flow":"{{ flow.id }}"}}::' + """; + AbstractGroovyTrigger trigger; + if (input.equals("script")) { + trigger = ScriptTrigger.builder().id(id()).type(ScriptTrigger.class.getName()) + .containerImage(Property.ofValue("groovy:jdk21")) + .script(Property.ofExpression(script)).exitCondition(Property.ofValue("exit 0")).build(); + } else { + String command = "groovy -e '" + script.strip().replace("\n", "; ").replace("'", "'\"'\"'") + "'"; + Property> commands = input.equals("list") + ? TestsUtils.propertyFromList(List.of(command)) + : Property.ofExpression( + "{{ [" + JacksonMapper.ofJson().writeValueAsString(command) + .replace("{{ workingDir }}", "\" ~ workingDir ~ \"") + .replace("{{ flow.id }}", "\" ~ flow.id ~ \"") + "] }}" + ); + // Exercise the same serializer/deserializer used for YAML flow properties. + commands = JacksonMapper.ofYaml().readValue( + JacksonMapper.ofYaml().writeValueAsString(commands), + JacksonMapper.ofYaml().getTypeFactory().constructParametricType(Property.class, List.class) + ); + trigger = CommandsTrigger.builder().id(id()).type(CommandsTrigger.class.getName()) + .containerImage(Property.ofValue("groovy:jdk21")) + .commands(commands).exitCondition(Property.ofValue("exit 0")).build(); + } + var context = TestsUtils.mockTrigger(runContextFactory, trigger); + var runContext = context.getKey().getRunContext(); + for (String absent : List.of("workingDir", "task", "execution")) { + assertFalse(runContext.getVariables().containsKey(absent), absent); + } + var variables = trigger.evaluate(context.getKey(), context.getValue()).orElseThrow().getTrigger().getVariables(); + var vars = (Map) variables.get("vars"); + String path = (String) vars.get("path"); + assertNotNull(path); + assertFalse(path.isBlank()); + assertTrue(Path.of(path).isAbsolute()); + assertFalse(path.contains("{{")); + assertEquals(((Map) runContext.getVariables().get("flow")).get("id"), vars.get("flow")); + } + + @ParameterizedTest + @ValueSource(booleans = { false, true }) + void writesOnlyTransitionsWithEdgeEnabled(boolean commands) throws Exception { + var original = trigger(commands, id(), "unused", "exit 0", true); + var context = TestsUtils.mockTrigger(runContextFactory, original); + var runContext = spy(context.getKey().getRunContext()); + KVStore store = spy(runContext.namespaceKv(context.getValue().getNamespace())); + doReturn(store).when(runContext).namespaceKv(context.getValue().getNamespace()); + var conditionContext = context.getKey().withRunContext(runContext); + boolean[] matched = { false, true, true, false, false, true, true, true }; + boolean[] emits = { false, true, false, false, false, true, false, false }; + int[] writes = { 0, 1, 1, 2, 2, 3, 3, 3 }; + for (int i = 0; i < matched.length; i++) { + var fresh = spy(trigger(commands, original.getId(), "unused", "exit 0", true)); + doReturn(ScriptOutput.builder().exitCode(matched[i] ? 0 : 1).build()).when(fresh).executeTask(any()); + assertEquals(emits[i], fresh.evaluate(conditionContext, context.getValue()).isPresent(), "poll " + i); + verify(store, times(writes[i])).put(anyString(), any(KVValueAndMetadata.class)); + } + } + + @ParameterizedTest + @ValueSource(booleans = { false, true }) + void edgeDisabledDoesNotAccessKv(boolean commands) throws Exception { + var trigger = spy(trigger(commands, id(), "unused", "exit 0", false)); + var context = TestsUtils.mockTrigger(runContextFactory, trigger); + var runContext = spy(context.getKey().getRunContext()); + doThrow(new IllegalStateException("KV unavailable")).when(runContext).namespaceKv(anyString()); + for (int exitCode : new int[] { 0, 0, 1 }) { + doReturn(ScriptOutput.builder().exitCode(exitCode).build()).when(trigger).executeTask(any()); + assertEquals(exitCode == 0, trigger.evaluate(context.getKey().withRunContext(runContext), context.getValue()).isPresent()); + } + verify(runContext, never()).namespaceKv(anyString()); + } + + @ParameterizedTest + @ValueSource(strings = { "booleanTrue", "booleanFalse", "true", "TRUE", "false", "missing", "null", "unexpected" }) + void persistedBooleanAndStringStateControlsEdges(String state) throws Exception { + Object stored = switch (state) { + case "booleanTrue" -> Boolean.TRUE; + case "booleanFalse" -> Boolean.FALSE; + case "missing", "null" -> null; + default -> state; + }; + var trigger = spy(trigger(false, id(), "unused", "exit 0", true)); + var context = TestsUtils.mockTrigger(runContextFactory, trigger); + var runContext = spy(context.getKey().getRunContext()); + var store = mock(KVStore.class); + doReturn(store).when(runContext).namespaceKv(anyString()); + when(store.getValue(anyString())).thenReturn( + state.equals("missing") + ? Optional.empty() + : Optional.of(new KVValue(stored)) + ); + doReturn(ScriptOutput.builder().exitCode(0).build()).when(trigger).executeTask(any()); + boolean previouslyMatched = Boolean.TRUE.equals(stored) || "true".equals(stored) || "TRUE".equals(stored); + assertEquals(!previouslyMatched, trigger.evaluate(context.getKey().withRunContext(runContext), context.getValue()).isPresent()); + verify(store, times(previouslyMatched ? 0 : 1)).put(anyString(), any(KVValueAndMetadata.class)); + } + + @ParameterizedTest + @MethodSource("priorEdgeStates") + void disabledObservationsPreserveStateWhenEdgeModeResumes(Boolean prior) throws Exception { + var original = trigger(false, id(), "unused", "exit 0", true); + var context = TestsUtils.mockTrigger(runContextFactory, original); + var store = spy(context.getKey().getRunContext().namespaceKv(context.getValue().getNamespace())); + var key = AbstractGroovyTrigger.edgeStateKey(context.getValue().getFlowId(), context.getValue().getTriggerId()); + if (prior != null) { + store.put(key, new KVValueAndMetadata(null, prior)); + } + var runContext = spy(context.getKey().getRunContext()); + doReturn(store).when(runContext).namespaceKv(anyString()); + var conditionContext = context.getKey().withRunContext(runContext); + var disabled = spy(trigger(false, original.getId(), "unused", "exit 0", false)); + doReturn(ScriptOutput.builder().exitCode(Boolean.TRUE.equals(prior) ? 1 : 0).build()).when(disabled).executeTask(any()); + assertEquals(!Boolean.TRUE.equals(prior), disabled.evaluate(conditionContext, context.getValue()).isPresent()); + assertEquals(prior, store.getValue(key).map(value -> (Boolean) value.value()).orElse(null)); + var enabled = spy(trigger(false, original.getId(), "unused", "exit 0", true)); + doReturn(ScriptOutput.builder().exitCode(0).build()).when(enabled).executeTask(any()); + assertEquals(!Boolean.TRUE.equals(prior), enabled.evaluate(conditionContext, context.getValue()).isPresent()); + doReturn(ScriptOutput.builder().exitCode(1).build()).when(enabled).executeTask(any()); + assertTrue(enabled.evaluate(conditionContext, context.getValue()).isEmpty()); + doReturn(ScriptOutput.builder().exitCode(0).build()).when(enabled).executeTask(any()); + assertTrue(enabled.evaluate(conditionContext, context.getValue()).isPresent()); + } + + static Stream priorEdgeStates() { + return Stream.of(true, false, null); + } + + @Test + void stringTrueIsClearedByNonmatchingPoll() throws Exception { + var trigger = spy(trigger(false, id(), "unused", "exit 0", true)); + var context = TestsUtils.mockTrigger(runContextFactory, trigger); + var store = context.getKey().getRunContext().namespaceKv(context.getValue().getNamespace()); + var key = AbstractGroovyTrigger.edgeStateKey(context.getValue().getFlowId(), context.getValue().getTriggerId()); + store.put(key, new KVValueAndMetadata(null, "true")); + doReturn(ScriptOutput.builder().exitCode(1).build()).when(trigger).executeTask(any()); + assertTrue(trigger.evaluate(context.getKey(), context.getValue()).isEmpty()); + assertEquals(Boolean.FALSE, store.getValue(key).orElseThrow().value()); + doReturn(ScriptOutput.builder().exitCode(0).build()).when(trigger).executeTask(any()); + assertTrue(trigger.evaluate(context.getKey(), context.getValue()).isPresent()); + } + + @ParameterizedTest + @ValueSource(booleans = { false, true }) + void missingAndInvalidInputsPreserveMatchingEdge(boolean commands) throws Exception { + var original = trigger(commands, id(), "println 'ok'", "exit 0", true); + var context = TestsUtils.mockTrigger(runContextFactory, original); + var store = context.getKey().getRunContext().namespaceKv(context.getValue().getNamespace()); + var key = AbstractGroovyTrigger.edgeStateKey(context.getValue().getFlowId(), context.getValue().getTriggerId()); + store.put(key, new KVValueAndMetadata(null, true)); + AbstractGroovyTrigger missing = commands + ? CommandsTrigger.builder().id(original.getId()).type(CommandsTrigger.class.getName()) + .exitCondition(Property.ofValue("exit 0")).build() + : ScriptTrigger.builder().id(original.getId()).type(ScriptTrigger.class.getName()) + .exitCondition(Property.ofValue("exit 0")).build(); + assertTrue(missing.evaluate(context.getKey(), context.getValue()).isEmpty()); + assertEquals(true, store.getValue(key).orElseThrow().value()); + var invalid = commands + ? CommandsTrigger.builder().id(original.getId()).type(CommandsTrigger.class.getName()) + .commands(TestsUtils.propertyFromList(List.of("echo {{ absent }}"))) + .exitCondition(Property.ofValue("exit 0")).build() + : ScriptTrigger.builder().id(original.getId()).type(ScriptTrigger.class.getName()) + .script(Property.ofExpression("println '{{ absent }}'")) + .exitCondition(Property.ofValue("exit 0")).build(); + assertTrue(invalid.evaluate(context.getKey(), context.getValue()).isEmpty()); + assertEquals(true, store.getValue(key).orElseThrow().value()); + } + + @ParameterizedTest + @ValueSource(booleans = { false, true }) + void missingAndInvalidInputsCannotEmitWithoutEdgeSuppression(boolean commands) throws Exception { + var original = trigger(commands, id(), "unused", "exit 0", false); + var context = TestsUtils.mockTrigger(runContextFactory, original); + var store = context.getKey().getRunContext().namespaceKv(context.getValue().getNamespace()); + var key = AbstractGroovyTrigger.edgeStateKey(context.getValue().getFlowId(), context.getValue().getTriggerId()); + assertTrue(store.getValue(key).isEmpty()); + AbstractGroovyTrigger missing = commands + ? CommandsTrigger.builder().id(original.getId()).type(CommandsTrigger.class.getName()) + .exitCondition(Property.ofValue("exit 0")).edge(Property.ofValue(false)).build() + : ScriptTrigger.builder().id(original.getId()).type(ScriptTrigger.class.getName()) + .exitCondition(Property.ofValue("exit 0")).edge(Property.ofValue(false)).build(); + assertTrue(missing.evaluate(context.getKey(), context.getValue()).isEmpty()); + assertTrue(store.getValue(key).isEmpty()); + AbstractGroovyTrigger invalid = commands + ? CommandsTrigger.builder().id(original.getId()).type(CommandsTrigger.class.getName()) + .commands(TestsUtils.propertyFromList(List.of("echo {{ absent }}"))) + .exitCondition(Property.ofValue("exit 0")).edge(Property.ofValue(false)).build() + : ScriptTrigger.builder().id(original.getId()).type(ScriptTrigger.class.getName()) + .script(Property.ofExpression("println '{{ absent }}'")) + .exitCondition(Property.ofValue("exit 0")).edge(Property.ofValue(false)).build(); + assertTrue(invalid.evaluate(context.getKey(), context.getValue()).isEmpty()); + assertTrue(store.getValue(key).isEmpty()); + } + + @Test + void persistedStateIsIsolatedByFlowTriggerAndNamespaceAndSurvivesInfrastructureFailure() throws Exception { + var trigger = spy(trigger(false, id(), "unused", "exit 0", true)); + var context = TestsUtils.mockTrigger(runContextFactory, trigger); + var base = context.getValue(); + var scopes = List.of( + base, + Trigger.builder().tenantId(base.getTenantId()) + .namespace(base.getNamespace()).flowId(base.getFlowId() + "-other").triggerId(base.getTriggerId()) + .date(ZonedDateTime.now()).build(), + Trigger.builder().tenantId(base.getTenantId()) + .namespace(base.getNamespace()).flowId(base.getFlowId()).triggerId(base.getTriggerId() + "-other") + .date(ZonedDateTime.now()).build(), + Trigger.builder().tenantId(base.getTenantId()) + .namespace(base.getNamespace() + ".other").flowId(base.getFlowId()).triggerId(base.getTriggerId()) + .date(ZonedDateTime.now()).build() + ); + var runContext = spy(context.getKey().getRunContext()); + var otherNamespaceStore = mock(KVStore.class); + when(otherNamespaceStore.getValue(anyString())).thenReturn(Optional.empty()) + .thenReturn(Optional.of(new KVValue(true))); + doReturn(otherNamespaceStore).when(runContext).namespaceKv(base.getNamespace() + ".other"); + var conditionContext = context.getKey().withRunContext(runContext); + for (var scope : scopes) { + doReturn(ScriptOutput.builder().exitCode(0).build()).when(trigger).executeTask(any()); + assertTrue(trigger.evaluate(conditionContext, scope).isPresent()); + assertTrue(trigger.evaluate(conditionContext, scope).isEmpty()); + doThrow(new IOException("Docker unavailable")).when(trigger).executeTask(any()); + assertTrue(trigger.evaluate(conditionContext, scope).isEmpty()); + doReturn(ScriptOutput.builder().exitCode(0).build()).when(trigger).executeTask(any()); + assertTrue(trigger.evaluate(conditionContext, scope).isEmpty()); + } + } + + @ParameterizedTest + @ValueSource(booleans = { false, true }) + void kvFailuresPropagate(boolean writeFailure) throws Exception { + var trigger = spy(trigger(false, id(), "unused", "exit 0", true)); + var context = TestsUtils.mockTrigger(runContextFactory, trigger); + var runContext = spy(context.getKey().getRunContext()); + var store = mock(KVStore.class); + doReturn(store).when(runContext).namespaceKv(anyString()); + doReturn(ScriptOutput.builder().exitCode(0).build()).when(trigger).executeTask(any()); + if (writeFailure) { + doThrow(new IOException("KV write unavailable")).when(store).put(anyString(), any(KVValueAndMetadata.class)); + } else { + doThrow(new IOException("KV read unavailable")).when(store).getValue(anyString()); + } + assertThrows(IOException.class, () -> trigger.evaluate(context.getKey().withRunContext(runContext), context.getValue())); + } + + @ParameterizedTest + @ValueSource(booleans = { false, true }) + void templateDelimitersKeepUnderlyingTaskSemantics(boolean commands) throws Exception { + String script = "println '::{\"outputs\":{\"literal\":\"{{ '{{ absent }}' }}\"}}::'"; + var trigger = commands + ? CommandsTrigger.builder().id(id()).type(CommandsTrigger.class.getName()) + .containerImage(Property.ofValue("groovy:jdk21")) + .commands(TestsUtils.propertyFromList(List.of("groovy -e \"" + script.replace("\"", "\\\"") + "\""))) + .exitCondition(Property.ofValue("exit 0")).build() + : ScriptTrigger.builder().id(id()).type(ScriptTrigger.class.getName()) + .containerImage(Property.ofValue("groovy:jdk21")) + .script(Property.ofExpression(script)).exitCondition(Property.ofValue("exit 0")).build(); + var context = TestsUtils.mockTrigger(runContextFactory, trigger); + if (commands) { + var embedded = (CommandsTrigger) trigger; + var task = Commands.builder().id(id()).type(Commands.class.getName()) + .containerImage(Property.ofValue("groovy:jdk21")).commands(embedded.getCommands()).build(); + var output = task.run(TestsUtils.mockRunContext(runContextFactory, task, Map.of())); + assertEquals(Map.of("literal", "{{ absent }}"), output.getVars()); + var vars = trigger.evaluate(context.getKey(), context.getValue()).orElseThrow().getTrigger().getVariables().get("vars"); + assertEquals(output.getVars(), vars); + } else { + // Script's input-file handling renders the script again in Kestra 1.3.39. + // Preserve the task's existing behavior rather than adding a trigger-specific escape. + var task = Script.builder().id(id()).type(Script.class.getName()) + .containerImage(Property.ofValue("groovy:jdk21")).script(Property.ofExpression(script)).build(); + assertThrows( + IllegalVariableEvaluationException.class, + () -> task.run(TestsUtils.mockRunContext(runContextFactory, task, Map.of())) + ); + assertTrue(trigger.evaluate(context.getKey(), context.getValue()).isEmpty()); + } + } + + @ParameterizedTest + @ValueSource(booleans = { false, true }) + void literalValuePropertiesKeepUnderlyingTaskSemantics(boolean commands) throws Exception { + String script = "println '::{\"outputs\":{\"literal\":\"{{ absent }}\"}}::'"; + String command = "groovy -e \"" + script.replace("\"", "\\\"") + "\""; + if (commands) { + var task = Commands.builder().id(id()).type(Commands.class.getName()) + .containerImage(Property.ofValue("groovy:jdk21")) + .commands(Property.ofValue(List.of(command))).build(); + var output = task.run(TestsUtils.mockRunContext(runContextFactory, task, Map.of())); + assertEquals(Map.of("literal", "{{ absent }}"), output.getVars()); + var trigger = CommandsTrigger.builder().id(id()).type(CommandsTrigger.class.getName()) + .containerImage(Property.ofValue("groovy:jdk21")) + .commands(Property.ofValue(List.of(command))).exitCondition(Property.ofValue("exit 0")).build(); + var context = TestsUtils.mockTrigger(runContextFactory, trigger); + var vars = trigger.evaluate(context.getKey(), context.getValue()).orElseThrow().getTrigger().getVariables().get("vars"); + assertEquals(output.getVars(), vars); + } else { + // Even literal Script values pass through the task's input-file rendering. + var task = Script.builder().id(id()).type(Script.class.getName()) + .containerImage(Property.ofValue("groovy:jdk21")).script(Property.ofValue(script)).build(); + assertThrows( + IllegalVariableEvaluationException.class, + () -> task.run(TestsUtils.mockRunContext(runContextFactory, task, Map.of())) + ); + var trigger = ScriptTrigger.builder().id(id()).type(ScriptTrigger.class.getName()) + .containerImage(Property.ofValue("groovy:jdk21")) + .script(Property.ofValue(script)).exitCondition(Property.ofValue("exit 0")).build(); + var context = TestsUtils.mockTrigger(runContextFactory, trigger); + assertTrue(trigger.evaluate(context.getKey(), context.getValue()).isEmpty()); + } + } + + @Test + void multipleCommandsPreserveFailureAndEarlierVars() throws Exception { + var trigger = CommandsTrigger.builder().id(id()).type(CommandsTrigger.class.getName()) + .containerImage(Property.ofValue("groovy:jdk21")) + .commands( + Property.ofValue( + List.of( + "groovy -e 'println \"::{\\\"outputs\\\":{\\\"status\\\":\\\"ready\\\"}}::\"'", + "groovy -e 'System.exit(1)'", + "groovy -e 'println \"::{\\\"outputs\\\":{\\\"unexpected\\\":true}}::\"'" + ) + ) + ) + .exitCondition(Property.ofValue("exit 1")).build(); + var context = TestsUtils.mockTrigger(runContextFactory, trigger); + var variables = trigger.evaluate(context.getKey(), context.getValue()).orElseThrow().getTrigger().getVariables(); + assertEquals(1, variables.get("exitCode")); + assertEquals(Map.of("status", "ready"), variables.get("vars")); + } +} diff --git a/plugin-script-node/src/test/java/io/kestra/core/tasks/scripts/NodeTest.java b/plugin-script-node/src/test/java/io/kestra/core/tasks/scripts/NodeTest.java index 15b26339..e9242222 100644 --- a/plugin-script-node/src/test/java/io/kestra/core/tasks/scripts/NodeTest.java +++ b/plugin-script-node/src/test/java/io/kestra/core/tasks/scripts/NodeTest.java @@ -6,6 +6,7 @@ import java.util.Map; import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.Timeout; import com.google.common.collect.ImmutableMap; @@ -22,6 +23,7 @@ import static org.hamcrest.MatcherAssert.assertThat; import static org.hamcrest.Matchers.*; +import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertThrows; @KestraTest @@ -77,25 +79,101 @@ void failed() throws Exception { assertThat(((io.kestra.plugin.scripts.exec.scripts.models.ScriptOutput) nodeException.getOutput()).getStdErrLineCount(), equalTo(0)); } - @Test - void requirements() throws Exception { + private Node requirementsNode(int status) { Map files = new HashMap<>(); - files.put("main.js", "require('axios').get('http://google.com').then(r => { console.log('::{\"outputs\": {\"extract\":\"' + r.status + '\"}}::') })"); + files.put("main.js", """ + const http = require('http'); + const axios = require('axios'); + const sockets = new Set(); + const server = http.createServer((req, res) => { + res.writeHead(%d, { 'Content-Type': 'text/plain' }); + res.end('kestra-axios-fixture'); + }); + server.on('connection', socket => { + sockets.add(socket); + socket.on('close', () => sockets.delete(socket)); + }); + const deadline = setTimeout(() => { + console.error('HTTP fixture timed out'); + process.exit(1); + }, 15000); + (async () => { + try { + await new Promise((resolve, reject) => { + server.once('error', reject); + server.listen(0, '127.0.0.1', resolve); + }); + const response = await axios.get('http://127.0.0.1:' + server.address().port, { + proxy: false, + timeout: 5000 + }); + if (response.status !== 200 || response.data !== 'kestra-axios-fixture') { + throw new Error('Unexpected HTTP fixture response'); + } + console.log('::' + JSON.stringify({ outputs: { extract: String(response.status), body: response.data } }) + '::'); + } finally { + sockets.forEach(socket => socket.destroy()); + await new Promise(resolve => server.close(resolve)); + clearTimeout(deadline); + } + })().catch(error => { + console.error(error.message); + process.exitCode = 1; + }); + """.formatted(status)); files.put("package.json", "{\"dependencies\":{\"axios\":\"^0.20.0\"}}"); - - Node node = Node.builder() + return Node.builder() .id("test-node-task") .type(Node.class.getName()) .nodePath(Property.ofValue("node")) .npmPath(Property.ofValue("npm")) .inputFiles(files) .build(); + } + @Test + @Timeout(90) + void requirements() throws Exception { + Node node = requirementsNode(200); RunContext runContext = TestsUtils.mockRunContext(runContextFactory, node, ImmutableMap.of()); ScriptOutput run = node.run(runContext); - assertThat(run.getExitCode(), is(0)); assertThat(run.getVars().get("extract"), is("200")); + assertThat(run.getVars().get("body"), is("kestra-axios-fixture")); + } + + @Test + @Timeout(90) + void requirementsHttpFailureDoesNotEmitSuccess() throws Exception { + Node node = requirementsNode(503); + RunContext runContext = TestsUtils.mockRunContext(runContextFactory, node, ImmutableMap.of()); + var failure = assertThrows(RunnableTaskException.class, () -> node.run(runContext)); + var output = (io.kestra.plugin.scripts.exec.scripts.models.ScriptOutput) failure.getOutput(); + assertThat(output.getExitCode(), is(1)); + assertFalse(output.getVars().containsKey("extract")); + } + + @Test + @Timeout(90) + void invalidRequirementsDoNotExecuteMain() throws Exception { + Node node = Node.builder() + .id("test-node-task") + .type(Node.class.getName()) + .nodePath(Property.ofValue("node")) + .npmPath(Property.ofValue("npm")) + .inputFiles( + Map.of( + "package.json", "{invalid json", + "main.js", "console.log('::{\"outputs\":{\"mainExecuted\":true}}::')" + ) + ) + .build(); + RunContext runContext = TestsUtils.mockRunContext(runContextFactory, node, ImmutableMap.of()); + var failure = assertThrows(RunnableTaskException.class, () -> node.run(runContext)); + var output = (io.kestra.plugin.scripts.exec.scripts.models.ScriptOutput) failure.getOutput(); + assertThat(output.getExitCode(), not(0)); + assertFalse(output.getVars().containsKey("mainExecuted")); + assertThat(output.getStdOutLineCount(), is(0)); } @Test