141 lines
5.6 KiB
C#
141 lines
5.6 KiB
C#
|
|
using System.Text.Json.Nodes;
|
||
|
|
using w4c_workflows.Models.Nodes;
|
||
|
|
using w4c_workflows.Services.Nodes;
|
||
|
|
using w4c_workflows.Services.Nodes.Executors;
|
||
|
|
using Xunit;
|
||
|
|
|
||
|
|
namespace w4c_workflows.Tests;
|
||
|
|
|
||
|
|
/// <summary>
|
||
|
|
/// Unit tests for the <c>core.executeWorkflow</c> node: parameter validation,
|
||
|
|
/// fan-out modes, child-failure mapping and the no-wait behaviour. The actual
|
||
|
|
/// child execution is exercised end-to-end in <c>SubWorkflowInvokerTests</c>.
|
||
|
|
/// </summary>
|
||
|
|
public class ExecuteWorkflowNodeExecutorTests
|
||
|
|
{
|
||
|
|
private static readonly NodeBlueprintCatalog Catalog = NodeTestData.CoreCatalog();
|
||
|
|
|
||
|
|
private static FlowItem Item(string json) => FlowItem.FromJson(JsonNode.Parse(json)!.AsObject());
|
||
|
|
|
||
|
|
private sealed class FakeInvoker : ISubWorkflowInvoker
|
||
|
|
{
|
||
|
|
public string? ChildJson { get; init; }
|
||
|
|
public NodeFailure? Failure { get; init; }
|
||
|
|
public List<SubWorkflowInvocation> Calls { get; } = new();
|
||
|
|
|
||
|
|
public int MaxDepth => 10;
|
||
|
|
|
||
|
|
public Task<SubWorkflowRunResult> InvokeAsync(SubWorkflowInvocation invocation, CancellationToken ct)
|
||
|
|
{
|
||
|
|
Calls.Add(invocation);
|
||
|
|
if (Failure != null)
|
||
|
|
return Task.FromResult(new SubWorkflowRunResult { Items = Array.Empty<FlowItem>(), Failure = Failure });
|
||
|
|
|
||
|
|
var items = ChildJson == null
|
||
|
|
? Array.Empty<FlowItem>()
|
||
|
|
: new FlowItem[] { Item(ChildJson) };
|
||
|
|
return Task.FromResult(SubWorkflowRunResult.Ok(items, Guid.NewGuid()));
|
||
|
|
}
|
||
|
|
}
|
||
|
|
|
||
|
|
private static NodeExecutionContext Context(
|
||
|
|
JsonObject parameters, ISubWorkflowInvoker? invoker, params IReadOnlyList<FlowItem>[] inputs)
|
||
|
|
=> new()
|
||
|
|
{
|
||
|
|
Blueprint = Catalog.Get("core.executeWorkflow")!,
|
||
|
|
Parameters = parameters,
|
||
|
|
Inputs = inputs,
|
||
|
|
SubWorkflows = invoker,
|
||
|
|
TaskId = "call",
|
||
|
|
};
|
||
|
|
|
||
|
|
[Fact]
|
||
|
|
public async Task Fails_when_no_sub_workflow_invoker_is_installed()
|
||
|
|
{
|
||
|
|
var outcome = await new ExecuteWorkflowNodeExecutor().RunAsync(
|
||
|
|
Context(new JsonObject { ["workflow"] = "child" }, null, Array.Empty<FlowItem>()), default);
|
||
|
|
|
||
|
|
Assert.False(outcome.Succeeded);
|
||
|
|
Assert.Equal("subworkflow_unavailable", outcome.Failure!.Code);
|
||
|
|
}
|
||
|
|
|
||
|
|
[Fact]
|
||
|
|
public async Task Requires_a_workflow_parameter()
|
||
|
|
{
|
||
|
|
var outcome = await new ExecuteWorkflowNodeExecutor().RunAsync(
|
||
|
|
Context(new JsonObject(), new FakeInvoker(), Array.Empty<FlowItem>()), default);
|
||
|
|
|
||
|
|
Assert.False(outcome.Succeeded);
|
||
|
|
Assert.Equal("invalid_parameter", outcome.Failure!.Code);
|
||
|
|
}
|
||
|
|
|
||
|
|
[Fact]
|
||
|
|
public async Task Runs_once_with_all_items_and_maps_the_child_output()
|
||
|
|
{
|
||
|
|
var invoker = new FakeInvoker { ChildJson = """{"child":true}""" };
|
||
|
|
var outcome = await new ExecuteWorkflowNodeExecutor().RunAsync(
|
||
|
|
Context(new JsonObject { ["workflow"] = "child-transform" }, invoker,
|
||
|
|
new[] { Item("""{"n":1}"""), Item("""{"n":2}""") }), default);
|
||
|
|
|
||
|
|
Assert.True(outcome.Succeeded);
|
||
|
|
var invocation = Assert.Single(invoker.Calls);
|
||
|
|
Assert.Equal(SubWorkflowMode.AllItems, invocation.Mode);
|
||
|
|
Assert.Equal(2, invocation.Input.Count);
|
||
|
|
Assert.True(invocation.RecordHistory);
|
||
|
|
var item = Assert.Single(outcome.Outputs[0]);
|
||
|
|
Assert.True(item.Json["child"]!.GetValue<bool>());
|
||
|
|
}
|
||
|
|
|
||
|
|
[Fact]
|
||
|
|
public async Task Forwards_each_item_mode_to_the_invoker()
|
||
|
|
{
|
||
|
|
// Fan-out itself is the invoker's job (covered end-to-end); the executor
|
||
|
|
// only validates and forwards the mode.
|
||
|
|
var invoker = new FakeInvoker { ChildJson = """{"child":true}""" };
|
||
|
|
var outcome = await new ExecuteWorkflowNodeExecutor().RunAsync(
|
||
|
|
Context(new JsonObject { ["workflow"] = "child", ["mode"] = "eachItem" }, invoker,
|
||
|
|
new[] { Item("""{"n":1}"""), Item("""{"n":2}""") }), default);
|
||
|
|
|
||
|
|
Assert.True(outcome.Succeeded);
|
||
|
|
var invocation = Assert.Single(invoker.Calls);
|
||
|
|
Assert.Equal(SubWorkflowMode.EachItem, invocation.Mode);
|
||
|
|
Assert.Equal(2, invocation.Input.Count);
|
||
|
|
}
|
||
|
|
|
||
|
|
[Fact]
|
||
|
|
public async Task Maps_a_child_failure_to_a_node_failure()
|
||
|
|
{
|
||
|
|
var invoker = new FakeInvoker { Failure = new NodeFailure("child boom", "child_code") };
|
||
|
|
var outcome = await new ExecuteWorkflowNodeExecutor().RunAsync(
|
||
|
|
Context(new JsonObject { ["workflow"] = "child" }, invoker, new[] { Item("{}") }), default);
|
||
|
|
|
||
|
|
Assert.False(outcome.Succeeded);
|
||
|
|
Assert.Equal("child boom", outcome.Failure!.Message);
|
||
|
|
Assert.Equal("child_code", outcome.Failure.Code);
|
||
|
|
}
|
||
|
|
|
||
|
|
[Fact]
|
||
|
|
public async Task Does_not_emit_child_items_when_not_waiting()
|
||
|
|
{
|
||
|
|
var invoker = new FakeInvoker { ChildJson = """{"child":true}""" };
|
||
|
|
var outcome = await new ExecuteWorkflowNodeExecutor().RunAsync(
|
||
|
|
Context(new JsonObject { ["workflow"] = "child", ["waitForCompletion"] = false },
|
||
|
|
invoker, new[] { Item("""{"n":1}""") }), default);
|
||
|
|
|
||
|
|
Assert.True(outcome.Succeeded);
|
||
|
|
Assert.Single(invoker.Calls); // the child still ran
|
||
|
|
Assert.Empty(outcome.Outputs); // but its items are not forwarded
|
||
|
|
}
|
||
|
|
|
||
|
|
[Fact]
|
||
|
|
public async Task Rejects_an_invalid_mode()
|
||
|
|
{
|
||
|
|
var outcome = await new ExecuteWorkflowNodeExecutor().RunAsync(
|
||
|
|
Context(new JsonObject { ["workflow"] = "child", ["mode"] = "sideways" },
|
||
|
|
new FakeInvoker(), new[] { Item("{}") }), default);
|
||
|
|
|
||
|
|
Assert.False(outcome.Succeeded);
|
||
|
|
Assert.Equal("invalid_parameter", outcome.Failure!.Code);
|
||
|
|
}
|
||
|
|
}
|