Merge pull request #481 from martincostello/Migrate-Publish-Integrati…

…on-Test

Migrate publish integration test

JustSaying.Extensions.DependencyInjection.Microsoft/IServiceCollectionExtensions.cs

@@ -193,5 +193,44 @@ public static IServiceCollection AddJustSayingHandler<TMessage, THandler>(this I
             services.TryAddTransient<IHandlerAsync<TMessage>, THandler>();
             return services;
         }
+
+        /// <summary>
+        /// Configures JustSaying using the specified service collection.
+        /// </summary>
+        /// <param name="services">The <see cref="IServiceCollection"/> to configure JustSaying with.</param>
+        /// <param name="configure">A delegate to a method to use to configure JustSaying.</param>
+        /// <returns>
+        /// The <see cref="IServiceCollection"/> specified by <paramref name="services"/>.
+        /// </returns>
+        /// <exception cref="ArgumentNullException">
+        /// <paramref name="services"/> or <paramref name="configure"/> is <see langword="null"/>.
+        /// </exception>
+        public static IServiceCollection ConfigureJustSaying(this IServiceCollection services, Action<MessagingBusBuilder> configure)
+        {
+            if (services == null)
+            {
+                throw new ArgumentNullException(nameof(services));
+            }
+
+            if (configure == null)
+            {
+                throw new ArgumentNullException(nameof(configure));
+            }
+
+            return services.AddSingleton<IMessageBusConfigurationContributor>(new DelegatingConfigurationContributor(configure));
+        }
+
+        private sealed class DelegatingConfigurationContributor : IMessageBusConfigurationContributor
+        {
+            private readonly Action<MessagingBusBuilder> _configure;
+
+            internal DelegatingConfigurationContributor(Action<MessagingBusBuilder> configure)
+            {
+                _configure = configure;
+            }
+
+            public void Configure(MessagingBusBuilder builder)
+                => _configure(builder);
+        }
     }
 }

JustSaying.IntegrationTests/Fluent/IntegrationTestBase.cs

@@ -0,0 +1,87 @@
+using System;
+using System.Threading;
+using System.Threading.Tasks;
+using JustSaying.Messaging;
+using JustSaying.Messaging.MessageHandling;
+using JustSaying.Models;
+using JustSaying.TestingFramework;
+using Microsoft.Extensions.DependencyInjection;
+using Microsoft.Extensions.Logging;
+using NSubstitute;
+using Xunit.Abstractions;
+
+namespace JustSaying.IntegrationTests.Fluent
+{
+    public abstract class IntegrationTestBase
+    {
+        protected IntegrationTestBase(ITestOutputHelper outputHelper)
+        {
+            OutputHelper = outputHelper;
+        }
+
+        protected virtual string AccessKeyId { get; } = "accessKeyId";
+
+        protected virtual string SecretAccessKey { get; } = "secretAccessKey";
+
+        protected virtual string SessionToken { get; } = "token";
+
+        protected ITestOutputHelper OutputHelper { get; }
+
+        protected virtual string Region => TestEnvironment.Region.SystemName;
+
+        protected virtual Uri ServiceUri => TestEnvironment.SimulatorUrl;
+
+        protected virtual TimeSpan Timeout => TimeSpan.FromSeconds(20);
+
+        protected virtual string UniqueName { get; } = $"{DateTime.UtcNow.Ticks}-integration-tests";
+
+        protected IServiceCollection GivenJustSaying()
+            => Given((_) => { });
+
+        protected IServiceCollection Given(Action<MessagingBusBuilder> configure)
+            => Given((builder, _) => configure(builder));
+
+        protected IServiceCollection Given(Action<MessagingBusBuilder, IServiceProvider> configure)
+        {
+            return new ServiceCollection()
+                .AddLogging((p) => p.AddXUnit(OutputHelper))
+                .AddJustSaying(
+                    (builder, serviceProvider) =>
+                    {
+                        builder.Messaging((options) => options.WithRegion(Region))
+                               .Client((options) =>
+                                {
+                                    options.WithSessionCredentials(AccessKeyId, SecretAccessKey, SessionToken)
+                                           .WithServiceUri(ServiceUri);
+                                });
+
+                        configure(builder, serviceProvider);
+                    });
+        }
+
+        protected IHandlerAsync<T> CreateHandler<T>(TaskCompletionSource<object> completionSource)
+            where T : Message
+        {
+            IHandlerAsync<T> handler = Substitute.For<IHandlerAsync<T>>();
+
+            handler.Handle(Arg.Any<T>())
+                   .Returns(true)
+                   .AndDoes((_) => completionSource.SetResult(null));
+
+            return handler;
+        }
+
+        protected async Task WhenAsync(IServiceCollection services, Func<IMessagePublisher, IMessagingBus, CancellationToken, Task> action)
+        {
+            IServiceProvider serviceProvider = services.BuildServiceProvider();
+
+            IMessagePublisher publisher = serviceProvider.GetRequiredService<IMessagePublisher>();
+            IMessagingBus listener = serviceProvider.GetRequiredService<IMessagingBus>();
+
+            using (var source = new CancellationTokenSource(Timeout))
+            {
+                await Task.WhenAny(action(publisher, listener, source.Token), Task.Delay(Timeout, source.Token)).ConfigureAwait(false);
+            }
+        }
+    }
+}

