Increase async test coverage

BrighterCommand · Jan 10, 2024 · 46cd384 · 46cd384
1 parent 79e086d
commit 46cd384
Show file tree

Hide file tree

Showing 4 changed files with 164 additions and 6 deletions.
diff --git a/.../Paramore.Brighter.Core.Tests/MessageDispatch/When_an_unacceptable_message_is_recieved.cs b/.../Paramore.Brighter.Core.Tests/MessageDispatch/When_an_unacceptable_message_is_recieved.cs
@@ -38,12 +38,11 @@ public class MessagePumpUnacceptableMessageTests
     {
         private readonly IAmAMessagePump _messagePump;
         private readonly FakeChannel _channel;
-        private readonly SpyRequeueCommandProcessor _commandProcessor;
 
         public MessagePumpUnacceptableMessageTests()
         {
-            _commandProcessor = new SpyRequeueCommandProcessor();
-            var provider = new CommandProcessorProvider(_commandProcessor);
+            SpyRequeueCommandProcessor commandProcessor = new();
+            var provider = new CommandProcessorProvider(commandProcessor);
             _channel = new FakeChannel();
             var messageMapperRegistry = new MessageMapperRegistry(
                 new SimpleMessageMapperFactory(_ => new MyEventMessageMapper()),

diff --git a/...ore.Brighter.Core.Tests/MessageDispatch/When_an_unacceptable_message_is_recieved_async.cs b/...ore.Brighter.Core.Tests/MessageDispatch/When_an_unacceptable_message_is_recieved_async.cs
@@ -0,0 +1,78 @@
+#region Licence
+/* The MIT License (MIT)
+Copyright © 2014 Ian Cooper <[email protected]>
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the “Software”), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in
+all copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED “AS IS”, WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+THE SOFTWARE. */
+
+#endregion
+
+using System;
+using System.Text.Json;
+using System.Threading.Tasks;
+using FluentAssertions;
+using Paramore.Brighter.Core.Tests.CommandProcessors.TestDoubles;
+using Paramore.Brighter.Core.Tests.MessageDispatch.TestDoubles;
+using Xunit;
+using Paramore.Brighter.ServiceActivator;
+using Paramore.Brighter.ServiceActivator.TestHelpers;
+
+namespace Paramore.Brighter.Core.Tests.MessageDispatch
+{
+    public class AsyncMessagePumpUnacceptableMessageTests
+    {
+        private readonly IAmAMessagePump _messagePump;
+        private readonly FakeChannel _channel;
+
+        public AsyncMessagePumpUnacceptableMessageTests()
+        {
+            SpyRequeueCommandProcessor commandProcessor = new();
+            var provider = new CommandProcessorProvider(commandProcessor);
+            _channel = new FakeChannel();
+            var messageMapperRegistry = new MessageMapperRegistry(
+                null,
+                new SimpleMessageMapperFactoryAsync(_ => new MyEventMessageMapperAsync()));
+            messageMapperRegistry.RegisterAsync<MyEvent, MyEventMessageMapperAsync>();
+
+            _messagePump = new MessagePumpAsync<MyEvent>(provider, messageMapperRegistry, null)
+            {
+                Channel = _channel, TimeoutInMilliseconds = 5000, RequeueCount = 3
+            };
+
+            var myMessage = JsonSerializer.Serialize(new MyEvent());
+            var unacceptableMessage = new Message(new MessageHeader(Guid.NewGuid(), "MyTopic", MessageType.MT_UNACCEPTABLE), new MessageBody(myMessage));
+
+            _channel.Enqueue(unacceptableMessage);
+        }
+
+        [Fact]
+        public async Task When_An_Unacceptable_Message_Is_Recieved()
+        {
+            var task = Task.Factory.StartNew(() => _messagePump.Run(), TaskCreationOptions.LongRunning);
+            await Task.Delay(1000);
+
+            var quitMessage = new Message(new MessageHeader(Guid.Empty, "", MessageType.MT_QUIT), new MessageBody(""));
+            _channel.Enqueue(quitMessage);
+
+            await Task.WhenAll(new[] { task });
+
+            //should_acknowledge_the_message
+            _channel.AcknowledgeHappened.Should().BeTrue();
+        }
+    }
+}
diff --git a/...more.Brighter.Core.Tests/MessageDispatch/When_an_unacceptable_message_limit_is_reached.cs b/...more.Brighter.Core.Tests/MessageDispatch/When_an_unacceptable_message_limit_is_reached.cs
@@ -37,12 +37,11 @@ public class MessagePumpUnacceptableMessageLimitBreachedTests
     {
         private readonly IAmAMessagePump _messagePump;
         private readonly FakeChannel _channel;
-        private readonly SpyRequeueCommandProcessor _commandProcessor;
 
         public MessagePumpUnacceptableMessageLimitBreachedTests()
         {
-            _commandProcessor = new SpyRequeueCommandProcessor();
-            var provider = new CommandProcessorProvider(_commandProcessor);
+            SpyRequeueCommandProcessor commandProcessor = new();
+            var provider = new CommandProcessorProvider(commandProcessor);
 
             _channel = new FakeChannel();
             var messageMapperRegistry = new MessageMapperRegistry(

diff --git a/...righter.Core.Tests/MessageDispatch/When_an_unacceptable_message_limit_is_reached_async.cs b/...righter.Core.Tests/MessageDispatch/When_an_unacceptable_message_limit_is_reached_async.cs
@@ -0,0 +1,82 @@
+#region Licence
+/* The MIT License (MIT)
+Copyright © 2014 Ian Cooper <[email protected]>
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the “Software”), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in
+all copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED “AS IS”, WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+THE SOFTWARE. */
+
+#endregion
+
+using System;
+using System.Threading.Tasks;
+using FluentAssertions;
+using Paramore.Brighter.Core.Tests.CommandProcessors.TestDoubles;
+using Paramore.Brighter.Core.Tests.MessageDispatch.TestDoubles;
+using Xunit;
+using Paramore.Brighter.ServiceActivator;
+using Paramore.Brighter.ServiceActivator.TestHelpers;
+
+namespace Paramore.Brighter.Core.Tests.MessageDispatch
+{
+    public class AsyncMessagePumpUnacceptableMessageLimitBreachedTests
+    {
+        private readonly IAmAMessagePump _messagePump;
+        private readonly FakeChannel _channel;
+
+        public AsyncMessagePumpUnacceptableMessageLimitBreachedTests()
+        {
+            SpyRequeueCommandProcessor commandProcessor = new();
+            var provider = new CommandProcessorProvider(commandProcessor);
+
+            _channel = new FakeChannel();
+            var messageMapperRegistry = new MessageMapperRegistry(
+                null,
+                new SimpleMessageMapperFactoryAsync(_ => new MyEventMessageMapperAsync()));
+            messageMapperRegistry.RegisterAsync<MyEvent, MyEventMessageMapperAsync>();
+
+            _messagePump = new MessagePumpBlocking<MyEvent>(provider, messageMapperRegistry, null)
+            {
+                Channel = _channel, TimeoutInMilliseconds = 5000, RequeueCount = 3, UnacceptableMessageLimit = 3
+            };
+
+            var unacceptableMessage1 = new Message(new MessageHeader(Guid.NewGuid(), "MyTopic", MessageType.MT_UNACCEPTABLE), new MessageBody(""));
+            var unacceptableMessage2 = new Message(new MessageHeader(Guid.NewGuid(), "MyTopic", MessageType.MT_UNACCEPTABLE), new MessageBody(""));
+            var unacceptableMessage3 = new Message(new MessageHeader(Guid.NewGuid(), "MyTopic", MessageType.MT_UNACCEPTABLE), new MessageBody(""));
+            var unacceptableMessage4 = new Message(new MessageHeader(Guid.NewGuid(), "MyTopic", MessageType.MT_UNACCEPTABLE), new MessageBody(""));
+
+            _channel.Enqueue(unacceptableMessage1);
+            _channel.Enqueue(unacceptableMessage2);
+            _channel.Enqueue(unacceptableMessage3);
+            _channel.Enqueue(unacceptableMessage4);
+        }
+
+        [Fact]
+        public async Task When_An_Unacceptable_Message_Limit_Is_Reached()
+        {
+            var task = Task.Factory.StartNew(() => _messagePump.Run(), TaskCreationOptions.LongRunning);
+
+            await Task.WhenAll(task);
+
+            //should_have_acknowledge_the_3_messages
+            _channel.AcknowledgeCount.Should().Be(3);
+            //should_dispose_the_input_channel
+            _channel.DisposeHappened.Should().BeTrue();
+        }
+    }
+}
+