feat(ServiceBus): added unit tests

Added some tests that should test the general behavior of ServiceBus.

Author: bruno-brant

src/Liquid.Activation/Worker/LightWorker.cs

@@ -25,7 +25,7 @@ public abstract class LightWorker : LightBackgroundTask, ILightWorker
         protected readonly static Dictionary<MethodInfo, QueueAttribute> _queues = new Dictionary<MethodInfo, QueueAttribute>();
         protected readonly static Dictionary<MethodInfo, TopicAttribute> _topics = new Dictionary<MethodInfo, TopicAttribute>();
         private readonly List<string> _inputValidationErrors = new List<string>();
-        protected ILightTelemetry Telemetry { get; } = Workbench.Instance.Telemetry != null ? (ILightTelemetry)Workbench.Instance.Telemetry.CloneService() : null;
+        protected ILightTelemetry Telemetry => Workbench.Instance.Telemetry;
         protected ILightCache Cache => Workbench.Instance.Cache;
         //Instance of CriticHandler to inject on the others classes
         private readonly CriticHandler _criticHandler = new CriticHandler();

src/Liquid.OnAzure/Liquid.OnAzure.csproj

@@ -15,6 +15,9 @@
     <DebugType>full</DebugType>
     <DebugSymbols>true</DebugSymbols>
   </PropertyGroup>
+  <ItemGroup>
+    <AdditionalFiles Include="..\..\stylecop.json" Link="stylecop.json" />
+  </ItemGroup>
   <ItemGroup>
     <Compile Remove="Telemetry\AppInsightsTelemetryMiddleware.cs" />
   </ItemGroup>
@@ -29,6 +32,10 @@
     <PackageReference Include="Microsoft.CodeAnalysis.FxCopAnalyzers" Version="2.9.3" />
     <PackageReference Include="Microsoft.Extensions.Caching.Abstractions" Version="2.2.0" />
     <PackageReference Include="Microsoft.Extensions.Caching.Redis" Version="2.2.0" />
+    <PackageReference Include="StyleCop.Analyzers" Version="1.2.0-beta.113">
+      <PrivateAssets>all</PrivateAssets>
+      <IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets>
+    </PackageReference>
     <PackageReference Include="WindowsAzure.Storage" Version="9.3.3" />
   </ItemGroup>
   <ItemGroup>

src/Liquid.OnAzure/MessageBuses/ServiceBus.cs

@@ -1,23 +1,125 @@
-using Liquid.Base.Interfaces;
+// Copyright (c) Avanade Inc. All rights reserved.
+// Licensed under the MIT License. See LICENSE in the project root for license information.
+
+using System;
+using System.Collections.Generic;
+using System.Reflection;
+using System.Threading.Tasks;
+using Liquid.Activation;
 using Liquid.Domain;
 using Liquid.Domain.Base;
-using Liquid.Activation;
 using Liquid.Runtime.Configuration.Base;
 using Liquid.Runtime.Telemetry;
 using Microsoft.Azure.ServiceBus;
