Add unit and integration tests for ParallelForEach activity covering behavior with different item types, fault handling, and scenario-specific conditions (#7120)

* Add unit and integration tests for `ParallelForEach` activity covering behavior with different item types, fault handling, and scenario-specific conditions

* Replace `Array.Empty<string>()` with `[]` in `ParallelForEachTests` to simplify syntax.

src/modules/Elsa.Workflows.Core/Activities/ParallelForEachT.cs

@@ -2,6 +2,7 @@ using System.Runtime.CompilerServices;
 using System.Text.Json;
 using System.Text.Json.Nodes;
 using Elsa.Expressions.Helpers;
+using Elsa.Expressions.Models;
 using Elsa.Extensions;
 using Elsa.Workflows.Attributes;
 using Elsa.Workflows.Memory;
@@ -19,6 +20,27 @@ public class ParallelForEach<T> : Activity
     private const string ScheduledTagsProperty = nameof(ScheduledTagsProperty);
     private const string CompletedTagsProperty = nameof(CompletedTagsProperty);
 
+    /// <inheritdoc />
+    public ParallelForEach(Func<ExpressionExecutionContext, ICollection<T>> @delegate, [CallerFilePath] string? source = null, [CallerLineNumber] int? line = null) : this(new Input<object>(@delegate), source, line)
+    {
+    }
+
+    /// <inheritdoc />
+    public ParallelForEach(Func<ICollection<T>> @delegate, [CallerFilePath] string? source = null, [CallerLineNumber] int? line = null) : this(new Input<object>(@delegate), source, line)
+    {
+    }
+
+    /// <inheritdoc />
+    public ParallelForEach(ICollection<T> items, [CallerFilePath] string? source = null, [CallerLineNumber] int? line = null) : this(new Input<object>(items), source, line)
+    {
+    }
+
+    /// <inheritdoc />
+    public ParallelForEach(Input<object> items, [CallerFilePath] string? source = null, [CallerLineNumber] int? line = null) : this(source, line)
+    {
+        Items = items;
+    }
+
     /// <inheritdoc />
     public ParallelForEach([CallerFilePath] string? source = null, [CallerLineNumber] int? line = null) : base(source, line)
     {
@@ -42,7 +64,7 @@ public class ParallelForEach<T> : Activity
         var items = context.GetItemSource<T>(Items);
         var tags = new List<Guid>();
         var currentIndex = 0;
-        
+
         await foreach (var item in items)
         {
             // For each item, declare a new variable for the work to be scheduled.

test/integration/Elsa.Activities.IntegrationTests/ParallelForEachTests.cs

@@ -0,0 +1,135 @@
+using Elsa.Extensions;
+using Elsa.Testing.Shared;
+using Elsa.Workflows;
+using Elsa.Workflows.Activities;
+using Elsa.Workflows.Models;
+using Xunit.Abstractions;
+
+namespace Elsa.Activities.IntegrationTests;
+
+public class ParallelForEachTests(ITestOutputHelper testOutputHelper)
+{
+    private readonly WorkflowTestFixture _fixture = new(testOutputHelper);
+    private const string CurrentValueVar = "CurrentValue";
+    private static readonly string[] ThreeItems = ["a", "b", "c"];
+
+    private CapturingTextWriter CapturingTextWriter => _fixture.CapturingTextWriter;
+
+    [Theory(DisplayName = "ParallelForEach executes body for all items")]
+    [MemberData(nameof(ItemTestCases))]
+    public async Task ParallelForEach_ExecutesBody_ForAllItems(string[] items)
+    {
+        await ExecuteAndAssertAllItems(items);
+    }
+
+    [Fact(DisplayName = "ParallelForEach completes when collection is empty")]
+    public async Task ParallelForEach_Completes_WhenCollectionEmpty()
+    {
+        await ExecuteAndAssertStatus([], ActivityStatus.Completed);
+        Assert.Empty(CapturingTextWriter.Lines);
+    }
+
+    [Fact(DisplayName = "ParallelForEach completes when collection is null")]
+    public async Task ParallelForEach_Completes_WhenCollectionNull()
+    {
+        await ExecuteAndAssertStatus(null, ActivityStatus.Completed);
+    }
+
+    [Fact(DisplayName = "ParallelForEach executes all items when one faults")]
+    public async Task ParallelForEach_ExecutesAllItems_WhenOneFaults()
+    {
+        var body = new Sequence
+        {
+            Activities =
+            [
+                new If(context => context.GetVariable<string>(CurrentValueVar) == "b")
+                {
+                    Then = new Fault { Message = new("Faulted") },
+                },
+                WriteCurrentValue()
+            ]
+        };
+
+        await RunActivityAsync(ThreeItems, body);
+
+        Assert.Contains("a", CapturingTextWriter.Lines);
+        Assert.DoesNotContain("b", CapturingTextWriter.Lines);
+        Assert.Contains("c", CapturingTextWriter.Lines);
+    }
+
+    [Fact(DisplayName = "ParallelForEach executes body for different item types")]
+    public async Task ParallelForEach_ExecutesBody_ForDifferentItemTypes()
+    {
+        var items = new object?[]
+        {
+            "a", 2, null, new Foo()
+        };
+        var parallelForEach = new ParallelForEach<object?>(items)
+        {
+            Body = new WriteLine(context => context.GetVariable<object>(CurrentValueVar)?.ToString() ?? "")
+        };
+
+        await _fixture.RunActivityAsync(parallelForEach);
+
+        Assert.Equal(items.Length, CapturingTextWriter.Lines.Count);
+        Assert.Contains("a", CapturingTextWriter.Lines);
+        Assert.Contains("2", CapturingTextWriter.Lines);
+        Assert.Contains("", CapturingTextWriter.Lines);
+        Assert.Contains("Baz", CapturingTextWriter.Lines);
+    }
+
+    [Fact(DisplayName = "ParallelForEach provides CurrentIndex variable")]
+    public async Task ParallelForEach_ProvidesCurrentIndex_ForEachIteration()
+    {
+        var body = new WriteLine(context => context.GetVariable<int>("CurrentIndex").ToString());
+
+        await RunActivityAsync(ThreeItems, body);
+
+        Assert.Equal(ThreeItems.Length, CapturingTextWriter.Lines.Count);
+        Assert.Contains("0", CapturingTextWriter.Lines);
+        Assert.Contains("1", CapturingTextWriter.Lines);
+        Assert.Contains("2", CapturingTextWriter.Lines);
+    }
+
+    [Fact(DisplayName = "ParallelForEach completes when all bodies complete")]
+    public async Task ParallelForEach_Completes_WhenAllBodiesComplete()
+    {
+        await ExecuteAndAssertStatus(ThreeItems, ActivityStatus.Completed);
+    }
+
+    public static TheoryData<string[]> ItemTestCases =>
+    [
+        ThreeItems,
+        ["single"]
+    ];
+
+    private async Task ExecuteAndAssertAllItems(string[] items)
+    {
+        await RunActivityAsync(items, WriteCurrentValue());
+
+        Assert.Equal(items.Length, CapturingTextWriter.Lines.Count);
+        Assert.All(items, item => Assert.Contains(item, CapturingTextWriter.Lines));
+    }
+
+    private async Task ExecuteAndAssertStatus(string[]? items, ActivityStatus expectedStatus)
+    {
+        var result = await RunActivityAsync(items, WriteCurrentValue());
+        var context = GetParallelForEachContext(result);
+        Assert.Equal(expectedStatus, context.Status);
+    }
+
+    private async Task<RunWorkflowResult> RunActivityAsync(string[]? items, IActivity body)
+    {
+        var parallelForEach = new ParallelForEach<string>(items!) { Body = body };
+        return await _fixture.RunActivityAsync(parallelForEach);
+    }
+
+    private static WriteLine WriteCurrentValue() => new(context => context.GetVariable<string>(CurrentValueVar));
+
+    private static ActivityExecutionContext GetParallelForEachContext(RunWorkflowResult result)
+    {
+        var context = result.Journal.ActivityExecutionContexts.FirstOrDefault(x => x.Activity is ParallelForEach<string>);
+        Assert.NotNull(context);
+        return context;
+    }
+}

test/integration/Elsa.Activities.IntegrationTests/ParallelTests.cs

@@ -1,7 +1,6 @@
 using Elsa.Testing.Shared;
 using Elsa.Workflows;
 using Elsa.Workflows.Activities;
-using Elsa.Workflows.Models;
 using Xunit.Abstractions;
 using Parallel = Elsa.Workflows.Activities.Parallel;
 

test/unit/Elsa.Activities.UnitTests/Looping/ParallelForEachTests.cs

@@ -0,0 +1,94 @@
+using Elsa.Testing.Shared;
+using Elsa.Workflows;
+
+namespace Elsa.Activities.UnitTests.Looping;
+
+public class ParallelForEachTests
+{
+    [Theory]
+    [InlineData(1)]
+    [InlineData(2)]
+    [InlineData(3)]
+    public async Task Should_Schedule_Body_For_Each_Item(int itemCount)
+    {
+        var items = Enumerable.Range(0, itemCount).Select(i => $"item{i}").ToArray();
+        var body = new MockBodyActivity();
+        var parallelForEach = new ParallelForEach<string>(items) { Body = body };
+
+        var context = await ExecuteAsync(parallelForEach);
+
+        AssertScheduledCount(context, itemCount);
+        Assert.True(context.HasScheduledActivity(body));
+    }
+
+    [Fact]
+    public async Task Should_Complete_When_Items_Empty()
+    {
+        var body = new MockBodyActivity();
+        var parallelForEach = new ParallelForEach<string>(Array.Empty<string>()) { Body = body };
+
+        var context = await ExecuteAsync(parallelForEach);
+
+        Assert.Equal(ActivityStatus.Completed, context.Status);
+        Assert.False(context.HasScheduledActivity(body));
+    }
+
+    [Fact]
+    public async Task Should_Not_Schedule_When_Body_Null()
+    {
+        var items = new[] { "a", "b" };
+        var parallelForEach = new ParallelForEach<string>(items) { Body = null! };
+
+        var context = await ExecuteAsync(parallelForEach);
+
+        AssertScheduledCount(context, 0);
+    }
+
+    [Fact]
+    public async Task Should_Handle_Integer_Items()
+    {
+        var items = new[] { 1, 2, 3 };
+        var body = new MockBodyActivity();
+        var parallelForEach = new ParallelForEach<int>(items) { Body = body };
+
+        var context = await ExecuteAsync(parallelForEach);
+
+        AssertScheduledCount(context, items.Length);
+    }
+
+    [Fact]
+    public void Verify_Activity_Attributes()
+    {
+        var parallelForEach = new ParallelForEach();
+        var fixture = new ActivityTestFixture(parallelForEach);
+
+        fixture.AssertActivityAttributes(
+            expectedNamespace: "Elsa",
+            expectedKind: ActivityKind.Action,
+            expectedCategory: "Looping",
+            expectedDescription: "Schedule an activity for each item in parallel."
+        );
+    }
+
+    [Fact]
+    public void Default_Property_Values()
+    {
+        var parallelForEach = new ParallelForEach<string>();
+
+        Assert.NotNull(parallelForEach.Items);
+    }
+
+    private static Task<ActivityExecutionContext> ExecuteAsync(IActivity activity) =>
+        new ActivityTestFixture(activity).ExecuteAsync();
+
+    private static void AssertScheduledCount(ActivityExecutionContext context, int expectedCount)
+    {
+        var scheduledActivities = context.WorkflowExecutionContext.Scheduler.List().ToList();
+        Assert.Equal(expectedCount, scheduledActivities.Count);
+    }
+
+    private class MockBodyActivity : Activity
+    {
+        protected override ValueTask ExecuteAsync(ActivityExecutionContext context) => ValueTask.CompletedTask;
+    }
+}