Adding tests to ExportProcessors (#1183)

* Adding tests to BatchExportProcessor

* updating value

* updating tests

* adding tests to reentrantprocessor

* adding more tests

* updating tests

* disposing blocking collection

* renaming variable

test/OpenTelemetry.Tests/Shared/TestActivityExporter.cs

@@ -0,0 +1,53 @@
+// <copyright file="TestActivityExporter.cs" company="OpenTelemetry Authors">
+// Copyright The OpenTelemetry Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+// </copyright>
+
+using System.Collections.Concurrent;
+using System.Diagnostics;
+using OpenTelemetry.Trace;
+
+namespace OpenTelemetry.Tests.Shared
+{
+    internal class TestActivityExporter : ActivityExporter
+    {
+        internal readonly BlockingCollection<Activity> Exported = new BlockingCollection<Activity>();
+        private bool disposed;
+
+        public override ExportResult Export(in Batch<Activity> batch)
+        {
+            foreach (var activity in batch)
+            {
+                this.Exported.Add(activity);
+            }
+
+            return ExportResult.Success;
+        }
+
+        protected override void Dispose(bool disposing)
+        {
+            if (!this.disposed)
+            {
+                if (disposing)
+                {
+                    this.Exported.Dispose();
+                }
+
+                this.disposed = true;
+            }
+
+            base.Dispose(disposing);
+        }
+    }
+}

test/OpenTelemetry.Tests/Trace/BatchExportActivityProcessorTest.cs

@@ -0,0 +1,143 @@
+// <copyright file="BatchExportActivityProcessorTest.cs" company="OpenTelemetry Authors">
+// Copyright The OpenTelemetry Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+// </copyright>
+
+using System;
+using System.Diagnostics;
+using System.Threading;
+using OpenTelemetry.Tests.Shared;
+using OpenTelemetry.Trace;
+using Xunit;
+
+namespace OpenTelemetry.Tests.Trace
+{
+    public class BatchExportActivityProcessorTest
+    {
+        [Fact]
+        public void CheckNullExporter()
+        {
+            Assert.Throws<ArgumentNullException>(() => new BatchExportActivityProcessor(null));
+        }
+
+        [Fact]
+        public void CheckConstructorWithInvalidValues()
+        {
+            Assert.Throws<ArgumentOutOfRangeException>(() => new BatchExportActivityProcessor(new TestActivityExporter(), maxQueueSize: 0));
+            Assert.Throws<ArgumentOutOfRangeException>(() => new BatchExportActivityProcessor(new TestActivityExporter(), maxExportBatchSize: 0));
+            Assert.Throws<ArgumentOutOfRangeException>(() => new BatchExportActivityProcessor(new TestActivityExporter(), maxQueueSize: 1, maxExportBatchSize: 2049));
+            Assert.Throws<ArgumentOutOfRangeException>(() => new BatchExportActivityProcessor(new TestActivityExporter(), scheduledDelayMilliseconds: 0));
+            Assert.Throws<ArgumentOutOfRangeException>(() => new BatchExportActivityProcessor(new TestActivityExporter(), exporterTimeoutMilliseconds: -1));
+        }
+
+        [Fact]
+        public void CheckIfBatchIsExportingOnQueueLimit()
+        {
+            using var exporter = new TestActivityExporter();
+            using var processor = new BatchExportActivityProcessor(
+                exporter,
+                maxQueueSize: 1,
+                maxExportBatchSize: 1,
+                scheduledDelayMilliseconds: 100_000);
+
+            processor.OnEnd(new Activity("start"));
+
+            for (int i = 0; i < 10 && exporter.Exported.Count == 0; i++)
+            {
+                Thread.Sleep(500);
+            }
+
+            Assert.Single(exporter.Exported);
+
+            Assert.Equal(1, processor.ProcessedCount);
+            Assert.Equal(1, processor.ReceivedCount);
+            Assert.Equal(0, processor.DroppedCount);
+        }
+
+        [Fact]
+        public void CheckForceFlushWithInvalidTimeout()
+        {
+            using var exporter = new TestActivityExporter();
+            using var processor = new BatchExportActivityProcessor(exporter, maxQueueSize: 2, maxExportBatchSize: 1);
+            Assert.Throws<ArgumentOutOfRangeException>(() => processor.ForceFlush(-2));
+        }
+
+        [Theory]
+        [InlineData(Timeout.Infinite)]
+        [InlineData(0)]
+        [InlineData(1)]
+        public void CheckForceFlushExport(int timeout)
+        {
+            using var exporter = new TestActivityExporter();
+            using var processor = new BatchExportActivityProcessor(
+                exporter,
+                maxQueueSize: 3,
+                maxExportBatchSize: 3,
+                exporterTimeoutMilliseconds: 30000);
+
+            processor.OnEnd(new Activity("start1"));
+            processor.OnEnd(new Activity("start2"));
+
+            Assert.Equal(0, processor.ProcessedCount);
+
+            // waiting to see if time is triggering the exporter
+            Thread.Sleep(1_000);
+            Assert.Empty(exporter.Exported);
+
+            // forcing flush
+            processor.ForceFlush(timeout);
+
+            if (timeout == 0)
+            {
+                // ForceFlush(0) will trigger flush and return immediately, so let's sleep for a while
+                Thread.Sleep(1_000);
+            }
+
+            Assert.Equal(2, exporter.Exported.Count);
+
+            Assert.Equal(2, processor.ProcessedCount);
+            Assert.Equal(2, processor.ReceivedCount);
+            Assert.Equal(0, processor.DroppedCount);
+        }
+
+        [Theory]
+        [InlineData(Timeout.Infinite)]
+        [InlineData(0)]
+        [InlineData(1)]
+        public void CheckShutdownExport(int timeout)
+        {
+            using var exporter = new TestActivityExporter();
+            using var processor = new BatchExportActivityProcessor(
+                exporter,
+                maxQueueSize: 3,
+                maxExportBatchSize: 3,
+                exporterTimeoutMilliseconds: 30000);
+
+            processor.OnEnd(new Activity("start"));
+            processor.Shutdown(timeout);
+
+            if (timeout == 0)
+            {
+                // ForceFlush(0) will trigger flush and return immediately, so let's sleep for a while
+                Thread.Sleep(1_000);
+            }
+
+            Assert.Single(exporter.Exported);
+
+            Assert.Equal(1, processor.ProcessedCount);
+            Assert.Equal(1, processor.ReceivedCount);
+            Assert.Equal(0, processor.DroppedCount);
+        }
+    }
+}

test/OpenTelemetry.Tests/Trace/CompositeActivityProcessorTests.cs

@@ -16,6 +16,7 @@
 
 using System;
 using System.Diagnostics;
+using System.Threading;
 using OpenTelemetry.Tests;
 using Xunit;
 
@@ -28,6 +29,10 @@ public void CompositeActivityProcessor_BadArgs()
         {
             Assert.Throws<ArgumentNullException>(() => new CompositeActivityProcessor(null));
             Assert.Throws<ArgumentException>(() => new CompositeActivityProcessor(new ActivityProcessor[0]));
+
+            using var p1 = new TestActivityProcessor(null, null);
+            using var processor = new CompositeActivityProcessor(new[] { p1 });
+            Assert.Throws<ArgumentNullException>(() => processor.AddProcessor(null));
         }
 
         [Fact]
@@ -83,17 +88,29 @@ public void CompositeActivityProcessor_ShutsDownAll()
             }
         }
 
