add tests for all messages

dominikus1993 · May 25, 2024 · 13b069c · 13b069c
1 parent 38ac94f
commit 13b069c
Show file tree

Hide file tree

Showing 2 changed files with 35 additions and 1 deletion.
diff --git a/tests/BuildingBlocks/Messaging.Tests/Extensions/RabbitMqTestConsumer.cs b/tests/BuildingBlocks/Messaging.Tests/Extensions/RabbitMqTestConsumer.cs
@@ -77,7 +77,7 @@ internal RabbitMqTestConsumer(RabbitMqMessageConsumer<T> consumer, Channel<T> ch
 
     public async IAsyncEnumerable<T> Consume()
     {
-        await foreach (var message in _channel.Reader.ReadAllAsync(_cancellationTokenSource.Token).WithCancellation(_cancellationTokenSource.Token))
+        await foreach (var message in _channel.Reader.ReadAllAsync(_cancellationTokenSource.Token))
         {
             yield return message;
         }

diff --git a/tests/BuildingBlocks/Messaging.Tests/Subscriber/RabbitMqMessageConsumerTests.cs b/tests/BuildingBlocks/Messaging.Tests/Subscriber/RabbitMqMessageConsumerTests.cs
@@ -61,4 +61,38 @@ public async Task TestMessageSubscription(RabbitMqPublisherConfig<Msg> config, M
         subject.Message.ShouldNotBeNullOrEmpty();
         subject.Message.ShouldBe(msg.Message);
     }
+
+    [Theory]
+    [InlineAutoData()]
+    public async Task TestMessagesSubscription(RabbitMqPublisherConfig<Msg> config, Msg[] msgs, string queueName)
+    {
+        // Arrange 
+        using var cts = new CancellationTokenSource();
+        cts.CancelAfter(TimeSpan.FromSeconds(30));
+        var publisher = new RabbitMqMessagePublisher<Msg>(_rabbitMqFixture.Bus.Advanced, config);
+        var exchange = await _rabbitMqFixture.Bus.Advanced.DeclareExchangeAsync(config, cancellationToken: cts.Token);
+        await using var consumer = await RabbitMqTestConsumer.CreateAsync(_rabbitMqFixture.Bus.Advanced,
+            new RabbitMqSubscriptionConfiguration<Msg>()
+            {
+                Exchange = exchange.Name, Topic = config.Topic, Queue = queueName
+            }, msgs.Length, cts.Token);
+        // Act
+
+        foreach (var msg in msgs)
+        {
+            await publisher.Publish(msg, cancellationToken: cts.Token);
+        }
+
+        var subject = await consumer.Consume().ToListAsync(cancellationToken: cts.Token);
+
+        subject.ShouldNotBeNull();
+        subject.ShouldNotBeEmpty();
+        subject.Count.ShouldBe(msgs.Length);
+        Assert.All(subject, msg =>
+        {
+            Assert.NotNull(msg);
+            Assert.Contains(msgs, x => x.Id == msg.Id);
+            Assert.Equivalent(msg.Message, msgs.First(x => x.Id == msg.Id).Message);
+        }); // Assert that all messages are received
+    }
 }