JustSaying.IntegrationTests/Fluent/Publishing/WhenAMessageIsPublishedToAQueue.cs

@@ -0,0 +1,53 @@
+using System;
+using System.Threading.Tasks;
+using JustSaying.Messaging;
+using JustSaying.TestingFramework;
+using Microsoft.Extensions.DependencyInjection;
+using NSubstitute;
+using Xunit.Abstractions;
+
+namespace JustSaying.IntegrationTests.Fluent.Publishing
+{
+    public class WhenAMessageIsPublishedToAQueue : IntegrationTestBase
+    {
+        public WhenAMessageIsPublishedToAQueue(ITestOutputHelper outputHelper)
+            : base(outputHelper)
+        {
+        }
+
+        [AwsFact]
+        public async Task Then_The_Message_Is_Handled()
+        {
+            // Arrange
+            var completionSource = new TaskCompletionSource<object>();
+            var handler = CreateHandler<SimpleMessage>(completionSource);
+
+            var services = GivenJustSaying()
+                .ConfigureJustSaying((builder) => builder.Publications((options) => options.WithQueue<SimpleMessage>(UniqueName)))
+                .ConfigureJustSaying((builder) => builder.Subscriptions((options) => options.ForQueue<SimpleMessage>(UniqueName)))
+                .AddSingleton(handler);
+
+            string content = Guid.NewGuid().ToString();
+
+            var message = new SimpleMessage()
+            {
+                Content = content
+            };
+
+            await WhenAsync(
+                services,
+                async (publisher, listener, cancellationToken) =>
+                {
+                    listener.Start(cancellationToken);
+
+                    // Act
+                    await publisher.PublishAsync(message, cancellationToken);
+
+                    // Assert
+                    completionSource.Task.Wait(cancellationToken);
+
+                    await handler.Received().Handle(Arg.Is<SimpleMessage>((m) => m.Content == content));
+                });
+        }
+    }
+}

JustSaying.IntegrationTests/Fluent/Publishing/WhenAMessageIsPublishedToATopic.cs

@@ -0,0 +1,53 @@
+using System;
+using System.Threading.Tasks;
+using JustSaying.Messaging;
+using JustSaying.TestingFramework;
+using Microsoft.Extensions.DependencyInjection;
+using NSubstitute;
+using Xunit.Abstractions;
+
+namespace JustSaying.IntegrationTests.Fluent.Publishing
+{
+    public class WhenAMessageIsPublishedToATopic : IntegrationTestBase
+    {
+        public WhenAMessageIsPublishedToATopic(ITestOutputHelper outputHelper)
+            : base(outputHelper)
+        {
+        }
+
+        [AwsFact]
+        public async Task Then_The_Message_Is_Handled()
+        {
+            // Arrange
+            var completionSource = new TaskCompletionSource<object>();
+            var handler = CreateHandler<SimpleMessage>(completionSource);
+
+            var services = GivenJustSaying()
+                .ConfigureJustSaying((builder) => builder.Publications((options) => options.WithTopic<SimpleMessage>()))
+                .ConfigureJustSaying((builder) => builder.Subscriptions((options) => options.ForTopic<SimpleMessage>(UniqueName)))
+                .AddSingleton(handler);
+
+            string content = Guid.NewGuid().ToString();
+
+            var message = new SimpleMessage()
+            {
+                Content = content
+            };
+
+            await WhenAsync(
+                services,
+                async (publisher, listener, cancellationToken) =>
+                {
+                    listener.Start(cancellationToken);
+
+                    // Act
+                    await publisher.PublishAsync(message, cancellationToken);
+
+                    // Assert
+                    completionSource.Task.Wait(cancellationToken);
+
+                    await handler.Received().Handle(Arg.Is<SimpleMessage>((m) => m.Content == content));
+                });
+        }
+    }
+}

JustSaying.IntegrationTests/Fluent/Publishing/WhenPublishingWithoutAMonitor.cs

