fix: fix ci failed for tests

src/BuildingBlocks/TestBase/TestBase.cs

@@ -42,8 +42,8 @@
     where TEntryPoint : class
 {
     private readonly WebApplicationFactory<TEntryPoint> _factory;
-    private int Timeout => 300; // Second
-    private ITestHarness TestHarness => ServiceProvider?.GetTestHarness();
+    private int Timeout => 120; // Second
+    public ITestHarness TestHarness => ServiceProvider?.GetTestHarness();
     private Action<IServiceCollection> TestRegistrationServices { get; set; }
     private PostgreSqlContainer PostgresTestcontainer;
     private PostgreSqlContainer PostgresPersistTestContainer;
@@ -136,10 +136,14 @@
     {
         CancellationTokenSource = new CancellationTokenSource();
         await StartTestContainerAsync();
+
+        await TestHarness.Start();
     }
 
     public async Task DisposeAsync()
     {
+        await TestHarness.Stop();
+
         await StopTestContainerAsync();
         await _factory.DisposeAsync();
         await CancellationTokenSource.CancelAsync();
@@ -201,83 +205,71 @@
     public async Task Publish<TMessage>(TMessage message, CancellationToken cancellationToken = default)
         where TMessage : class, IEvent
     {
+        // Use harness bus to ensure publish happens only after the bus is started.
         await TestHarness.Bus.Publish(message, cancellationToken);
     }
 
     public async Task<bool> WaitForPublishing<TMessage>(CancellationToken cancellationToken = default)
         where TMessage : class, IEvent
     {
-        var result = await WaitUntilConditionMet(async () =>
-        {
-            var published = await TestHarness.Published.Any<TMessage>(cancellationToken);
-
-            return published;
-        });
-
-        return result;
-    }
-
-    public async Task<bool> WaitForConsuming<TMessage>(CancellationToken cancellationToken = default)
-        where TMessage : class, IEvent
-    {
-        var result = await WaitUntilConditionMet(async () =>
-        {
-            var consumed = await TestHarness.Consumed.Any<TMessage>(cancellationToken);
+        var result = await WaitUntilConditionMet(
+            async () =>
+            {
+                var published = await TestHarness.Published.Any<TMessage>(cancellationToken);
 
-            return consumed;
-        });
+                return published;
+            },
+            cancellationToken: cancellationToken
+        );
 
         return result;
     }
 
-    public async Task<bool> ShouldProcessedPersistInternalCommand<TInternalCommand>(
+    public Task<bool> WaitUntilAsync(
+        Func<Task<bool>> condition,
+        TimeSpan? timeout = null,
+        TimeSpan? pollInterval = null,
         CancellationToken cancellationToken = default
     )
-        where TInternalCommand : class, IInternalCommand
     {
-        var result = await WaitUntilConditionMet(async () =>
-        {
-            return await ExecuteScopeAsync(async sp =>
-            {
-                var persistMessageProcessor = sp.GetService<IPersistMessageProcessor>();
+        var effectiveTimeout = timeout ?? TimeSpan.FromSeconds(Timeout);
+        var effectivePollInterval = pollInterval ?? TimeSpan.FromMilliseconds(200);
 
-                Guard.Against.Null(persistMessageProcessor, nameof(persistMessageProcessor));
-
-                var filter = await persistMessageProcessor.GetByFilterAsync(x =>
-                    x.DeliveryType == MessageDeliveryType.Internal && typeof(TInternalCommand).ToString() == x.DataType
-                );
-
-                var res = filter.Any(x => x.MessageStatus == MessageStatus.Processed);
-
-                return res;
-            });
-        });
-
-        return result;
+        return WaitUntilConditionMet(
+            conditionToMet: async () =>
+            {
+                cancellationToken.ThrowIfCancellationRequested();
+                return await condition();
+            },
+            timeoutSecond: (int)Math.Ceiling(effectiveTimeout.TotalSeconds),
+            pollInterval: effectivePollInterval,
+            cancellationToken: cancellationToken
+        );
     }
 
     // Ref: https://tech.energyhelpline.com/in-memory-testing-with-masstransit/
-    private async Task<bool> WaitUntilConditionMet(Func<Task<bool>> conditionToMet, int? timeoutSecond = null)
+    private async Task<bool> WaitUntilConditionMet(
+        Func<Task<bool>> conditionToMet,
+        int? timeoutSecond = null,
+        TimeSpan? pollInterval = null,
+        CancellationToken cancellationToken = default
+    )
     {
         var time = timeoutSecond ?? Timeout;
+        var delay = pollInterval ?? TimeSpan.FromMilliseconds(100);
 
-        var startTime = DateTime.Now;
-        var timeoutExpired = false;
-        var meet = await conditionToMet.Invoke();
-
-        while (!meet)
+        var startTime = DateTime.UtcNow;
+        while (DateTime.UtcNow - startTime <= TimeSpan.FromSeconds(time))
         {
-            if (timeoutExpired)
-            {
-                return false;
-            }
+            cancellationToken.ThrowIfCancellationRequested();
+
+            if (await conditionToMet.Invoke())
+                return true;
 
-            await Task.Delay(100);
-            meet = await conditionToMet.Invoke();
-            timeoutExpired = DateTime.Now - startTime > TimeSpan.FromSeconds(time);
+            await Task.Delay(delay, cancellationToken);
         }
 
-        return true;
+        return false;
     }
 
     private async Task StartTestContainerAsync()
@@ -288,17 +280,14 @@
         MongoDbTestContainer = TestContainers.MongoTestContainer();
         EventStoreDbTestContainer = TestContainers.EventStoreTestContainer();
 
-        // Start containers in parallel for speed
         await Task.WhenAll(
             MongoDbTestContainer.StartAsync(),
             PostgresTestcontainer.StartAsync(),
             PostgresPersistTestContainer.StartAsync(),
             EventStoreDbTestContainer.StartAsync()
         );
 
-        // Start RabbitMQ last and wait extra time
         await RabbitMqTestContainer.StartAsync();
-        await Task.Delay(5000); // Give RabbitMQ extra time to initialize
     }
 
     private async Task StopTestContainerAsync()
@@ -321,7 +310,7 @@
                 new("PostgresOptions:ConnectionString:Identity", PostgresTestcontainer.GetConnectionString()),
                 new("PostgresOptions:ConnectionString:Passenger", PostgresTestcontainer.GetConnectionString()),
                 new("PersistMessageOptions:ConnectionString", PostgresPersistTestContainer.GetConnectionString()),
-                new("RabbitMqOptions:HostName", RabbitMqTestContainer.Hostname),
+                new("RabbitMqOptions:HostName", "127.0.0.1"),
                 new("RabbitMqOptions:UserName", TestContainers.RabbitMqContainerConfiguration.UserName),
                 new("RabbitMqOptions:Password", TestContainers.RabbitMqContainerConfiguration.Password),
                 new(
@@ -534,7 +523,7 @@
     public TestFixtureCore(
         TestFixture<TEntryPoint> integrationTestFixture,
         ITestOutputHelper outputHelper,
         Type dbContextType = null
     )
     {
         Fixture = integrationTestFixture;
@@ -680,14 +669,14 @@
 {
     protected TestReadBase(
         TestReadFixture<TEntryPoint, TRContext> integrationTestFixture,
         ITestOutputHelper outputHelper = null
     )
         : base(integrationTestFixture, outputHelper)
     {
         Fixture = integrationTestFixture;
     }
 
     public TestReadFixture<TEntryPoint, TRContext> Fixture { get; }
 }
 
 public abstract class TestWriteBase<TEntryPoint, TWContext> : TestFixtureCore<TEntryPoint>
@@ -697,14 +686,14 @@
 {
     protected TestWriteBase(
         TestWriteFixture<TEntryPoint, TWContext> integrationTestFixture,
         ITestOutputHelper outputHelper = null
     )
         : base(integrationTestFixture, outputHelper, typeof(TWContext))
     {
         Fixture = integrationTestFixture;
     }
 
     public TestWriteFixture<TEntryPoint, TWContext> Fixture { get; }
 }
 
 public abstract class TestBase<TEntryPoint, TWContext, TRContext> : TestFixtureCore<TEntryPoint>
@@ -715,12 +704,12 @@
 {
     protected TestBase(
         TestFixture<TEntryPoint, TWContext, TRContext> integrationTestFixture,
         ITestOutputHelper outputHelper = null
     )
         : base(integrationTestFixture, outputHelper, typeof(TWContext))
     {
         Fixture = integrationTestFixture;
     }
 
     public TestFixture<TEntryPoint, TWContext, TRContext> Fixture { get; }
 }

src/Services/Identity/tests/IntegrationTest/Identity/Features/RegisterNewUserTests.cs

@@ -10,10 +10,8 @@ namespace Integration.Test.Identity.Features;
 
 public class RegisterNewUserTests : IdentityIntegrationTestBase
 {
-    public RegisterNewUserTests(
-        TestWriteFixture<Program, IdentityContext> integrationTestFactory) : base(integrationTestFactory)
-    {
-    }
+    public RegisterNewUserTests(TestWriteFixture<Program, IdentityContext> integrationTestFactory)
+        : base(integrationTestFactory) { }
 
     [Fact]
     public async Task should_create_new_user_to_db_and_publish_message_to_broker()
@@ -30,4 +28,4 @@ public async Task should_create_new_user_to_db_and_publish_message_to_broker()
 
         (await Fixture.WaitForPublishing<UserCreated>()).Should().Be(true);
     }
-}
+}

src/Services/Passenger/tests/IntegrationTest/Passenger/Features/CompleteRegisterPassengerTests.cs

@@ -1,4 +1,3 @@
-using BuildingBlocks.Contracts.EventBus.Messages;
 using BuildingBlocks.TestBase;
 using FluentAssertions;
 using Integration.Test.Fakes;
@@ -21,8 +20,8 @@
         var userCreated = new FakeUserCreated().Generate();
 
         await Fixture.Publish(userCreated);
-        (await Fixture.WaitForPublishing<UserCreated>()).Should().Be(true);
-        (await Fixture.WaitForConsuming<UserCreated>()).Should().Be(true);
+
+        (await WaitUntilPassengerCreatedAsync(userCreated.PassportNumber)).Should().BeTrue();
 
         var command = new FakeCompleteRegisterPassengerCommand(userCreated.PassportNumber).Generate();
 
@@ -36,4 +35,14 @@
         response?.PassengerDto?.PassengerType.ToString().Should().Be(command.PassengerType.ToString());
         response?.PassengerDto?.Age.Should().Be(command.Age);
     }
+
+    private Task<bool> WaitUntilPassengerCreatedAsync(string passportNumber)
+    {
+        return Fixture.WaitUntilAsync(async () =>
+        {
+            return await Fixture.ExecuteDbContextAsync(db =>
+                ValueTask.FromResult(db.Passengers.Any(p => p.PassportNumber.Value == passportNumber))
+            );
+        });
+    }
 }