feat: Add PubSub module

Testcontainers.sln

@@ -145,6 +145,10 @@ Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Testcontainers.Tests", "tes
 EndProject
 Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Testcontainers.WebDriver.Tests", "tests\Testcontainers.WebDriver.Tests\Testcontainers.WebDriver.Tests.csproj", "{EBA72C3B-57D5-43FF-A5B4-3D55B3B6D4C2}"
 EndProject
+Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Testcontainers.PubSub", "src\Testcontainers.PubSub\Testcontainers.PubSub.csproj", "{E6642255-667D-476B-B584-089AA5E6C0B1}"
+EndProject
+Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Testcontainers.PubSub.Tests", "tests\Testcontainers.PubSub.Tests\Testcontainers.PubSub.Tests.csproj", "{0F86BCE8-62E1-4BFC-AA84-63C7514C90AC}"
+EndProject
 Global
  GlobalSection(SolutionConfigurationPlatforms) = preSolution
  Debug|Any CPU = Debug|Any CPU
@@ -418,6 +422,14 @@ Global
  {EBA72C3B-57D5-43FF-A5B4-3D55B3B6D4C2}.Debug|Any CPU.Build.0 = Debug|Any CPU
  {EBA72C3B-57D5-43FF-A5B4-3D55B3B6D4C2}.Release|Any CPU.ActiveCfg = Release|Any CPU
  {EBA72C3B-57D5-43FF-A5B4-3D55B3B6D4C2}.Release|Any CPU.Build.0 = Release|Any CPU
+ {E6642255-667D-476B-B584-089AA5E6C0B1}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+ {E6642255-667D-476B-B584-089AA5E6C0B1}.Debug|Any CPU.Build.0 = Debug|Any CPU
+ {E6642255-667D-476B-B584-089AA5E6C0B1}.Release|Any CPU.ActiveCfg = Release|Any CPU
+ {E6642255-667D-476B-B584-089AA5E6C0B1}.Release|Any CPU.Build.0 = Release|Any CPU
+ {0F86BCE8-62E1-4BFC-AA84-63C7514C90AC}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+ {0F86BCE8-62E1-4BFC-AA84-63C7514C90AC}.Debug|Any CPU.Build.0 = Debug|Any CPU
+ {0F86BCE8-62E1-4BFC-AA84-63C7514C90AC}.Release|Any CPU.ActiveCfg = Release|Any CPU
+ {0F86BCE8-62E1-4BFC-AA84-63C7514C90AC}.Release|Any CPU.Build.0 = Release|Any CPU
  EndGlobalSection
  GlobalSection(NestedProjects) = preSolution
  {3F2E254F-C203-43FD-A078-DC3E2CBC0F9F} = {673F23AE-7694-4BB9-ABD4-136D6C13634E}
@@ -486,5 +498,7 @@ Global
  {1A1983E6-5297-435F-B467-E8E1F11277D6} = {7164F1FB-7F24-444A-ACD2-2C329C2B3CCF}
  {27CDB869-A150-4593-958F-6F26E5391E7C} = {7164F1FB-7F24-444A-ACD2-2C329C2B3CCF}
  {EBA72C3B-57D5-43FF-A5B4-3D55B3B6D4C2} = {7164F1FB-7F24-444A-ACD2-2C329C2B3CCF}
+ {E6642255-667D-476B-B584-089AA5E6C0B1} = {673F23AE-7694-4BB9-ABD4-136D6C13634E}
+ {0F86BCE8-62E1-4BFC-AA84-63C7514C90AC} = {7164F1FB-7F24-444A-ACD2-2C329C2B3CCF}
  EndGlobalSection
 EndGlobal

src/Testcontainers.PubSub/.editorconfig

@@ -0,0 +1 @@
+root = true

src/Testcontainers.PubSub/PubSubBuilder.cs