@@ -0,0 +1,91 @@
+using System;
+using System.Threading;
+using System.Threading.Tasks;
+using JustSaying.Messaging;
+using JustSaying.Messaging.MessageHandling;
+using JustSaying.Models;
+using JustSaying.TestingFramework;
+using Microsoft.Extensions.DependencyInjection;
+using NSubstitute;
+using Xunit.Abstractions;
+
+namespace JustSaying.IntegrationTests.Fluent.Publishing
+{
+    public class WhenPublishingWithoutAMonitor : IntegrationTestBase
+    {
+        public WhenPublishingWithoutAMonitor(ITestOutputHelper outputHelper)
+            : base(outputHelper)
+        {
+        }
+
+        [AwsFact]
+        public async Task A_Message_Can_Still_Be_Published_To_A_Queue()
+        {
+            // Arrange
+            var completionSource = new TaskCompletionSource<object>();
+            var handler = CreateHandler<SimpleMessage>(completionSource);
+
+            IServiceCollection services = GivenJustSaying()
+                .ConfigureJustSaying((builder) => builder.Publications((publication) => publication.WithQueue<SimpleMessage>(UniqueName)))
+                .ConfigureJustSaying((builder) => builder.Subscriptions((subscription) => subscription.ForQueue<SimpleMessage>(UniqueName)))
+                .AddSingleton(handler);
+
+            // Act and Assert
+            await AssertMessagePublishedAndReceivedAsync(services, handler, completionSource);
+        }
+
+        [AwsFact]
+        public async Task A_Message_Can_Still_Be_Published_To_A_Topic()
+        {
+            // Arrange
+            var completionSource = new TaskCompletionSource<object>();
+            var handler = CreateHandler<SimpleMessage>(completionSource);
+
+            IServiceCollection services = Given(
+                (builder) =>
+                {
+                    builder.Publications((publication) => publication.WithTopic<SimpleMessage>());
+
+                    builder.Messaging(
+                        (config) => config.WithPublishFailureBackoff(TimeSpan.FromMilliseconds(1))
+                                          .WithPublishFailureReattempts(1));
+
+                    builder.Subscriptions(
+                        (subscription) => subscription.ForTopic<SimpleMessage>(
+                            (topic) => topic.WithName(UniqueName).WithReadConfiguration(
+                                (config) => config.WithInstancePosition(1))));
+                })
+                .AddSingleton(handler);
+
+            // Act and Assert
+            await AssertMessagePublishedAndReceivedAsync(services, handler, completionSource);
+        }
+
+        private async Task AssertMessagePublishedAndReceivedAsync<T>(
+            IServiceCollection services,
+            IHandlerAsync<T> handler,
+            TaskCompletionSource<object> completionSource)
+            where T : Message
+        {
+            IServiceProvider serviceProvider = services.BuildServiceProvider();
+
+            IMessagePublisher publisher = serviceProvider.GetRequiredService<IMessagePublisher>();
+            IMessagingBus listener = serviceProvider.GetRequiredService<IMessagingBus>();
+
+            using (var source = new CancellationTokenSource(Timeout))
+            {
+                listener.Start(source.Token);
+
+                var message = new SimpleMessage();
+
+                // Act
+                await publisher.PublishAsync(message, source.Token);
+
+                // Assert
+                completionSource.Task.Wait(source.Token);
+
+                await handler.Received(1).Handle(Arg.Any<T>());
+            }
+        }
+    }
+}

JustSaying.IntegrationTests/Fluent/Subscribing/WhenAHandlerThrowsAnExceptionWithAMonitor.cs

@@ -0,0 +1,55 @@
+using System;
+using System.Threading.Tasks;
+using JustSaying.IntegrationTests.TestHandlers;
+using JustSaying.Messaging;
+using JustSaying.Messaging.MessageHandling;
+using JustSaying.Messaging.Monitoring;
+using JustSaying.TestingFramework;
+using Microsoft.Extensions.DependencyInjection;
+using NSubstitute;
+using Shouldly;
+using Xunit.Abstractions;
+
+namespace JustSaying.IntegrationTests.Fluent.Subscribing
+{
+    public class WhenAHandlerThrowsAnExceptionWithAMonitor : IntegrationTestBase
+    {
+        public WhenAHandlerThrowsAnExceptionWithAMonitor(ITestOutputHelper outputHelper)
+            : base(outputHelper)
+        {
+        }
+
+        [AwsFact]
+        public async Task Then_The_Message_Is_Handled()
+        {
+            // Arrange
+            var handler = new ThrowingHandler();
+            var monitoring = Substitute.For<IMessageMonitor>();
+
+            var services = GivenJustSaying()
+                .ConfigureJustSaying((builder) => builder.Publications((options) => options.WithQueue<SimpleMessage>(UniqueName)))
+                .ConfigureJustSaying((builder) => builder.Subscriptions((options) => options.ForQueue<SimpleMessage>(UniqueName)))
+                .ConfigureJustSaying((builder) => builder.Services((options) => options.WithMessageMonitoring(() => monitoring)))
+                .AddSingleton<IHandlerAsync<SimpleMessage>>(handler);
+
+            var message = new SimpleMessage();
+
+            await WhenAsync(
+                services,
+                async (publisher, listener, cancellationToken) =>
+                {
+                    listener.Start(cancellationToken);
+
+                    // Act
+                    await publisher.PublishAsync(message, cancellationToken);
+                    await handler.DoneSignal.Task;
+
+                    // Assert
+                    await handler.Received().Handle(Arg.Any<SimpleMessage>());
+                    handler.MessageReceived.ShouldNotBeNull();
+
+                    monitoring.Received().HandleException(Arg.Any<Type>());
+                });
+        }
+    }
+}