w4c-workflows-api/w4c-workflows-api.Tests/ControlNodeExecutorTests.cs

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 core.* control nodes (filter / switch / merge / split / wait).</summary>
public class ControlNodeExecutorTests
{
    private static readonly NodeBlueprintCatalog Catalog = NodeTestData.CoreCatalog();

    private static FlowItem Item(string json) => FlowItem.FromJson(JsonNode.Parse(json)!.AsObject());

    private static NodeExecutionContext Context(
        string type, JsonObject parameters, params IReadOnlyList<FlowItem>[] inputs)
        => new()
        {
            Blueprint = Catalog.Get(type)!,
            Parameters = parameters,
            Inputs = inputs,
        };

    private static JsonArray Rules(params (string Left, string Op, string Right)[] rules)
    {
        var array = new JsonArray();
        foreach (var (left, op, right) in rules)
        {
            array.Add(new JsonObject { ["left"] = left, ["operator"] = op, ["right"] = right });
        }
        return array;
    }

    // ------------------------------------------------------------------ filter

    [Fact]
    public async Task Filter_keeps_matching_items_and_discards_the_rest()
    {
        var matched = await new FilterNodeExecutor().RunAsync(
            Context("core.filter", new JsonObject { ["left"] = "ok", ["operator"] = "equals", ["right"] = "ok" },
                new[] { Item("""{"id":1}""") }), default);

        Assert.True(matched.Succeeded);
        Assert.Single(matched.Outputs[0]);
        Assert.Empty(matched.Outputs[1]);

        var dropped = await new FilterNodeExecutor().RunAsync(
            Context("core.filter", new JsonObject { ["left"] = "ok", ["operator"] = "equals", ["right"] = "no" },
                new[] { Item("""{"id":1}""") }), default);

        Assert.Empty(dropped.Outputs[0]);
        Assert.Single(dropped.Outputs[1]);
    }

    // ------------------------------------------------------------------ switch

    [Fact]
    public async Task Switch_routes_to_the_first_matching_rule()
    {
        var outcome = await new SwitchNodeExecutor().RunAsync(
            Context("core.switch", new JsonObject
            {
                ["rules"] = Rules(("a", "equals", "b"), ("a", "equals", "a")),
                ["fallbackEnabled"] = true,
            }, new[] { Item("""{"x":1}""") }), default);

        Assert.True(outcome.Succeeded);
        Assert.Equal(5, outcome.Outputs.Count);
        Assert.Single(outcome.Outputs[1]); // second rule matched
        Assert.Empty(outcome.Outputs[0]);
        Assert.Empty(outcome.Outputs[4]);
    }

    [Fact]
    public async Task Switch_sends_unmatched_items_to_fallback()
    {
        var outcome = await new SwitchNodeExecutor().RunAsync(
            Context("core.switch", new JsonObject
            {
                ["rules"] = Rules(("a", "equals", "b")),
                ["fallbackEnabled"] = true,
            }, new[] { Item("""{"x":1}""") }), default);

        Assert.Single(outcome.Outputs[4]);
    }

    [Fact]
    public async Task Switch_drops_unmatched_when_fallback_is_disabled()
    {
        var outcome = await new SwitchNodeExecutor().RunAsync(
            Context("core.switch", new JsonObject
            {
                ["rules"] = Rules(("a", "equals", "b")),
                ["fallbackEnabled"] = false,
            }, new[] { Item("""{"x":1}""") }), default);

        Assert.All(outcome.Outputs, port => Assert.Empty(port));
    }

    // ------------------------------------------------------------------ merge

    [Fact]
    public async Task Merge_append_concatenates_both_inputs()
    {
        var outcome = await new MergeNodeExecutor().RunAsync(
            Context("core.merge", new JsonObject { ["mode"] = "append" },
                new[] { Item("""{"a":1}""") }, new[] { Item("""{"b":2}""") }), default);

        Assert.Equal(2, outcome.Outputs[0].Count);
    }