@@ -0,0 +1,64 @@
+namespace Testcontainers.PubSub;
+
+/// <inheritdoc cref="ContainerBuilder{TBuilderEntity, TContainerEntity, TConfigurationEntity}" />
+[PublicAPI]
+public sealed class PubSubBuilder : ContainerBuilder<PubSubBuilder, PubSubContainer, PubSubConfiguration>
+{
+ public const string GoogleCloudCliImage = "gcr.io/google.com/cloudsdktool/google-cloud-cli:446.0.1-emulators";
+ public const ushort PubSubPort = 8085;
+ /// <summary>
+ /// Initializes a new instance of the <see cref="PubSubBuilder" /> class.
+ /// </summary>
+ public PubSubBuilder()
+ : this(new PubSubConfiguration())
+ {
+ DockerResourceConfiguration = Init().DockerResourceConfiguration;
+ }
+
+ /// <summary>
+ /// Initializes a new instance of the <see cref="PubSubBuilder" /> class.
+ /// </summary>
+ /// <param name="resourceConfiguration">The Docker resource configuration.</param>
+ private PubSubBuilder(PubSubConfiguration resourceConfiguration)
+ : base(resourceConfiguration)
+ {
+ DockerResourceConfiguration = resourceConfiguration;
+ }
+
+ protected override PubSubConfiguration DockerResourceConfiguration { get; }
+
+ /// <inheritdoc />
+ public override PubSubContainer Build()
+ {
+ Validate();
+ return new PubSubContainer(DockerResourceConfiguration, TestcontainersSettings.Logger);
+ }
+
+ /// <inheritdoc />
+ protected override PubSubBuilder Clone(IResourceConfiguration<CreateContainerParameters> resourceConfiguration)
+ {
+ return Merge(DockerResourceConfiguration, new PubSubConfiguration(resourceConfiguration));
+ }
+
+ /// <inheritdoc />
+ protected override PubSubBuilder Clone(IContainerConfiguration resourceConfiguration)
+ {
+ return Merge(DockerResourceConfiguration, new PubSubConfiguration(resourceConfiguration));
+ }
+
+ /// <inheritdoc />
+ protected override PubSubBuilder Merge(PubSubConfiguration oldValue, PubSubConfiguration newValue)
+ {
+ return new PubSubBuilder(new PubSubConfiguration(oldValue, newValue));
+ }
+
+ protected override PubSubBuilder Init()
+ {
+ return base.Init()
+ .WithImage(GoogleCloudCliImage)
+ .WithPortBinding(PubSubPort, true)
+ .WithEntrypoint("gcloud")
+ .WithCommand("beta", "emulators", "pubsub", "start", "--host-port", "0.0.0.0:" + PubSubPort)
+ .WithWaitStrategy(Wait.ForUnixContainer().UntilMessageIsLogged("(?s).*started.*$"));
+ }
+}

src/Testcontainers.PubSub/PubSubConfiguration.cs

@@ -0,0 +1,63 @@
+namespace Testcontainers.PubSub;
+
+/// <inheritdoc cref="ContainerConfiguration" />
+[PublicAPI]
+public sealed class PubSubConfiguration : ContainerConfiguration
+{
+ /// <summary>
+ /// Initializes a new instance of the <see cref="PubSubConfiguration" /> class.
+ /// </summary>
+ /// <param name="config">The PubSub config.</param>
+ public PubSubConfiguration(object config = null)
+ {
+ // // Sets the custom builder methods property values.
+ // Config = config;
+ }
+
+ /// <summary>
+ /// Initializes a new instance of the <see cref="PubSubConfiguration" /> class.
+ /// </summary>
+ /// <param name="resourceConfiguration">The Docker resource configuration.</param>
+ public PubSubConfiguration(IResourceConfiguration<CreateContainerParameters> resourceConfiguration)
+ : base(resourceConfiguration)
+ {
+ // Passes the configuration upwards to the base implementations to create an updated immutable copy.
+ }
+
+ /// <summary>
+ /// Initializes a new instance of the <see cref="PubSubConfiguration" /> class.
+ /// </summary>
+ /// <param name="resourceConfiguration">The Docker resource configuration.</param>
+ public PubSubConfiguration(IContainerConfiguration resourceConfiguration)
+ : base(resourceConfiguration)
+ {
+ // Passes the configuration upwards to the base implementations to create an updated immutable copy.
+ }
+
+ /// <summary>
+ /// Initializes a new instance of the <see cref="PubSubConfiguration" /> class.
+ /// </summary>
+ /// <param name="resourceConfiguration">The Docker resource configuration.</param>
+ public PubSubConfiguration(PubSubConfiguration resourceConfiguration)
+ : this(new PubSubConfiguration(), resourceConfiguration)
+ {
+ // Passes the configuration upwards to the base implementations to create an updated immutable copy.
+ }
+
+ /// <summary>
+ /// Initializes a new instance of the <see cref="PubSubConfiguration" /> class.
+ /// </summary>
+ /// <param name="oldValue">The old Docker resource configuration.</param>
+ /// <param name="newValue">The new Docker resource configuration.</param>
+ public PubSubConfiguration(PubSubConfiguration oldValue, PubSubConfiguration newValue)
+ : base(oldValue, newValue)
+ {
+ // // Create an updated immutable copy of the module configuration.
+ // Config = BuildConfiguration.Combine(oldValue.Config, newValue.Config);
+ }
+
+ // /// <summary>
+ // /// Gets the PubSub config.
+ // /// </summary>
+ // public object Config { get; }
+}

src/Testcontainers.PubSub/PubSubContainer.cs

