Add Google Pub/Sub emulator integration test

Signed-off-by: Tomasz Maruszak <maruszaktomasz@gmail.com>

Author: zarusz

.github/workflows/build.yml

@@ -178,6 +178,12 @@ jobs:
             test_path: "Tests/SlimMessageBus.Host.PostgreSql.Test"
             docker_services: ""
 
+          - transport: GooglePubSub
+            name: Google Pub/Sub
+            filter: "Category=Integration&Transport=GooglePubSub"
+            test_path: "Tests/SlimMessageBus.Host.GooglePubSub.Test"
+            docker_services: ""
+
           # Outbox – four parallel sub-legs:
           # 1a. Sql.DbContext OutboxTests – AzureSB + Kafka; SQL Server via local docker
           - transport: Outbox.SqlDbContext.Tests

src/Tests/SlimMessageBus.Host.GooglePubSub.Test/GooglePubSubFixture.cs

@@ -0,0 +1,19 @@
+namespace SlimMessageBus.Host.GooglePubSub.Test;
+
+public sealed class GooglePubSubFixture : IAsyncLifetime
+{
+    private readonly PubSubContainer _container = new PubSubBuilder(
+        "gcr.io/google.com/cloudsdktool/google-cloud-cli:446.0.1-emulators").Build();
+
+    public string EmulatorEndpoint => _container.GetEmulatorEndpoint();
+
+    public Task InitializeAsync() => _container.StartAsync();
+
+    public async Task DisposeAsync()
+    {
+        await _container.DisposeAsync();
+    }
+}
+
+[CollectionDefinition(nameof(GooglePubSubCollection))]
+public sealed class GooglePubSubCollection : ICollectionFixture<GooglePubSubFixture>;

src/Tests/SlimMessageBus.Host.GooglePubSub.Test/GooglePubSubMessageBusIt.cs

@@ -0,0 +1,80 @@
+namespace SlimMessageBus.Host.GooglePubSub.Test;
+
+[Trait("Category", "Integration")]
+[Trait("Transport", "GooglePubSub")]
+[Collection(nameof(GooglePubSubCollection))]
+public class GooglePubSubMessageBusIt(ITestOutputHelper output, GooglePubSubFixture fixture)
+    : BaseIntegrationTest<GooglePubSubMessageBusIt>(output)
+{
+    private readonly string _topic = $"orders-{Guid.NewGuid():N}";
+    private readonly string _subscription = $"orders-worker-{Guid.NewGuid():N}";
+
+    protected override void SetupServices(ServiceCollection services, IConfigurationRoot configuration)
+    {
+        services.AddSlimMessageBus(mbb =>
+        {
+            mbb.WithProviderGooglePubSub(settings =>
+            {
+                settings.ProjectId = "slim-message-bus-tests";
+                ConfigureEmulatorClients(settings, fixture.EmulatorEndpoint);
+            });
+            mbb.Produce<OrderSubmitted>(x => x.DefaultTopic(_topic));
+            mbb.Consume<OrderSubmitted>(x => x.Topic(_topic).SubscriptionName(_subscription));
+            mbb.AddServicesFromAssemblyContaining<OrderSubmittedConsumer>();
+            mbb.AddJsonSerializer();
+        });
+
+        services.AddSingleton<TestEventCollector<OrderSubmitted>>();
+    }
+
+    [Fact]
+    public async Task ProvisionsTopologyAndDeliversPublishedMessage()
+    {
+        var messageBus = ServiceProvider.GetRequiredService<IMessageBus>();
+        var consumedMessages = ServiceProvider.GetRequiredService<TestEventCollector<OrderSubmitted>>();
+        var message = new OrderSubmitted(Guid.NewGuid(), 42);
+
+        await EnsureConsumersStarted();
+        await messageBus.Publish(message, headers: new Dictionary<string, object> { ["source"] = "integration-test" });
+        await consumedMessages.WaitUntilArriving(newMessagesTimeout: 30, expectedCount: 1);
+
+        consumedMessages.Snapshot().Should().ContainSingle().Which.Should().Be(message);
+    }
+
+    private static void ConfigureEmulatorClients(GooglePubSubMessageBusSettings settings, string endpoint)
+    {
+        settings.PublisherClientFactory = (_, topic, cancellationToken) => new PublisherClientBuilder
+        {
+            TopicName = topic,
+            Endpoint = endpoint,
+            ChannelCredentials = ChannelCredentials.Insecure
+        }.BuildAsync(cancellationToken);
+        settings.SubscriberClientFactory = (_, subscription, cancellationToken) => new SubscriberClientBuilder
+        {
+            SubscriptionName = subscription,
+            Endpoint = endpoint,
+            ChannelCredentials = ChannelCredentials.Insecure
+        }.BuildAsync(cancellationToken);
+        settings.PublisherServiceApiClientFactory = (_, cancellationToken) => new PublisherServiceApiClientBuilder
+        {
+            Endpoint = endpoint,
+            ChannelCredentials = ChannelCredentials.Insecure
+        }.BuildAsync(cancellationToken);
+        settings.SubscriberServiceApiClientFactory = (_, cancellationToken) => new SubscriberServiceApiClientBuilder
+        {
+            Endpoint = endpoint,
+            ChannelCredentials = ChannelCredentials.Insecure
+        }.BuildAsync(cancellationToken);
+    }
+
+    public record OrderSubmitted(Guid OrderId, int Quantity);
+
+    public class OrderSubmittedConsumer(TestEventCollector<OrderSubmitted> messages) : IConsumer<OrderSubmitted>
+    {
+        public Task OnHandle(OrderSubmitted message, CancellationToken cancellationToken)
+        {
+            messages.Add(message);
+            return Task.CompletedTask;
+        }
+    }
+}

