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; /// Unit tests for the core.* control nodes (filter / switch / merge / split / wait). 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[] 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(), new[] { Item("""{"b":2}""") }), default); Assert.Single(outcome.Outputs[0]); Assert.Equal(2, outcome.Outputs[0][0].Json["b"]!.GetValue()); } [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()); Assert.Equal(2, merged["b"]!.GetValue()); Assert.Equal("right", merged["shared"]!.GetValue()); // 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(); var executor = new SplitInBatchesNodeExecutor(); NodeExecutionContext Loop() => new() { Blueprint = Catalog.Get("core.splitInBatches")!, Parameters = new JsonObject { ["batchSize"] = 2 }, Inputs = new IReadOnlyList[] { 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]); } }