@@ -0,0 +1,25 @@
+namespace Testcontainers.PubSub;
+
+/// <inheritdoc cref="DockerContainer" />
+[PublicAPI]
+public sealed class PubSubContainer : DockerContainer
+{
+ /// <summary>
+ /// Initializes a new instance of the <see cref="PubSubContainer" /> class.
+ /// </summary>
+ /// <param name="configuration">The container configuration.</param>
+ /// <param name="logger">The logger.</param>
+ public PubSubContainer(PubSubConfiguration configuration, ILogger logger)
+ : base(configuration, logger)
+ {
+ }
+
+ /// <summary>
+ /// Gets the PubSub emulator endpoint.
+ /// </summary>
+ /// <returns>The PubSub emulator endpoint.</returns>
+ public string GetEmulatorEndpoint()
+ {
+ return new UriBuilder(Uri.UriSchemeHttp, Hostname, GetMappedPublicPort(PubSubBuilder.PubSubPort)).ToString();
+ }
+}

src/Testcontainers.PubSub/Testcontainers.PubSub.csproj

@@ -0,0 +1,13 @@
+<Project Sdk="Microsoft.NET.Sdk">
+ <PropertyGroup>
+ <TargetFrameworks>netstandard2.0;netstandard2.1</TargetFrameworks>
+ <LangVersion>latest</LangVersion>
+ </PropertyGroup>
+ <ItemGroup>
+ <PackageReference Include="Microsoft.SourceLink.GitHub" Version="1.1.1" PrivateAssets="All"/>
+ <PackageReference Include="JetBrains.Annotations" Version="2022.3.1" PrivateAssets="All"/>
+ </ItemGroup>
+ <ItemGroup>
+ <ProjectReference Include="$(SolutionDir)src/Testcontainers/Testcontainers.csproj"/>
+ </ItemGroup>
+</Project>

src/Testcontainers.PubSub/Usings.cs

@@ -0,0 +1,10 @@
+global using System;
+global using System.Collections.Generic;
+global using System.Linq;
+global using Docker.DotNet.Models;
+global using DotNet.Testcontainers;
+global using DotNet.Testcontainers.Builders;
+global using DotNet.Testcontainers.Configurations;
+global using DotNet.Testcontainers.Containers;
+global using JetBrains.Annotations;
+global using Microsoft.Extensions.Logging;

tests/Testcontainers.PubSub.Tests/PubSubContainerTests.cs

@@ -0,0 +1,48 @@
+
+namespace Testcontainers.PubSub.Tests;
+
+public class PubSubContainerTests : IAsyncLifetime
+{
+ private readonly PubSubContainer _pubSubContainer = new PubSubBuilder().Build();
+
+ [Fact]
+ [Trait(nameof(DockerCli.DockerPlatform), nameof(DockerCli.DockerPlatform.Linux))]
+ public async Task PublishSubscribeTest()
+ {
+ const string projectId = "testProj";
+ const string topicName = "testTopic";
+ const string subscriptionName = "test-sub";
+
+ var (publisher, subscriber) = await _pubSubContainer.CreatePublisherAndSubscriber();
+ var (topic,subscription) = await PubSubHelper.EnsureTopicAndSubscription(publisher, subscriber, projectId, topicName, subscriptionName);
+
+ var dataToSent = new testData();
+
+ await publisher.PublishAsync(topic, dataToSent.ToPubsubMessage().AsList());
+
+ var messages = await subscriber.PullAsyncImmediatelly(subscription);
+
+ Assert.Single(messages.ReceivedMessages);
+
+ await subscriber.AcknowledgeAllMessages(subscription, messages.ReceivedMessages);
+
+ var receivedObject = messages.ReceivedMessages.FirstOrDefault().GetStringData().AsObject<testData>();
+
+ Assert.Equal(receivedObject.Id,dataToSent.Id);
+ }
+
+ public Task InitializeAsync()
+ {
+ return _pubSubContainer.StartAsync();
+ }
+
+ public Task DisposeAsync()
+ {
+ return _pubSubContainer.DisposeAsync().AsTask();
+ }
+
+ public class testData
+ {
+ public string Id { get; set; } = Guid.NewGuid().ToString();
+ }
+}

tests/Testcontainers.PubSub.Tests/PubSubHelper.cs