-using System;
-using System.Collections.Generic;
-using System.Reflection;
-using System.Text;
-using System.Threading.Tasks;
 
 namespace Liquid.OnAzure
 {
     /// <summary>
-    /// Implementation of the communication component between queues and topics of the Azure, this class is specific to azure
+    /// Defines an object capable of creating instances of <see cref="IQueueClient"/>.
+    /// </summary>
+    public interface IQueueClientFactory
+    {
+        /// <summary>
+        /// Creates a new instance of <see cref="IQueueClient"/>.
+        /// </summary>
+        /// <param name="connectionString">The connection string for the client.</param>
+        /// <param name="queueName">The name of the queue.</param>
+        /// <param name="receiveMode">The receive mode that the client will connect to the queue.</param>
+        /// <returns>A new instance of <see cref="IQueueClient"/>.</returns>
+        IQueueClient CreateClient(string connectionString, string queueName, ReceiveMode receiveMode);
+    }
+
+    /// <summary>
+    /// Defines an object capable of creating instances of <see cref="ISubscriptionClient"/>.
+    /// </summary>
+    public interface ISubscriptionClientFactory
+    {
+        /// <summary>
+        /// Creates a new instance of <see cref="ISubscriptionClient"/>.
+        /// </summary>
+        /// <param name="connectionString">The connection string for the client.</param>
+        /// <param name="topicName">The name of the topic to connect to.</param>
+        /// <param name="subscriptionName">Identifies the subscription to this topic.</param>
+        /// <param name="receiveMode">The receive mode that the client will connect to the queue.</param>
+        /// <returns>A new instance of <see cref="ISubscriptionClient"/>.</returns>
+        ISubscriptionClient CreateClient(string connectionString, string topicName, string subscriptionName, ReceiveMode receiveMode);
+    }
+
+    /// <summary>
+    /// Configuration source for <see cref="ServiceBus"/>.
+    /// </summary>
+    // TODO: should remove this class once we move to .NET configuration system
+    public interface IServiceBusConfigurationProvider
+    {
+        /// <summary>
+        /// Gets the configuration for a <see cref="ServiceBus"/>.
+        /// </summary>
+        /// <returns>
+        /// The current configuration for a service bus.
+        /// </returns>
+        ServiceBusConfiguration GetConfiguration();
+
+        /// <summary>
+        /// Gets the configuration for a <see cref="ServiceBus"/>.
+        /// </summary>
+        /// <param name="connectionName">
+        /// Identifies which connection should be retrieved from the file.
+        /// </param>
+        /// <returns>
+        /// The current configuration for a service bus.
+        /// </returns>
+        ServiceBusConfiguration GetConfiguration(string connectionName);
+    }
+
+    /// <summary>
+    /// Implementation of the communication component between queues and topics of the Azure, this class is specific to azure.
     /// </summary>
     public class ServiceBus : LightWorker, IWorkbenchService
     {
+        /// <summary>
+        /// Factory used to create a <see cref="IQueueClient"/>.
+        /// </summary>
+        private readonly IQueueClientFactory _queueClientFactory = new DefaultQueueClientFactory();
+
+        /// <summary>
+        /// Factory used to create a <see cref="ISubscriptionClient"/>.
+        /// </summary>
+        private readonly ISubscriptionClientFactory _subscriptionClientFactory = new DefaultSubscriptionClientFactory();
+
+        /// <summary>
+        /// Service that retrives a <see cref="ServiceBusConfiguration"/>.
+        /// </summary>
+        private readonly IServiceBusConfigurationProvider _configurationProvider = new DefaultServiceBusConfigurationProvider();
+
+        /// <summary>
+        /// Initializes a new instance of the <see cref="ServiceBus"/> class.
+        /// </summary>
+        public ServiceBus()
+        {
+        }
+
+        /// <summary>
+        /// Initializes a new instance of the <see cref="ServiceBus"/> class.
+        /// </summary>
+        /// <param name="queueClientFactory">
+        /// Dependency. Used to obtain new instances of a <see cref="IQueueClient"/>.
+        /// </param>
+        /// <param name="subscriptionClientFactory">
+        /// Dependency. Used to obtain new instances of a <see cref="ISubscriptionClient"/>.
+        /// </param>
+        /// <param name="configurationProvider">
+        /// Dependency. Used to retrieve a configuration for this class.
+        /// </param>
+        public ServiceBus(
+            IQueueClientFactory queueClientFactory,
+            ISubscriptionClientFactory subscriptionClientFactory,
+            IServiceBusConfigurationProvider configurationProvider)
+        {
+            _queueClientFactory = queueClientFactory ?? throw new ArgumentNullException(nameof(queueClientFactory));
+            _subscriptionClientFactory = subscriptionClientFactory ?? throw new ArgumentNullException(nameof(subscriptionClientFactory));
+            _configurationProvider = configurationProvider ?? throw new ArgumentNullException(nameof(configurationProvider));
+        }
+
         /// <summary>
         /// Implementation of the start process queue and process topic. It must be called  parent before start processes.
         /// </summary>
@@ -36,22 +138,22 @@ public override void Initialize()
         /// <returns>StringConnection of the ServiceBus</returns>
         private string GetConnection<T>(KeyValuePair<MethodInfo, T> item)
         {
-            MethodInfo method = item.Key;
-            string connectionKey = GetKeyConnection(method);
-            ServiceBusConfiguration config = null;
+            var method = item.Key;
+            var connectionKey = GetKeyConnection(method);
+
+            ServiceBusConfiguration config;
             if (string.IsNullOrEmpty(connectionKey)) // Load specific settings if provided
             {
-                config = LightConfigurator.Config<ServiceBusConfiguration>($"{nameof(ServiceBus)}");
+                config = _configurationProvider.GetConfiguration();//LightConfigurator.Config<ServiceBusConfiguration>($"{nameof(ServiceBus)}");
             }
             else
             {
-                config = LightConfigurator.Config<ServiceBusConfiguration>($"{nameof(ServiceBus)}_{connectionKey}");
+                config = _configurationProvider.GetConfiguration(connectionKey);//LightConfigurator.Config<ServiceBusConfiguration>($"{nameof(ServiceBus)}_{connectionKey}");
             }
 
             return config.ConnectionString;
         }
 
-
         /// <summary>
         /// If  an error occurs in the processing, this method going to called
         /// </summary>
@@ -80,7 +182,7 @@ public void ProcessQueue()
                     int takeQuantity = queue.Value.TakeQuantity;
 
                     //Register Trace on the telemetry 
-                    QueueClient queueReceiver = new QueueClient(GetConnection(queue), queueName, receiveMode);
+                    var queueReceiver = _queueClientFactory.CreateClient(GetConnection(queue), queueName, receiveMode);
 
                     //Register the method to process receive message
                     //The RegisterMessageHandler is validate for all register exist on the queue, without need loop for items
@@ -96,7 +198,7 @@ public void ProcessQueue()
                             {
                                 Exception moreInfo = new Exception($"Exception reading message from queue {queueName}. See inner exception for details. Message={exRegister.Message}", exRegister);
                                 //Use the class instead of interface because tracking exceptions directly is not supposed to be done outside AMAW (i.e. by the business code)
-                                ((LightTelemetry)Workbench.Instance.Telemetry).TrackException(moreInfo);
+                                (Workbench.Instance.Telemetry as LightTelemetry)?.TrackException(moreInfo);
 
                                 //If there is a error , set DeadLetter on register
                                 if (queueReceiver.ReceiveMode == ReceiveMode.PeekLock)
@@ -114,7 +216,7 @@ await queueReceiver.DeadLetterAsync(message.SystemProperties.LockToken,
             {
                 Exception moreInfo = new Exception($"Error setting up queue consumption from service bus. See inner exception for details. Message={exception.Message}", exception);
                 //Use the class instead of interface because tracking exceptions directly is not supposed to be done outside AMAW (i.e. by the business code)
-                ((LightTelemetry)Workbench.Instance.Telemetry).TrackException(moreInfo);
+                (Workbench.Instance.Telemetry as LightTelemetry)?.TrackException(moreInfo);
             }
         }
 
@@ -132,14 +234,17 @@ private void ProcessSubscription()
                     string topicName = topic.Value.TopicName;
                     string subscriptName = topic.Value.Subscription;
                     ReceiveMode receiveMode = ReceiveMode.PeekLock;
+
                     if (topic.Value.DeleteAfterRead)
                     {
                         receiveMode = ReceiveMode.ReceiveAndDelete;
                     }
+
                     int takeQuantity = topic.Value.TakeQuantity;
 
                     //Register Trace on the telemetry 
-                    SubscriptionClient subscriptionClient = new SubscriptionClient(GetConnection(topic), topicName, subscriptName, receiveMode, null);
+                    var subscriptionClient = _subscriptionClientFactory.CreateClient(
+                        GetConnection(topic), topicName, subscriptName, receiveMode);
 
                     //Register the method to process receive message
                     //The RegisterMessageHandler is validate for all register exist on the queue, without need loop for items
@@ -155,7 +260,7 @@ private void ProcessSubscription()
                             {
                                 Exception moreInfo = new Exception($"Exception reading message from topic {topicName} and subscriptName {subscriptName}. See inner exception for details. Message={exRegister.Message}", exRegister);
                                 //Use the class instead of interface because tracking exceptions directly is not supposed to be done outside AMAW (i.e. by the business code)
-                                ((LightTelemetry)Workbench.Instance.Telemetry).TrackException(moreInfo);
+                                (Workbench.Instance.Telemetry as LightTelemetry)?.TrackException(moreInfo);
 
                                 var exceptionDetails = $"{exRegister.Message}";
 
@@ -200,13 +305,57 @@ await subscriptionClient.DeadLetterAsync(message.SystemProperties.LockToken,
             {
                 Exception moreInfo = new Exception($"Error setting up subscription consumption from service bus. See inner exception for details. Message={exception.Message}", exception);
                 //Use the class instead of interface because tracking exceptions directly is not supposed to be done outside AMAW (i.e. by the business code)
-                ((LightTelemetry)Workbench.Instance.Telemetry).TrackException(moreInfo);
+                (Workbench.Instance.Telemetry as LightTelemetry)?.TrackException(moreInfo);
             }
         }
 
         protected override Task ProcessAsync()
         {
             throw new NotImplementedException();
         }
+
+        /// <summary>
+        /// Default implementation for <see cref="IQueueClientFactory"/>,
+        /// creates instances of <see cref="IQueueClient"/>.
+        /// </summary>
+        private class DefaultQueueClientFactory : IQueueClientFactory
+        {
+            /// <inheritdoc/>
+            public IQueueClient CreateClient(string connectionString, string queueName, ReceiveMode receiveMode)
+            {
+                return new QueueClient(connectionString, queueName, receiveMode);
+            }
+        }
+
+        /// <summary>
+        /// Default implementation for <see cref="ISubscriptionClientFactory"/>,
+        /// creates instances of <see cref="SubscriptionClient"/>.
+        /// </summary>
+        private class DefaultSubscriptionClientFactory : ISubscriptionClientFactory
+        {
+            /// <inheritdoc/>
+            public ISubscriptionClient CreateClient(string connectionString, string topicName, string subscriptionName, ReceiveMode mode)
+            {
+                return new SubscriptionClient(connectionString, topicName, subscriptionName, mode, null);
+            }
+        }
+
+        /// <summary>
+        /// Retrieves configuration using <see cref="LightConfigurator"/>.
+        /// </summary>
+        private class DefaultServiceBusConfigurationProvider : IServiceBusConfigurationProvider
+        {
+            /// <inheritdoc/>
+            public ServiceBusConfiguration GetConfiguration()
+            {
+                return LightConfigurator.Config<ServiceBusConfiguration>($"{nameof(ServiceBus)}");
+            }
+
+            /// <inheritdoc/>
+            public ServiceBusConfiguration GetConfiguration(string connectionKey)
+            {
+                return LightConfigurator.Config<ServiceBusConfiguration>($"{nameof(ServiceBus)}_{connectionKey}");
+            }
+        }
     }
 }

test/Liquid.OnAzure.Tests/AzureBlobTests.cs

@@ -36,8 +36,6 @@ public class AzureBlobTests : IDisposable
 
         public AzureBlobTests()
         {
-            Workbench.Instance.Reset();
-
             Workbench.Instance.AddToCache(WorkbenchServiceType.Repository, _fakeLightRepository);
 
             _sut = new AzureBlob(new MediaStorageConfiguration
@@ -156,6 +154,8 @@ protected virtual void Dispose(bool isDisposing)
             if (isDisposing)
             {
                 _stream?.Dispose();
+
+                Workbench.Instance.Reset();
             }
         }
 

test/Liquid.OnAzure.Tests/Liquid.OnAzure.Tests.csproj

@@ -13,6 +13,10 @@
   <ItemGroup>
     <PackageReference Include="AutoFixture.AutoNSubstitute" Version="4.11.0" />
     <PackageReference Include="AutoFixture.Xunit2" Version="4.11.0" />
+    <PackageReference Include="Microsoft.CodeAnalysis.FxCopAnalyzers" Version="2.9.8">
+      <PrivateAssets>all</PrivateAssets>
+      <IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets>
+    </PackageReference>
     <PackageReference Include="Microsoft.NET.Test.Sdk" Version="16.2.0" />
     <PackageReference Include="StyleCop.Analyzers" Version="1.2.0-beta.113">
       <PrivateAssets>all</PrivateAssets>