Tests

Author: theodorzoulias

.gitignore

@@ -343,5 +343,4 @@ apps/
 _VARIOUS/
 .editorconfig
 SimpleTplDataflowPipelines.Local.sln
-logo_square.png
 tests/SimpleTplDataflowPipelines.Tests.Framework/

tests/SimpleTplDataflowPipelines.Tests/PipelineBuilder.FunctionalityAdvanced.cs

@@ -32,9 +32,40 @@ public async Task PostCompletionAction()
             pipeline.Complete();
             await pipeline.Completion.WithTimeout(1000);
             Console.WriteLine(String.Join(", ", timestamps));
+            Assert.IsTrue(timestamps.All(x => x > 0L));
             Assert.IsTrue(timestamps.Zip(timestamps.Skip(1)).All(e => e.First <= e.Second));
         }
 
+        [TestMethod]
+        public void FailedPostCompletionAction()
+        {
+            var block1 = new TransformBlock<int, int>(x => x,
+                new ExecutionDataflowBlockOptions() { BoundedCapacity = 1 });
+            var block2 = new ActionBlock<int>(x => Thread.Sleep(20));
+
+            var pipeline = PipelineBuilder
+                .BeginWith(block1)
+                .WithPostCompletionAction(t => throw new ApplicationException(t.Status.ToString()))
+                .LinkTo(block2)
+                .WithPostCompletionAction(t => throw new ApplicationException(t.Status.ToString()))
+                .ToPipeline();
+
+            pipeline.Post(13);
+            pipeline.Post(42);
+            pipeline.Complete();
+
+            var aex = Assert.ThrowsException<AggregateException>(
+                () => pipeline.Completion.Wait(100));
+            Console.WriteLine(String.Join(", ", aex.InnerExceptions.Select(ex => ex.Message)));
+            Assert.IsTrue(aex.InnerExceptions.Count == 2);
+            Assert.IsTrue(aex.InnerExceptions.All(ex => ex is ApplicationException));
+            Assert.IsTrue(aex.InnerExceptions.Select(ex => ex.Message).SequenceEqual(new[] { "RanToCompletion", "Faulted" }));
+            Assert.IsTrue(block1.Completion.IsCompletedSuccessfully());
+            Assert.IsTrue(block2.Completion.IsFaulted);
+            Assert.IsTrue(block2.Completion.Exception.InnerExceptions.Count == 1);
+            Assert.IsTrue(block2.Completion.Exception.InnerException is PipelineException);
+        }
+
         [TestMethod]
         public async Task UnlinkedBlocks()
         {
@@ -56,5 +87,39 @@ public async Task UnlinkedBlocks()
             var list = await ((IReceivableSourceBlock<int>)pipeline).ToListAsync(new CancellationTokenSource(1000).Token);
             Assert.IsTrue(list.SequenceEqual(source.Append(100).Append(200)));
         }
+
+        [TestMethod]
+        public async Task FailedUnlinkedBlock()
+        {
+            var block2 = new ActionBlock<int>(async x => { await Task.Delay(20); throw new ApplicationException(x.ToString()); },
+                new ExecutionDataflowBlockOptions() { MaxDegreeOfParallelism = 2, BoundedCapacity = 2 });
+            var block1 = new ActionBlock<int>(async x => await block2.SendAsync(x),
+                new ExecutionDataflowBlockOptions() { BoundedCapacity = 2 });
+
+            var pipeline = PipelineBuilder
+                .BeginWith(block1)
+                .AddUnlinked(block2)
+                .ToPipeline();
+
+            var source = Enumerable.Range(1, 10);
+            await Task.Run(async () =>
+            {
+                foreach (var item in source) await pipeline.SendAsync(item);
+            }).WithTimeout(500);
+
+            pipeline.Complete();
+            var aex = Assert.ThrowsException<AggregateException>(
+                () => pipeline.Completion.Wait(100));
+            Console.WriteLine(String.Join(", ", aex.InnerExceptions.Select(ex => ex.Message)));
+            Assert.IsTrue(aex.InnerExceptions.Count == 2);
+            Assert.IsTrue(aex.InnerExceptions.All(ex => ex is ApplicationException));
+            Assert.IsTrue(aex.InnerExceptions.Select(ex => ex.Message).OrderBy(x => x).SequenceEqual(new[] { "1", "2" }));
+            Assert.IsTrue(block1.Completion.IsFaulted);
+            Assert.IsTrue(block1.Completion.Exception.InnerExceptions.Count == 1);
+            Assert.IsTrue(block1.Completion.Exception.InnerException is PipelineException);
+            Assert.IsTrue(block2.Completion.IsFaulted);
+            Assert.IsTrue(block2.Completion.Exception.InnerExceptions.Count == 2);
+            Assert.IsTrue(block2.Completion.Exception.InnerExceptions.All(ex => ex is ApplicationException));
+        }
     }
 }

tests/SimpleTplDataflowPipelines.Tests/PipelineBuilder.FunctionalityBasic.cs

@@ -290,7 +290,7 @@ public async Task Cancellation()
             await Task.Run(async () =>
             {
                 while (true) if (!await pipeline.SendAsync(0)) break;
-            });
+            }).WithTimeout(500);
 
             pipeline.Complete();