src/Tests/SlimMessageBus.Host.GooglePubSub.Test/SlimMessageBus.Host.GooglePubSub.Test.csproj

@@ -3,8 +3,20 @@
   <Import Project="../Host.Test.Properties.xml" />
 
   <ItemGroup>
+    <ProjectReference Include="..\SlimMessageBus.Host.Test.Common\SlimMessageBus.Host.Test.Common.csproj" />
     <ProjectReference Include="..\..\SlimMessageBus.Host.GooglePubSub\SlimMessageBus.Host.GooglePubSub.csproj" />
     <ProjectReference Include="..\..\SlimMessageBus.Host\SlimMessageBus.Host.csproj" />
+    <ProjectReference Include="..\..\SlimMessageBus.Host.Serialization.SystemTextJson\SlimMessageBus.Host.Serialization.SystemTextJson.csproj" />
+  </ItemGroup>
+
+  <ItemGroup>
+    <PackageReference Include="Testcontainers.PubSub" Version="4.14.0" />
+  </ItemGroup>
+
+  <ItemGroup>
+    <None Update="appsettings.json">
+      <CopyToOutputDirectory>Always</CopyToOutputDirectory>
+    </None>
   </ItemGroup>
 
 </Project>

src/Tests/SlimMessageBus.Host.GooglePubSub.Test/Usings.cs

@@ -2,12 +2,23 @@
 
 global using Google.Cloud.PubSub.V1;
 
+global using Grpc.Core;
+
+global using Microsoft.Extensions.Configuration;
+global using Microsoft.Extensions.DependencyInjection;
 global using Microsoft.Extensions.Logging.Abstractions;
 
 global using Moq;
 
+global using SlimMessageBus;
+global using SlimMessageBus.Host;
 global using SlimMessageBus.Host.Collections;
 global using SlimMessageBus.Host.Interceptor;
 global using SlimMessageBus.Host.Serialization;
+global using SlimMessageBus.Host.Serialization.SystemTextJson;
+global using SlimMessageBus.Host.Test.Common.IntegrationTest;
+
+global using Testcontainers.PubSub;
 
 global using Xunit;
+global using Xunit.Abstractions;

src/Tests/SlimMessageBus.Host.GooglePubSub.Test/appsettings.json

@@ -0,0 +1 @@
+{}