    [Fact]
    public async Task Merge_chooseBranch_takes_the_first_non_empty_input()
    {
        var outcome = await new MergeNodeExecutor().RunAsync(
            Context("core.merge", new JsonObject { ["mode"] = "chooseBranch" },
                Array.Empty<FlowItem>(), new[] { Item("""{"b":2}""") }), default);

        Assert.Single(outcome.Outputs[0]);
        Assert.Equal(2, outcome.Outputs[0][0].Json["b"]!.GetValue<int>());
    }

    [Fact]
    public async Task Merge_combineByPosition_merges_item_fields()
    {
        var outcome = await new MergeNodeExecutor().RunAsync(
            Context("core.merge", new JsonObject { ["mode"] = "combineByPosition" },
                new[] { Item("""{"a":1,"shared":"left"}""") },
                new[] { Item("""{"b":2,"shared":"right"}""") }), default);

        var merged = Assert.Single(outcome.Outputs[0]).Json;
        Assert.Equal(1, merged["a"]!.GetValue<int>());
        Assert.Equal(2, merged["b"]!.GetValue<int>());
        Assert.Equal("right", merged["shared"]!.GetValue<string>()); // later input wins
    }

    // ------------------------------------------------------------------ split

    [Fact]
    public async Task SplitInBatches_chunks_items_and_passes_all_to_done()
    {
        var items = Enumerable.Range(1, 5).Select(i => Item($$"""{"n":{{i}}}""")).ToArray();

        var outcome = await new SplitInBatchesNodeExecutor().RunAsync(
            Context("core.splitInBatches", new JsonObject { ["batchSize"] = 2 }, items), default);

        var batches = outcome.Outputs[0];
        Assert.Equal(3, batches.Count); // 2 + 2 + 1
        Assert.Equal(2, batches[0].Json["items"]!.AsArray().Count);
        Assert.Single(batches[2].Json["items"]!.AsArray());
        Assert.Equal(5, outcome.Outputs[1].Count); // done
    }

    [Fact]
    public async Task SplitInBatches_rejects_a_non_positive_batch_size()
    {
        var outcome = await new SplitInBatchesNodeExecutor().RunAsync(
            Context("core.splitInBatches", new JsonObject { ["batchSize"] = 0 },
                new[] { Item("{}") }), default);

        Assert.False(outcome.Succeeded);
        Assert.Equal("invalid_parameter", outcome.Failure!.Code);
    }

    [Fact]
    public async Task SplitInBatches_in_loop_mode_advances_one_batch_per_invocation()
    {
        var items = Enumerable.Range(1, 5).Select(i => Item($$"""{"n":{{i}}}""")).ToArray();
        var state = new Dictionary<string, object?>();
        var executor = new SplitInBatchesNodeExecutor();

        NodeExecutionContext Loop() => new()
        {
            Blueprint = Catalog.Get("core.splitInBatches")!,
            Parameters = new JsonObject { ["batchSize"] = 2 },
            Inputs = new IReadOnlyList<FlowItem>[] { items },
            LoopBackInput = true,
            State = state,
            TaskId = "loop",
        };

        var first = await executor.RunAsync(Loop(), default);
        Assert.False(first.LoopComplete);
        Assert.Equal(2, Assert.Single(first.Outputs[0]).Json["items"]!.AsArray().Count);

        var second = await executor.RunAsync(Loop(), default);
        Assert.Equal(2, Assert.Single(second.Outputs[0]).Json["items"]!.AsArray().Count);

        var third = await executor.RunAsync(Loop(), default);
        Assert.Single(Assert.Single(third.Outputs[0]).Json["items"]!.AsArray());

        var fourth = await executor.RunAsync(Loop(), default);
        Assert.True(fourth.LoopComplete);
        Assert.Empty(fourth.Outputs[0]);
        Assert.Equal(5, fourth.Outputs[1].Count); // done carries the original items
    }

    // ------------------------------------------------------------------ wait

    [Fact]
    public async Task Wait_passes_items_through_after_the_delay()
    {
        var outcome = await new WaitNodeExecutor().RunAsync(
            Context("core.wait", new JsonObject { ["amount"] = 1, ["unit"] = "milliseconds" },
                new[] { Item("""{"a":1}""") }), default);

        Assert.True(outcome.Succeeded);
        Assert.Single(outcome.Outputs[0]);
    }
}