Refactor PgMqIntegrationTests to improve readability and consistency

Author: doxthree

test/Motor.Extensions.Hosting.PgMq_IntegrationTest/PgMqIntegrationTests.cs

@@ -18,16 +18,10 @@
 namespace Motor.Extensions.Hosting.PgMq_IntegrationTest;
 
 [Collection("PgMqMessage")]
-public class PgMqIntegrationTests : IClassFixture<PostgresFixture>
+public class PgMqIntegrationTests(PostgresFixture fixture) : IClassFixture<PostgresFixture>
 {
-    private readonly IRandomizerString _randomizerString;
-    private readonly NpgsqlConnectionStringBuilder _pgConnectionStringBuilder;
-
-    public PgMqIntegrationTests(PostgresFixture fixture)
-    {
-        _randomizerString = RandomizerFactory.GetRandomizer(new FieldOptionsTextRegex { Pattern = @"^[a-z]{10}" });
-        _pgConnectionStringBuilder = new NpgsqlConnectionStringBuilder(fixture.ConnectionString);
-    }
+    private readonly IRandomizerString _randomizerString = RandomizerFactory.GetRandomizer(new FieldOptionsTextRegex { Pattern = @"^[a-z]{10}" });
+    private readonly NpgsqlConnectionStringBuilder _pgConnectionStringBuilder = new(fixture.ConnectionString);
 
     [Fact(Timeout = 50000)]
     public async Task Consume_ProduceAndConsumeProtocolFormat_ConsumedDataEqualsPublished()
@@ -47,15 +41,14 @@ public async Task Consume_ProduceAndConsumeProtocolFormat_ConsumedDataEqualsPubl
         using var cts = new CancellationTokenSource();
         var executeTask = consumer.ExecuteAsync(cts.Token);
         await producer.PublishMessageAsync(
-            MotorCloudEvent.CreateTestCloudEvent(Encoding.UTF8.GetBytes(expectedMessage))
-        );
+            MotorCloudEvent.CreateTestCloudEvent(Encoding.UTF8.GetBytes(expectedMessage)), cts.Token);
         var received =
             await Task.WhenAny(tcs.Task, executeTask, Task.Delay(TimeSpan.FromSeconds(30))) == tcs.Task
                 ? await tcs.Task
                 : null;
         await cts.CancelAsync();
         await executeTask;
-        await consumer.StopAsync();
+        await consumer.StopAsync(cts.Token);
         Assert.NotNull(received);
         Assert.Equal(expectedMessage, Encoding.UTF8.GetString(received!));
     }
@@ -108,11 +101,11 @@ public async Task Consume_CallbackReturnsFailureStatus_StopsApplicationAndLogsEr
         consumer.ConsumeCallbackAsync = (_, _) => Task.FromResult(ProcessedMessageStatus.CriticalFailure);
         using var cts = new CancellationTokenSource();
         var executeTask = consumer.ExecuteAsync(cts.Token);
-        await producer.PublishMessageAsync(MotorCloudEvent.CreateTestCloudEvent(Encoding.UTF8.GetBytes("fail-me")));
-        await Task.WhenAny(executeTask, Task.Delay(TimeSpan.FromSeconds(30)));
+        await producer.PublishMessageAsync(MotorCloudEvent.CreateTestCloudEvent(Encoding.UTF8.GetBytes("fail-me")), cts.Token);
+        await Task.WhenAny(executeTask, Task.Delay(TimeSpan.FromSeconds(30), cts.Token));
         await cts.CancelAsync();
         await executeTask;
-        await consumer.StopAsync();
+        await consumer.StopAsync(cts.Token);
         lifetimeMock.Verify(x => x.StopApplication(), Times.AtLeastOnce);
     }
 
@@ -129,11 +122,11 @@ public async Task Consume_UnhandledExceptionInCallback_StopsApplicationAndLogsCr
         consumer.ConsumeCallbackAsync = (_, _) => throw new InvalidOperationException("boom");
         using var cts = new CancellationTokenSource();
         var executeTask = consumer.ExecuteAsync(cts.Token);