@@ -0,0 +1,108 @@
+
+
+namespace Testcontainers.PubSub.Tests
+{
+ public static class PubSubHelper
+ {
+ public static async Task<(TopicName, SubscriptionName)> EnsureTopicAndSubscription(
+ PublisherServiceApiClient publisher, SubscriberServiceApiClient subscriber, string projectId, string topicName,
+ string subscriptionName)
+ {
+ var topic = TopicName.FromProjectTopic(projectId, topicName);
+ var subscription = new SubscriptionName(projectId, subscriptionName);
+
+ await publisher.CreateTopicAsync(topic);
+
+ await subscriber.CreateSubscriptionAsync(new Subscription
+ {
+ SubscriptionName = subscription,
+ TopicAsTopicName = topic,
+ PushConfig = null,
+ AckDeadlineSeconds = 0,
+ RetainAckedMessages = false,
+ Labels = {{"", ""},},
+ EnableMessageOrdering = false,
+ ExpirationPolicy = new ExpirationPolicy(),
+ Filter = "",
+ DeadLetterPolicy = null,
+ RetryPolicy = new RetryPolicy(),
+ Detached = false,
+ EnableExactlyOnceDelivery = false,
+ TopicMessageRetentionDuration = new Duration(),
+ BigqueryConfig = null,
+ State = Subscription.Types.State.Unspecified,
+ });
+
+ return (topic, subscription);
+ }
+
+ public static async Task<(PublisherServiceApiClient publisher, SubscriberServiceApiClient subscriber)>
+ CreatePublisherAndSubscriber(this PubSubContainer container)
+ {
+ var publisherBuilder = new PublisherServiceApiClientBuilder()
+ {
+ ChannelCredentials = ChannelCredentials.Insecure,
+ Endpoint = container.GetEmulatorEndpoint(),
+ };
+
+ var subscriberBuilder = new SubscriberServiceApiClientBuilder()
+ {
+ ChannelCredentials = ChannelCredentials.Insecure,
+ Endpoint = container.GetEmulatorEndpoint(),
+ };
+
+ var publisher = await publisherBuilder.BuildAsync();
+ var subscriber = await subscriberBuilder.BuildAsync();
+
+ return (publisher, subscriber);
+ }
+
+ public static PubsubMessage ToPubsubMessage<T>(this T data)
+ {
+ return new PubsubMessage()
+ {
+ Data = ByteString.CopyFrom(System.Text.Json.JsonSerializer.Serialize(data), System.Text.Encoding.UTF8)
+ };
+ }
+
+ public static List<PubsubMessage> AsList(this PubsubMessage message)
+ {
+ return new List<PubsubMessage>()
+ {
+ message
+ };
+ }
+
+ public static async Task<PullResponse> PullAsyncImmediatelly(this SubscriberServiceApiClient subscriber,
+ SubscriptionName subscription, int maxMessages = 100)
+ {
+ return await subscriber.PullAsync(new PullRequest
+ {
+ SubscriptionAsSubscriptionName = subscription,
+ ReturnImmediately = true,
+ MaxMessages = maxMessages,
+ });
+ }
+
+ public static async Task AcknowledgeAllMessages(this SubscriberServiceApiClient subscriber,
+ SubscriptionName subscription, RepeatedField<ReceivedMessage> messages)
+ {
+ await subscriber.AcknowledgeAsync(subscription, messages.Select(x => x.AckId));
+ }
+
+ public static string GetStringData(this ReceivedMessage message)
+ {
+ return System.Text.Encoding.UTF8.GetString(message.Message.Data.ToArray());
+ }
+
+ public static T AsObject<T>(this string text)
+ {
+ return System.Text.Json.JsonSerializer.Deserialize<T>(text
+ , new JsonSerializerOptions()
+ {
+ PropertyNameCaseInsensitive = true
+ }
+ );
+ }
+ }
+}

tests/Testcontainers.PubSub.Tests/Testcontainers.PubSub.Tests.csproj

@@ -0,0 +1,18 @@
+<Project Sdk="Microsoft.NET.Sdk">
+ <PropertyGroup>
+ <TargetFrameworks>net6.0</TargetFrameworks>
+ <IsPackable>false</IsPackable>
+ <IsPublishable>false</IsPublishable>
+ </PropertyGroup>
+ <ItemGroup>
+ <PackageReference Include="Microsoft.NET.Test.Sdk" Version="17.7.2"/>
+ <PackageReference Include="coverlet.collector" Version="6.0.0"/>
+ <PackageReference Include="xunit.runner.visualstudio" Version="2.5.0"/>
+ <PackageReference Include="xunit" Version="2.5.0"/>
+ <PackageReference Include="Google.Cloud.PubSub.V1" Version="3.5.0" />
+ </ItemGroup>
+ <ItemGroup>
+ <ProjectReference Include="$(SolutionDir)src/Testcontainers.PubSub/Testcontainers.PubSub.csproj"/>
+ <ProjectReference Include="$(SolutionDir)tests/Testcontainers.Commons/Testcontainers.Commons.csproj"/>
+ </ItemGroup>
+</Project>