-        [Fact]
-        public void CompositeActivityProcessor_ForceFlush()
+        [Theory]
+        [InlineData(Timeout.Infinite)]
+        [InlineData(0)]
+        [InlineData(1)]
+        public void CompositeActivityProcessor_ForceFlush(int timeout)
         {
             using var p1 = new TestActivityProcessor(null, null);
             using var p2 = new TestActivityProcessor(null, null);
 
             using (var processor = new CompositeActivityProcessor(new[] { p1, p2 }))
             {
-                processor.ForceFlush();
-                Assert.True(p1.ForceFlushCalled);
-                Assert.True(p2.ForceFlushCalled);
+                processor.ForceFlush(timeout);
+
+                if (timeout != 0)
+                {
+                    Assert.True(p1.ForceFlushCalled);
+                    Assert.True(p2.ForceFlushCalled);
+                }
+                else
+                {
+                    Assert.False(p1.ForceFlushCalled);
+                    Assert.False(p2.ForceFlushCalled);
+                }
             }
         }
     }

test/OpenTelemetry.Tests/Trace/ReentrantExportActivityProcessorTest.cs

@@ -0,0 +1,85 @@
+// <copyright file="ReentrantExportActivityProcessorTest.cs" company="OpenTelemetry Authors">
+// Copyright The OpenTelemetry Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+// </copyright>
+
+using System;
+using System.Diagnostics;
+using System.Threading;
+using OpenTelemetry.Tests.Shared;
+using OpenTelemetry.Trace;
+using Xunit;
+
+namespace OpenTelemetry.Tests.Trace
+{
+    public class ReentrantExportActivityProcessorTest
+    {
+        [Fact]
+        public void CheckNullExporter()
+        {
+            Assert.Throws<ArgumentNullException>(() => new ReentrantExportActivityProcessor(null));
+        }
+
+        [Fact]
+        public void CheckExportedOnEnd()
+        {
+            using var exporter = new TestActivityExporter();
+            using var processor = new ReentrantExportActivityProcessor(exporter);
+
+            processor.OnEnd(new Activity("start1"));
+            Assert.Single(exporter.Exported);
+
+            processor.OnEnd(new Activity("start2"));
+            Assert.Equal(2, exporter.Exported.Count);
+        }
+
+        [Theory]
+        [InlineData(Timeout.Infinite)]
+        [InlineData(0)]
+        [InlineData(1)]
+        public void CheckForceFlushExport(int timeout)
+        {
+            using var exporter = new TestActivityExporter();
+            using var processor = new ReentrantExportActivityProcessor(exporter);
+
+            processor.OnEnd(new Activity("start1"));
+            processor.OnEnd(new Activity("start2"));
+
+            // checking before force flush
+            Assert.Equal(2, exporter.Exported.Count);
+
+            // forcing flush
+            processor.ForceFlush(timeout);
+            Assert.Equal(2, exporter.Exported.Count);
+        }
+
+        [Theory]
+        [InlineData(Timeout.Infinite)]
+        [InlineData(0)]
+        [InlineData(1)]
+        public void CheckShutdownExport(int timeout)
+        {
+            using var exporter = new TestActivityExporter();
+            using var processor = new ReentrantExportActivityProcessor(exporter);
+
+            processor.OnEnd(new Activity("start"));
+
+            // checking before shutdown
+            Assert.Single(exporter.Exported);
+
+            processor.Shutdown(timeout);
+            Assert.Single(exporter.Exported);
+        }
+    }
+}