-        await producer.PublishMessageAsync(MotorCloudEvent.CreateTestCloudEvent(Encoding.UTF8.GetBytes("boom")));
-        await Task.WhenAny(executeTask, Task.Delay(TimeSpan.FromSeconds(30)));
+        await producer.PublishMessageAsync(MotorCloudEvent.CreateTestCloudEvent(Encoding.UTF8.GetBytes("boom")), cts.Token);
+        await Task.WhenAny(executeTask, Task.Delay(TimeSpan.FromSeconds(30), cts.Token));
         await cts.CancelAsync();
         await executeTask;
-        await consumer.StopAsync();
+        await consumer.StopAsync(cts.Token);
         lifetimeMock.Verify(x => x.StopApplication(), Times.AtLeastOnce);
         loggerMock.Verify(
             x =>
@@ -165,7 +158,9 @@ public async Task Consume_MultipleMessages_AllConsumedSequentiallyAndAcknowledge
             {
                 received.Add(Encoding.UTF8.GetString(evt.TypedData));
                 if (received.Count == messageCount)
+                {
                     tcs.TrySetResult();
+                }
             }
             return Task.FromResult(ProcessedMessageStatus.Success);
         };
@@ -177,7 +172,7 @@ await producer.PublishMessageAsync(
         }
         using var cts = new CancellationTokenSource();
         var executeTask = consumer.ExecuteAsync(cts.Token);
-        await Task.WhenAny(tcs.Task, executeTask, Task.Delay(TimeSpan.FromSeconds(45)));
+        await Task.WhenAny(tcs.Task, executeTask, Task.Delay(TimeSpan.FromSeconds(45), cts.Token));
         Assert.Equal(messageCount, received.Count);
         await cts.CancelAsync();
         await executeTask;
@@ -198,7 +193,7 @@ public async Task Consume_TokenCancelledDuringPoll_ExitsGracefully()
         await consumer.StartAsync();
         using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(2));
         await consumer.ExecuteAsync(cts.Token);
-        await consumer.StopAsync();
+        await consumer.StopAsync(cts.Token);
         lifetimeMock.Verify(x => x.StopApplication(), Times.Never);
     }
 
@@ -258,16 +253,16 @@ public async Task Consume_SuccessfulMessage_ProtocolFormat_CloudEventAttributesR
             null,
             null
         );
-        await producer.PublishMessageAsync(cloudEvent);
+        await producer.PublishMessageAsync(cloudEvent, cts.Token);
         var receivedEvent =
-            await Task.WhenAny(tcs.Task, executeTask, Task.Delay(TimeSpan.FromSeconds(30))) == tcs.Task
+            await Task.WhenAny(tcs.Task, executeTask, Task.Delay(TimeSpan.FromSeconds(30), cts.Token)) == tcs.Task
                 ? await tcs.Task
                 : null;
         await cts.CancelAsync();
         await executeTask;
-        await consumer.StopAsync();
+        await consumer.StopAsync(cts.Token);
         Assert.NotNull(receivedEvent);
-        Assert.Equal(expectedId, receivedEvent!.Id);
+        Assert.Equal(expectedId, receivedEvent.Id);
         Assert.Equal(expectedSource, receivedEvent.Source);
     }
 
@@ -323,7 +318,7 @@ options with
             Password = _pgConnectionStringBuilder.Password ?? string.Empty,
         };
 
-    private IApplicationNameService GetApplicationNameService(string source = "motor://test")
+    private static IApplicationNameService GetApplicationNameService(string source = "motor://test")
     {
         var mock = new Mock<IApplicationNameService>();
         mock.Setup(t => t.GetSource()).Returns(new Uri(source));

test/Motor.Extensions.Hosting.PgMq_IntegrationTest/PostgresFixture.cs

@@ -9,7 +9,7 @@ public class PgMqMessageCollection : ICollectionFixture<PostgresFixture> { }
 
 public class PostgresFixture : IAsyncLifetime
 {
-    private readonly PostgreSqlContainer _container = new PostgreSqlBuilder("ghcr.io/pgmq/pg18-pgmq:latest").Build();
+    private readonly PostgreSqlContainer _container = new PostgreSqlBuilder("ghcr.io/pgmq/pg18-pgmq:v1.11.1").Build();
 
     // Get ConnectionString from TestContainer.
     public string ConnectionString => _container.GetConnectionString();