550 lines
23 KiB
C#
550 lines
23 KiB
C#
using System.Text.Json;
|
|
using Microsoft.Extensions.Logging.Abstractions;
|
|
using VolcanicCore.Api.Engine;
|
|
using VolcanicCore.Api.Registries;
|
|
using VolcanicCore.Core;
|
|
using VolcanicCore.Persistence;
|
|
using VolcanicCore.Persistence.Encryption;
|
|
using VolcanicCore.Persistence.Migrations;
|
|
using VolcanicCore.Persistence.Repositories;
|
|
using System.Security.Cryptography;
|
|
|
|
namespace VolcanicCore.Tests;
|
|
|
|
public class ExecutorTests
|
|
{
|
|
// --- test node types ---------------------------------------------------
|
|
private sealed class TestTrigger : ITriggerNode
|
|
{
|
|
public string Id { get; init; } = "test.trigger";
|
|
public string DisplayName => Id;
|
|
public NodeCategory Category => NodeCategory.Trigger;
|
|
public int InputCount => 0;
|
|
public int OutputCount => 1;
|
|
public string? ConnectionProviderId => null;
|
|
public JsonSchema ConfigSchema => new(JsonDocument.Parse("{}").RootElement);
|
|
public Task<IAsyncDisposable> StartAsync(INodeContext ctx, ITriggerSink sink, CancellationToken ct)
|
|
=> throw new NotSupportedException("triggered externally in tests");
|
|
}
|
|
|
|
private sealed class EchoAction : IActionNode
|
|
{
|
|
public string Id { get; init; } = "test.echo";
|
|
public string DisplayName => Id;
|
|
public NodeCategory Category => NodeCategory.Action;
|
|
public int InputCount => 1;
|
|
public int OutputCount => 1;
|
|
public string? ConnectionProviderId => null;
|
|
public JsonSchema ConfigSchema => new(JsonDocument.Parse("{}").RootElement);
|
|
public Task<IReadOnlyList<IReadOnlyList<Item>>> ExecuteAsync(
|
|
INodeContext ctx, IReadOnlyList<IReadOnlyList<Item>> inputsByPort, CancellationToken ct)
|
|
=> Task.FromResult<IReadOnlyList<IReadOnlyList<Item>>>(new[] { inputsByPort[0] });
|
|
}
|
|
|
|
private sealed class TerminalSink : IActionNode
|
|
{
|
|
public string Id { get; init; } = "test.sink";
|
|
public string DisplayName => Id;
|
|
public NodeCategory Category => NodeCategory.Terminal;
|
|
public int InputCount => 1;
|
|
public int OutputCount => 0;
|
|
public string? ConnectionProviderId => null;
|
|
public JsonSchema ConfigSchema => new(JsonDocument.Parse("{}").RootElement);
|
|
public Task<IReadOnlyList<IReadOnlyList<Item>>> ExecuteAsync(
|
|
INodeContext ctx, IReadOnlyList<IReadOnlyList<Item>> inputsByPort, CancellationToken ct)
|
|
=> Task.FromResult<IReadOnlyList<IReadOnlyList<Item>>>(Array.Empty<IReadOnlyList<Item>>());
|
|
}
|
|
|
|
// --- setup helpers -----------------------------------------------------
|
|
private sealed class NullInstanceRegistry : IConnectionInstanceRegistry
|
|
{
|
|
public Task<object> GetAsync(string id, CancellationToken ct)
|
|
=> throw new InvalidOperationException("no connections in this test");
|
|
}
|
|
|
|
private static (ExecutionRepository ex, NodeExecutionRepository ne, Executor exec, NodeRegistry nr)
|
|
Build(params INodeType[] nodes)
|
|
{
|
|
var path = Path.Combine(Path.GetTempPath(), $"vc-{Guid.NewGuid():N}.db");
|
|
var f = new SqliteConnectionFactory(path);
|
|
using (var cn = f.Open())
|
|
new Migrator(new IMigration[]
|
|
{
|
|
new Migration001Initial(), new Migration002Connections(), new Migration003Workflows(),
|
|
new Migration004Executions(), new Migration005NodeExecutions(),
|
|
}).Apply(cn);
|
|
var er = new ExecutionRepository(f);
|
|
var nr = new NodeExecutionRepository(f);
|
|
var registry = new NodeRegistry(nodes);
|
|
var exec = new Executor(er, nr, registry, new NullInstanceRegistry(), NullLoggerFactory.Instance);
|
|
return (er, nr, exec, registry);
|
|
}
|
|
|
|
private static JsonElement Item(int v)
|
|
=> JsonSerializer.SerializeToElement(new { v });
|
|
|
|
// --- tests -------------------------------------------------------------
|
|
|
|
[Fact]
|
|
public async Task Linear_trigger_action_terminal_records_all_three_node_executions_and_completes()
|
|
{
|
|
var (er, nr, exec, _) = Build(new TestTrigger(), new EchoAction(), new TerminalSink());
|
|
|
|
var graph = new WorkflowGraph
|
|
{
|
|
Nodes = new[] {
|
|
new WorkflowNode("t", "test.trigger", null, "{}"),
|
|
new WorkflowNode("a", "test.echo", null, "{}"),
|
|
new WorkflowNode("s", "test.sink", null, "{}"),
|
|
},
|
|
Edges = new[] {
|
|
new WorkflowEdge("t", 0, "a", 0),
|
|
new WorkflowEdge("a", 0, "s", 0),
|
|
},
|
|
};
|
|
|
|
var eid = await exec.StartExecutionAsync(
|
|
workflowId: "wf1",
|
|
graph: graph,
|
|
triggerNodeId: "t",
|
|
triggerKind: "manual",
|
|
triggerItems: new[] { new Core.Item(Item(1)) },
|
|
triggerData: null,
|
|
answerChannelId: null,
|
|
ct: CancellationToken.None);
|
|
|
|
var run = er.Get(eid)!;
|
|
Assert.Equal("completed", run.Status);
|
|
var rows = nr.ListByExecution(eid);
|
|
Assert.Equal(3, rows.Count);
|
|
Assert.All(rows, r => Assert.Equal("completed", r.Status));
|
|
// trigger has empty inputs and one output list
|
|
var trig = rows.Single(r => r.NodeId == "t");
|
|
Assert.Equal("[]", trig.InputsJson);
|
|
}
|
|
|
|
private sealed class Throwing : IActionNode
|
|
{
|
|
public string Id { get; init; } = "test.throw";
|
|
public string DisplayName => Id;
|
|
public NodeCategory Category => NodeCategory.Action;
|
|
public int InputCount => 1;
|
|
public int OutputCount => 1;
|
|
public string? ConnectionProviderId => null;
|
|
public JsonSchema ConfigSchema => new(JsonDocument.Parse("{}").RootElement);
|
|
public Task<IReadOnlyList<IReadOnlyList<Item>>> ExecuteAsync(
|
|
INodeContext ctx, IReadOnlyList<IReadOnlyList<Item>> inputsByPort, CancellationToken ct)
|
|
=> throw new InvalidOperationException("boom");
|
|
}
|
|
|
|
private sealed class Recorder : IActionNode
|
|
{
|
|
public List<INodeContext> Calls { get; } = new();
|
|
public string Id { get; init; } = "test.record";
|
|
public string DisplayName => Id;
|
|
public NodeCategory Category => NodeCategory.Terminal;
|
|
public int InputCount => 1;
|
|
public int OutputCount => 0;
|
|
public string? ConnectionProviderId => null;
|
|
public JsonSchema ConfigSchema => new(JsonDocument.Parse("{}").RootElement);
|
|
public Task<IReadOnlyList<IReadOnlyList<Item>>> ExecuteAsync(
|
|
INodeContext ctx, IReadOnlyList<IReadOnlyList<Item>> inputsByPort, CancellationToken ct)
|
|
{
|
|
lock (Calls) Calls.Add(ctx);
|
|
return Task.FromResult<IReadOnlyList<IReadOnlyList<Item>>>(Array.Empty<IReadOnlyList<Item>>());
|
|
}
|
|
}
|
|
|
|
[Fact]
|
|
public async Task Fanout_two_downstream_terminals_both_receive_and_complete()
|
|
{
|
|
var recA = new Recorder { Id = "sink.a" };
|
|
var recB = new Recorder { Id = "sink.b" };
|
|
var (er2, nr2, exec2, _) = Build(new TestTrigger(), recA, recB);
|
|
|
|
var graph = new WorkflowGraph
|
|
{
|
|
Nodes = new[] {
|
|
new WorkflowNode("t", "test.trigger", null, "{}"),
|
|
new WorkflowNode("a", "sink.a", null, "{}"),
|
|
new WorkflowNode("b", "sink.b", null, "{}"),
|
|
},
|
|
Edges = new[] {
|
|
new WorkflowEdge("t", 0, "a", 0),
|
|
new WorkflowEdge("t", 0, "b", 0),
|
|
},
|
|
};
|
|
|
|
var eid = await exec2.StartExecutionAsync("wf", graph, "t", "manual",
|
|
new[] { new Core.Item(Item(1)) }, null, null, CancellationToken.None);
|
|
|
|
Assert.Equal("completed", er2.Get(eid)!.Status);
|
|
Assert.Single(recA.Calls);
|
|
Assert.Single(recB.Calls);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task Failure_propagates_downstream_as_skipped_and_rolls_up_to_partial_or_failed()
|
|
{
|
|
var recDownstreamOfFailure = new Recorder { Id = "sink.after_throw" };
|
|
var recDownstreamOfSuccess = new Recorder { Id = "sink.after_echo" };
|
|
var (er2, nr2, exec2, _) = Build(new TestTrigger(), new Throwing(), new EchoAction(),
|
|
recDownstreamOfFailure, recDownstreamOfSuccess);
|
|
|
|
var graph = new WorkflowGraph
|
|
{
|
|
Nodes = new[] {
|
|
new WorkflowNode("t", "test.trigger", null, "{}"),
|
|
new WorkflowNode("tf", "test.throw", null, "{}"),
|
|
new WorkflowNode("te", "test.echo", null, "{}"),
|
|
new WorkflowNode("sf", "sink.after_throw", null, "{}"),
|
|
new WorkflowNode("se", "sink.after_echo", null, "{}"),
|
|
},
|
|
Edges = new[] {
|
|
new WorkflowEdge("t", 0, "tf", 0),
|
|
new WorkflowEdge("t", 0, "te", 0),
|
|
new WorkflowEdge("tf", 0, "sf", 0),
|
|
new WorkflowEdge("te", 0, "se", 0),
|
|
},
|
|
};
|
|
|
|
var eid = await exec2.StartExecutionAsync("wf", graph, "t", "manual",
|
|
new[] { new Core.Item(Item(1)) }, null, null, CancellationToken.None);
|
|
|
|
var rows = nr2.ListByExecution(eid);
|
|
Assert.Equal("failed", rows.Single(r => r.NodeId == "tf").Status);
|
|
Assert.Equal("skipped_due_to_failure", rows.Single(r => r.NodeId == "sf").Status);
|
|
Assert.Equal("completed", rows.Single(r => r.NodeId == "te").Status);
|
|
Assert.Equal("completed", rows.Single(r => r.NodeId == "se").Status);
|
|
// Trigger succeeded + one branch succeeded + one branch failed → partial.
|
|
Assert.Equal("partial", er2.Get(eid)!.Status);
|
|
Assert.Empty(recDownstreamOfFailure.Calls);
|
|
Assert.Single(recDownstreamOfSuccess.Calls);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task Config_string_fields_are_templated_from_first_input_item_before_node_runs()
|
|
{
|
|
var rec = new Recorder { Id = "sink.template" };
|
|
var (er, nr, exec, _) = Build(new TestTrigger(), rec);
|
|
|
|
var graph = new WorkflowGraph
|
|
{
|
|
Nodes = new[] {
|
|
new WorkflowNode("t", "test.trigger", null, "{}"),
|
|
new WorkflowNode("s", "sink.template", null,
|
|
"""{"greeting":"hi {input.name}","meta":{"deep":"v={input.tags[0]}"},"list":["a","b {input.name}"]}"""),
|
|
},
|
|
Edges = new[] { new WorkflowEdge("t", 0, "s", 0) },
|
|
};
|
|
|
|
var trig = new Core.Item(JsonSerializer.SerializeToElement(new { name = "world", tags = new[] { "x", "y" } }));
|
|
var eid = await exec.StartExecutionAsync("wf", graph, "t", "manual",
|
|
new[] { trig }, null, null, CancellationToken.None);
|
|
|
|
Assert.Equal("completed", er.Get(eid)!.Status);
|
|
var cfg = rec.Calls.Single().Config;
|
|
Assert.Equal("hi world", cfg.GetProperty("greeting").GetString());
|
|
Assert.Equal("v=x", cfg.GetProperty("meta").GetProperty("deep").GetString());
|
|
var list = cfg.GetProperty("list");
|
|
Assert.Equal("a", list[0].GetString());
|
|
Assert.Equal("b world", list[1].GetString());
|
|
}
|
|
|
|
[Fact]
|
|
public async Task Missing_input_property_renders_as_empty_in_templated_config()
|
|
{
|
|
var rec = new Recorder { Id = "sink.template2" };
|
|
var (er, _, exec, _) = Build(new TestTrigger(), rec);
|
|
|
|
var graph = new WorkflowGraph
|
|
{
|
|
Nodes = new[] {
|
|
new WorkflowNode("t", "test.trigger", null, "{}"),
|
|
new WorkflowNode("s", "sink.template2", null, """{"msg":"value: {input.missing}"}"""),
|
|
},
|
|
Edges = new[] { new WorkflowEdge("t", 0, "s", 0) },
|
|
};
|
|
|
|
var trig = new Core.Item(JsonSerializer.SerializeToElement(new { other = 1 }));
|
|
var eid = await exec.StartExecutionAsync("wf", graph, "t", "manual",
|
|
new[] { trig }, null, null, CancellationToken.None);
|
|
|
|
Assert.Equal("completed", er.Get(eid)!.Status);
|
|
Assert.Equal("value: ", rec.Calls.Single().Config.GetProperty("msg").GetString());
|
|
}
|
|
|
|
[Fact]
|
|
public async Task All_downstream_failed_rolls_up_to_failed()
|
|
{
|
|
var (er, nr, exec, _) = Build(new TestTrigger(), new Throwing());
|
|
var graph = new WorkflowGraph
|
|
{
|
|
Nodes = new[] {
|
|
new WorkflowNode("t", "test.trigger", null, "{}"),
|
|
new WorkflowNode("tf", "test.throw", null, "{}"),
|
|
},
|
|
Edges = new[] { new WorkflowEdge("t", 0, "tf", 0) },
|
|
};
|
|
var eid = await exec.StartExecutionAsync("wf", graph, "t", "manual",
|
|
new[] { new Core.Item(Item(1)) }, null, null, CancellationToken.None);
|
|
Assert.Equal("failed", er.Get(eid)!.Status);
|
|
}
|
|
|
|
// --- coercion of templated primitives ---------------------------------
|
|
|
|
// A test node that requires status to be a JSON number, mirroring
|
|
// WebResponseNode's real check pattern. We assert on ctx.Config here.
|
|
private sealed class NumberFieldNode : IActionNode
|
|
{
|
|
public List<INodeContext> Calls { get; } = new();
|
|
public string Id { get; init; } = "test.numberfield";
|
|
public string DisplayName => Id;
|
|
public NodeCategory Category => NodeCategory.Terminal;
|
|
public int InputCount => 1;
|
|
public int OutputCount => 0;
|
|
public string? ConnectionProviderId => null;
|
|
public JsonSchema ConfigSchema => new(JsonDocument.Parse("""
|
|
{
|
|
"type": "object",
|
|
"properties": { "status": { "type": "integer" } }
|
|
}
|
|
""").RootElement);
|
|
public Task<IReadOnlyList<IReadOnlyList<Item>>> ExecuteAsync(
|
|
INodeContext ctx, IReadOnlyList<IReadOnlyList<Item>> _, CancellationToken __)
|
|
{
|
|
// Strict JsonValueKind.Number check — the coercion pass must produce
|
|
// a number here, not a string.
|
|
var s = ctx.Config.GetProperty("status");
|
|
if (s.ValueKind != JsonValueKind.Number)
|
|
throw new InvalidOperationException($"expected number, got {s.ValueKind}");
|
|
lock (Calls) Calls.Add(ctx);
|
|
return Task.FromResult<IReadOnlyList<IReadOnlyList<Item>>>(Array.Empty<IReadOnlyList<Item>>());
|
|
}
|
|
}
|
|
|
|
[Fact]
|
|
public async Task Template_into_integer_config_field_is_coerced_to_number()
|
|
{
|
|
var rec = new NumberFieldNode();
|
|
var (er, nr, exec, _) = Build(new TestTrigger(), rec);
|
|
|
|
var graph = new WorkflowGraph
|
|
{
|
|
Nodes = new[]
|
|
{
|
|
new WorkflowNode("t", "test.trigger", null, "{}"),
|
|
new WorkflowNode("s", "test.numberfield", null, """{"status":"{input.code}"}"""),
|
|
},
|
|
Edges = new[] { new WorkflowEdge("t", 0, "s", 0) },
|
|
};
|
|
|
|
var trig = new Core.Item(JsonSerializer.SerializeToElement(new { code = 201 }));
|
|
var eid = await exec.StartExecutionAsync("wf", graph, "t", "manual",
|
|
new[] { trig }, null, null, CancellationToken.None);
|
|
|
|
Assert.Equal("completed", er.Get(eid)!.Status);
|
|
var got = rec.Calls.Single().Config.GetProperty("status");
|
|
Assert.Equal(JsonValueKind.Number, got.ValueKind);
|
|
Assert.Equal(201, got.GetInt32());
|
|
}
|
|
|
|
[Fact]
|
|
public async Task Non_parseable_integer_config_fails_node_with_clear_message()
|
|
{
|
|
var rec = new NumberFieldNode();
|
|
var (er, nr, exec, _) = Build(new TestTrigger(), rec);
|
|
|
|
var graph = new WorkflowGraph
|
|
{
|
|
Nodes = new[]
|
|
{
|
|
new WorkflowNode("t", "test.trigger", null, "{}"),
|
|
new WorkflowNode("s", "test.numberfield", null, """{"status":"{input.code}"}"""),
|
|
},
|
|
Edges = new[] { new WorkflowEdge("t", 0, "s", 0) },
|
|
};
|
|
|
|
var trig = new Core.Item(JsonSerializer.SerializeToElement(new { code = "abc" }));
|
|
var eid = await exec.StartExecutionAsync("wf", graph, "t", "manual",
|
|
new[] { trig }, null, null, CancellationToken.None);
|
|
|
|
Assert.Equal("failed", er.Get(eid)!.Status);
|
|
var row = nr.ListByExecution(eid).Single(r => r.NodeId == "s");
|
|
Assert.Equal("failed", row.Status);
|
|
Assert.NotNull(row.Error);
|
|
Assert.Contains("expected integer", row.Error!);
|
|
Assert.Contains("status", row.Error!);
|
|
Assert.Empty(rec.Calls);
|
|
}
|
|
|
|
// --- WrapPin helper --------------------------------------------------
|
|
|
|
[Fact]
|
|
public void WrapPin_object_produces_a_single_item_wrapping_the_object()
|
|
{
|
|
var pin = JsonDocument.Parse("""{"a":1}""").RootElement;
|
|
var items = Executor.WrapPin(pin);
|
|
Assert.Single(items);
|
|
Assert.Equal("""{"a":1}""", items[0].Json.GetRawText());
|
|
Assert.Null(items[0].Binary);
|
|
}
|
|
|
|
[Fact]
|
|
public void WrapPin_array_of_objects_produces_one_item_per_element()
|
|
{
|
|
var pin = JsonDocument.Parse("""[{"a":1},{"a":2}]""").RootElement;
|
|
var items = Executor.WrapPin(pin);
|
|
Assert.Equal(2, items.Count);
|
|
Assert.Equal("""{"a":1}""", items[0].Json.GetRawText());
|
|
Assert.Equal("""{"a":2}""", items[1].Json.GetRawText());
|
|
}
|
|
|
|
[Fact]
|
|
public void WrapPin_empty_array_produces_zero_items()
|
|
{
|
|
var pin = JsonDocument.Parse("[]").RootElement;
|
|
var items = Executor.WrapPin(pin);
|
|
Assert.Empty(items);
|
|
}
|
|
|
|
[Fact]
|
|
public void WrapPin_scalar_produces_a_single_item_with_the_scalar()
|
|
{
|
|
var pin = JsonDocument.Parse("42").RootElement;
|
|
var items = Executor.WrapPin(pin);
|
|
Assert.Single(items);
|
|
Assert.Equal("42", items[0].Json.GetRawText());
|
|
}
|
|
|
|
// --- trigger pinned_output on manual runs ------------------------------
|
|
|
|
[Fact]
|
|
public async Task Trigger_pinned_output_overrides_trigger_items_on_manual_run()
|
|
{
|
|
var rec = new Recorder { Id = "sink.trig_pin_manual" };
|
|
var (er, nr, exec, _) = Build(new TestTrigger(), rec);
|
|
|
|
var graph = new WorkflowGraph
|
|
{
|
|
Nodes = new[]
|
|
{
|
|
new WorkflowNode("t", "test.trigger", null, "{}",
|
|
PinnedOutputJson: """{"pinned":true}"""),
|
|
new WorkflowNode("s", "sink.trig_pin_manual", null, "{}"),
|
|
},
|
|
Edges = new[] { new WorkflowEdge("t", 0, "s", 0) },
|
|
};
|
|
|
|
var passedIn = new Core.Item(JsonSerializer.SerializeToElement(new { pinned = false }));
|
|
var eid = await exec.StartExecutionAsync("wf", graph, "t", "manual",
|
|
new[] { passedIn }, null, null, CancellationToken.None);
|
|
|
|
Assert.Equal("completed", er.Get(eid)!.Status);
|
|
var sinkRow = nr.ListByExecution(eid).Single(r => r.NodeId == "s");
|
|
using var inputs = JsonDocument.Parse(sinkRow.InputsJson);
|
|
var port0Item0Json = inputs.RootElement[0][0].GetProperty("json");
|
|
Assert.True(port0Item0Json.GetProperty("pinned").GetBoolean());
|
|
}
|
|
|
|
[Fact]
|
|
public async Task Trigger_pinned_output_is_ignored_on_non_manual_run()
|
|
{
|
|
var rec = new Recorder { Id = "sink.trig_pin_webhook" };
|
|
var (er, nr, exec, _) = Build(new TestTrigger(), rec);
|
|
|
|
var graph = new WorkflowGraph
|
|
{
|
|
Nodes = new[]
|
|
{
|
|
new WorkflowNode("t", "test.trigger", null, "{}",
|
|
PinnedOutputJson: """{"pinned":true}"""),
|
|
new WorkflowNode("s", "sink.trig_pin_webhook", null, "{}"),
|
|
},
|
|
Edges = new[] { new WorkflowEdge("t", 0, "s", 0) },
|
|
};
|
|
|
|
var passedIn = new Core.Item(JsonSerializer.SerializeToElement(new { pinned = false }));
|
|
var eid = await exec.StartExecutionAsync("wf", graph, "t", "http.webhook",
|
|
new[] { passedIn }, null, null, CancellationToken.None);
|
|
|
|
Assert.Equal("completed", er.Get(eid)!.Status);
|
|
var sinkRow = nr.ListByExecution(eid).Single(r => r.NodeId == "s");
|
|
using var inputs = JsonDocument.Parse(sinkRow.InputsJson);
|
|
var port0Item0Json = inputs.RootElement[0][0].GetProperty("json");
|
|
Assert.False(port0Item0Json.GetProperty("pinned").GetBoolean());
|
|
}
|
|
|
|
// --- node pinned_input on manual runs ---------------------------------
|
|
|
|
[Fact]
|
|
public async Task Node_pinned_input_overrides_upstream_output_on_manual_run()
|
|
{
|
|
var rec = new Recorder { Id = "sink.node_pin_manual" };
|
|
var (er, nr, exec, _) = Build(new TestTrigger(), new EchoAction(), rec);
|
|
|
|
var graph = new WorkflowGraph
|
|
{
|
|
Nodes = new[]
|
|
{
|
|
new WorkflowNode("t", "test.trigger", null, "{}"),
|
|
new WorkflowNode("a", "test.echo", null, "{}"),
|
|
new WorkflowNode("s", "sink.node_pin_manual", null, "{}",
|
|
PinnedInputJson: """{"pinned":true}"""),
|
|
},
|
|
Edges = new[]
|
|
{
|
|
new WorkflowEdge("t", 0, "a", 0),
|
|
new WorkflowEdge("a", 0, "s", 0),
|
|
},
|
|
};
|
|
|
|
var passedIn = new Core.Item(JsonSerializer.SerializeToElement(new { pinned = false }));
|
|
var eid = await exec.StartExecutionAsync("wf", graph, "t", "manual",
|
|
new[] { passedIn }, null, null, CancellationToken.None);
|
|
|
|
Assert.Equal("completed", er.Get(eid)!.Status);
|
|
// Sink's persisted inputs must be the pin, not what `a` emitted.
|
|
var sinkRow = nr.ListByExecution(eid).Single(r => r.NodeId == "s");
|
|
using var inputs = JsonDocument.Parse(sinkRow.InputsJson);
|
|
var port0Item0Json = inputs.RootElement[0][0].GetProperty("json");
|
|
Assert.True(port0Item0Json.GetProperty("pinned").GetBoolean());
|
|
// `a` still ran and its own output row exists — confirming we don't skip upstream.
|
|
Assert.Contains(nr.ListByExecution(eid), r => r.NodeId == "a" && r.Status == "completed");
|
|
}
|
|
|
|
[Fact]
|
|
public async Task Node_pinned_input_is_ignored_on_non_manual_run()
|
|
{
|
|
var rec = new Recorder { Id = "sink.node_pin_webhook" };
|
|
var (er, nr, exec, _) = Build(new TestTrigger(), new EchoAction(), rec);
|
|
|
|
var graph = new WorkflowGraph
|
|
{
|
|
Nodes = new[]
|
|
{
|
|
new WorkflowNode("t", "test.trigger", null, "{}"),
|
|
new WorkflowNode("a", "test.echo", null, "{}"),
|
|
new WorkflowNode("s", "sink.node_pin_webhook", null, "{}",
|
|
PinnedInputJson: """{"pinned":true}"""),
|
|
},
|
|
Edges = new[]
|
|
{
|
|
new WorkflowEdge("t", 0, "a", 0),
|
|
new WorkflowEdge("a", 0, "s", 0),
|
|
},
|
|
};
|
|
|
|
var passedIn = new Core.Item(JsonSerializer.SerializeToElement(new { pinned = false }));
|
|
var eid = await exec.StartExecutionAsync("wf", graph, "t", "http.webhook",
|
|
new[] { passedIn }, null, null, CancellationToken.None);
|
|
|
|
Assert.Equal("completed", er.Get(eid)!.Status);
|
|
var sinkRow = nr.ListByExecution(eid).Single(r => r.NodeId == "s");
|
|
using var inputs = JsonDocument.Parse(sinkRow.InputsJson);
|
|
var port0Item0Json = inputs.RootElement[0][0].GetProperty("json");
|
|
// Real upstream (echo of the trigger's `pinned:false`) reaches the sink.
|
|
Assert.False(port0Item0Json.GetProperty("pinned").GetBoolean());
|
|
}
|
|
}
|