WIP

Directory.Packages.props

@@ -32,6 +32,7 @@
  <PackageVersion Include="Confluent.Kafka" Version="2.0.2"/>
  <PackageVersion Include="Consul" Version="1.6.10.9"/>
  <PackageVersion Include="CouchbaseNetClient" Version="3.4.3"/>
+ <PackageVersion Include="Docker.DotNet" Version="3.1.2"/>
  <PackageVersion Include="Elastic.Clients.Elasticsearch" Version="8.0.5"/>
  <PackageVersion Include="EventStore.Client.Grpc.Streams" Version="22.0.0"/>
  <PackageVersion Include="FirebirdSql.Data.FirebirdClient" Version="10.0.0"/>
@@ -59,4 +60,4 @@
  <PackageVersion Include="Selenium.WebDriver" Version="4.8.1"/>
  <PackageVersion Include="StackExchange.Redis" Version="2.6.90"/>
  </ItemGroup>
-</Project>
+</Project>

Testcontainers.sln

@@ -191,6 +191,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.Pulsar", "src\Testcontainers.Pulsar\Testcontainers.Pulsar.csproj", "{27D46863-65B9-4934-B3C8-2383B217A477}"
+EndProject
+Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Testcontainers.Pulsar.Tests", "tests\Testcontainers.Pulsar.Tests\Testcontainers.Pulsar.Tests.csproj", "{D05FCB31-793E-43E0-BD6C-077013AE9113}"
+EndProject
 Global
  GlobalSection(SolutionConfigurationPlatforms) = preSolution
  Debug|Any CPU = Debug|Any CPU
@@ -556,6 +560,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
+ {27D46863-65B9-4934-B3C8-2383B217A477}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+ {27D46863-65B9-4934-B3C8-2383B217A477}.Debug|Any CPU.Build.0 = Debug|Any CPU
+ {27D46863-65B9-4934-B3C8-2383B217A477}.Release|Any CPU.ActiveCfg = Release|Any CPU
+ {27D46863-65B9-4934-B3C8-2383B217A477}.Release|Any CPU.Build.0 = Release|Any CPU
+ {D05FCB31-793E-43E0-BD6C-077013AE9113}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+ {D05FCB31-793E-43E0-BD6C-077013AE9113}.Debug|Any CPU.Build.0 = Debug|Any CPU
+ {D05FCB31-793E-43E0-BD6C-077013AE9113}.Release|Any CPU.ActiveCfg = Release|Any CPU
+ {D05FCB31-793E-43E0-BD6C-077013AE9113}.Release|Any CPU.Build.0 = Release|Any CPU
  EndGlobalSection
  GlobalSection(NestedProjects) = preSolution
  {5365F780-0E6C-41F0-B1B9-7DC34368F80C} = {673F23AE-7694-4BB9-ABD4-136D6C13634E}
@@ -647,5 +659,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}
+ {27D46863-65B9-4934-B3C8-2383B217A477} = {673F23AE-7694-4BB9-ABD4-136D6C13634E}
+ {D05FCB31-793E-43E0-BD6C-077013AE9113} = {7164F1FB-7F24-444A-ACD2-2C329C2B3CCF}
  EndGlobalSection
 EndGlobal

src/Testcontainers.Pulsar/.editorconfig

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

src/Testcontainers.Pulsar/PulsarBuilder.cs

@@ -0,0 +1,98 @@
+using Testcontainers.Pulsar;
+
+/// <inheritdoc cref="ContainerBuilder{TBuilderEntity, TContainerEntity, TConfigurationEntity}" />
+[PublicAPI]
+public sealed class PulsarBuilder : ContainerBuilder<PulsarBuilder, PulsarContainer, PulsarConfiguration>
+{
+ private const string AuthenticationPlugin = "org.apache.pulsar.client.impl.auth.AuthenticationToken";
+ private const string SecretKeyPath = "/pulsar/secret.key";
+ private const string UserName = "test-user";
+ private const string PulsarImage = "apachepulsar/pulsar:3.0.2";
+ private const string AdminClustersEndpoint = "/admin/v2/clusters";
+
+ private Dictionary<string, string> _environmentVariables = new Dictionary<string, string>
+ {
+ { "PULSAR_PREFIX_tokenSecretKey", $"file://{SecretKeyPath}" },
+ { "PULSAR_PREFIX_authenticationRefreshCheckSeconds", "5" },
+ { "superUserRoles", UserName },
+ { "authenticationEnabled", "true" },
+ { "authorizationEnabled", "true" },
+ { "authenticationProviders", "org.apache.pulsar.broker.authentication.AuthenticationProviderToken" },
+ { "authenticateOriginalAuthData", "false" },
+ { "brokerClientAuthenticationPlugin", AuthenticationPlugin },
+ { "CLIENT_PREFIX_authPlugin", AuthenticationPlugin }
+ };
+
+ public const ushort PulsarBrokerPort = 6650;
+ public const ushort PulsarBrokerHttpPort = 8080;
+
+ /// <summary>
+ /// Initializes a new instance of the <see cref="PulsarBuilder" /> class.
+ /// </summary>
+ public PulsarBuilder()
+ : this(new PulsarConfiguration())
+ {
+ DockerResourceConfiguration = Init().DockerResourceConfiguration;
+ }
+
+ /// <summary>
+ /// Initializes a new instance of the <see cref="PulsarBuilder" /> class.
+ /// </summary>
+ /// <param name="resourceConfiguration">The Docker resource configuration.</param>
+ private PulsarBuilder(PulsarConfiguration resourceConfiguration)
+ : base(resourceConfiguration)
+ {
+ DockerResourceConfiguration = resourceConfiguration;
+ }
+
+ /// <inheritdoc />
+ protected override PulsarConfiguration DockerResourceConfiguration { get; }
+
+ /// <inheritdoc />
+ public override PulsarContainer Build()
+ {
+ Validate();
+ return new PulsarContainer(DockerResourceConfiguration, TestcontainersSettings.Logger);
+ }
+
+ /// <inheritdoc />
+ protected override PulsarBuilder Init()
+ {
+ return base.Init()
+ .WithImage(PulsarImage)
+ .WithPortBinding(PulsarBrokerPort, true)
+ .WithPortBinding(PulsarBrokerHttpPort, true)
+ .WithCommand("/bin/bash", "-c",
+ "/pulsar/bin/apply-config-from-env.py /pulsar/conf/standalone.conf && bin/pulsar standalone")
+ .WithWaitStrategy(Wait.ForUnixContainer()
+ .UntilHttpRequestIsSucceeded(x => x.ForPath(AdminClustersEndpoint).ForPort(PulsarBrokerHttpPort)));
+ }
+
+ public PulsarBuilder WithTokenAuthentication()
+ {
+ return Merge(DockerResourceConfiguration, new PulsarConfiguration(true))
+ .WithEnvironment(_environmentVariables)
+ .WithCommand("/bin/bash", "-c", $"bin/pulsar tokens create-secret-key --output {SecretKeyPath} && " +
+ $"export brokerClientAuthenticationParameters=token:$(bin/pulsar tokens create --secret-key {SecretKeyPath} --subject {UserName}) && " +
+ $"export CLIENT_PREFIX_authParams=$brokerClientAuthenticationParameters && bin/apply-config-from-env.py conf/standalone.conf && " +
+ $"bin/apply-config-from-env-with-prefix.py CLIENT_PREFIX_ conf/client.conf && bin/pulsar standalone --no-functions-worker");
+ }
+
+ /// <inheritdoc />
+ protected override PulsarBuilder Clone(IResourceConfiguration<CreateContainerParameters> resourceConfiguration)
+ {
+ return Merge(DockerResourceConfiguration, new PulsarConfiguration(resourceConfiguration));
+ }
+
+ /// <inheritdoc />
+ protected override PulsarBuilder Clone(IContainerConfiguration resourceConfiguration)
+ {
+ return Merge(DockerResourceConfiguration, new PulsarConfiguration(resourceConfiguration));
+ }
+
+ /// <inheritdoc />
+ protected override PulsarBuilder Merge(PulsarConfiguration oldValue, PulsarConfiguration newValue)
+ {
+ return new PulsarBuilder(new PulsarConfiguration(oldValue, newValue));
+ }
+}

src/Testcontainers.Pulsar/PulsarConfiguration.cs

@@ -0,0 +1,57 @@
+namespace Testcontainers.Pulsar;
+
+/// <inheritdoc cref="ContainerConfiguration" />
+[PublicAPI]
+public sealed class PulsarConfiguration : ContainerConfiguration
+{
+ /// <summary>
+ /// Initializes a new instance of the <see cref="PulsarConfiguration" /> class.
+ /// </summary>
+ public PulsarConfiguration(bool authentication = false) => Authentication = authentication;
+
+ /// <summary>
+ /// Initializes a new instance of the <see cref="PulsarConfiguration" /> class.
+ /// </summary>
+ /// <param name="resourceConfiguration">The Docker resource configuration.</param>
+ public PulsarConfiguration(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="PulsarConfiguration" /> class.
+ /// </summary>
+ /// <param name="resourceConfiguration">The Docker resource configuration.</param>
+ public PulsarConfiguration(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="PulsarConfiguration" /> class.
+ /// </summary>
+ /// <param name="resourceConfiguration">The Docker resource configuration.</param>
+ public PulsarConfiguration(PulsarConfiguration resourceConfiguration)
+ : this(new PulsarConfiguration(), resourceConfiguration)
+ {
+ // Passes the configuration upwards to the base implementations to create an updated immutable copy.
+ }
+
+ /// <summary>
+ /// Initializes a new instance of the <see cref="PulsarConfiguration" /> class.
+ /// </summary>
+ /// <param name="oldValue">The old Docker resource configuration.</param>
+ /// <param name="newValue">The new Docker resource configuration.</param>
+ public PulsarConfiguration(PulsarConfiguration oldValue, PulsarConfiguration newValue)
+ : base(oldValue, newValue)
+ {
+ Authentication = BuildConfiguration.Combine(oldValue.Authentication, newValue.Authentication);
+ }
+
+ /// <summary>
+ /// Gets auth.
+ /// </summary>
+ public bool Authentication { get; }
+}

src/Testcontainers.Pulsar/PulsarContainer.cs

@@ -0,0 +1,58 @@
+namespace Testcontainers.Pulsar;
+
+/// <inheritdoc cref="DockerContainer" />
+[PublicAPI]
+public sealed class PulsarContainer : DockerContainer
+{
+ private readonly PulsarConfiguration _configuration;
+
+ /// <summary>
+ /// Initializes a new instance of the <see cref="PulsarContainer" /> class.
+ /// </summary>
+ /// <param name="configuration">The container configuration.</param>
+ /// <param name="logger">The logger.</param>
+ public PulsarContainer(PulsarConfiguration configuration, ILogger logger)
+ : base(configuration, logger)
+ {
+ _configuration = configuration;
+ }
+
+ /// <summary>
+ /// Gets the Pulsar broker url.
+ /// </summary>
+ /// <returns>The Pulsar broker url.</returns>
+ public string GetPulsarBrokerUrl() =>
+ new UriBuilder("pulsar://", Hostname, GetMappedPublicPort(PulsarBuilder.PulsarBrokerPort)).ToString();
+
+ /// <summary>
+ /// Gets the Pulsar service url.
+ /// </summary>
+ /// <returns>The Pulsar service url.</returns>
+ public string GetHttpServiceUrl() =>
+ new UriBuilder("http", Hostname, GetMappedPublicPort(PulsarBuilder.PulsarBrokerHttpPort)).ToString();
+
+ /// <summary>
+ /// Creates Authentication token
+ /// </summary>
+ /// <param name="expiryTime">Relative expiry time for the token (eg: 1h, 3d, 10y)</param>
+ /// <param name="cancellationToken"></param>
+ /// <returns>Authentication token</returns>
+ /// <exception cref="Exception"></exception>
+ public async Task<string> CreateToken(TimeSpan expiryTime, CancellationToken cancellationToken = default)
+ {
+ if (!_configuration.Authentication)
+ throw new Exception($"Could not create the token, because WithAuthentication is not used.");
+
+ var arguments = $"bin/pulsar tokens create --secret-key /pulsar/secret.key --subject test-user";
+
+ if (expiryTime != Timeout.InfiniteTimeSpan)
+ arguments += $" --expiry-time {expiryTime.TotalSeconds}s";
+
+ var result = await ExecAsync(new[] { "/bin/bash", "-c", arguments }, cancellationToken);
+
+ if (result.ExitCode != 0)
+ throw new Exception($"Could not create the token: {result.Stderr}");
+
+ return result.Stdout;
+ }
+}

src/Testcontainers.Pulsar/Testcontainers.Pulsar.csproj

@@ -0,0 +1,14 @@
+<Project Sdk="Microsoft.NET.Sdk">
+ <PropertyGroup>
+ <ImplicitUsings>enable</ImplicitUsings>
+ <Nullable>enable</Nullable>
+ <TargetFrameworks>netstandard2.0;netstandard2.1</TargetFrameworks>
+ <LangVersion>latest</LangVersion>
+ </PropertyGroup>
+ <ItemGroup>
+ <PackageReference Include="JetBrains.Annotations" VersionOverride="2023.3.0" PrivateAssets="All"/>
+ </ItemGroup>
+ <ItemGroup>
+ <ProjectReference Include="../Testcontainers/Testcontainers.csproj"/>
+ </ItemGroup>
+</Project>

src/Testcontainers.Pulsar/Usings.cs

@@ -0,0 +1,8 @@
+global using System;
+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.Pulsar.Tests/PulsarContainerTest.cs

@@ -0,0 +1,80 @@
+using System.Collections.Generic;
+using System.Threading;
+using DotPulsar;
+using DotPulsar.Abstractions;
+using DotPulsar.Extensions;
+using Xunit.Abstractions;
+
+namespace Testcontainers.Pulsar.Tests;
+
+public sealed class PulsarContainerTest : IAsyncLifetime
+{
+ private readonly CancellationTokenSource _cts;
+ private readonly PulsarContainer _pulsarContainer;
+ private readonly ITestOutputHelper _testOutputHelper;
+
+ public PulsarContainerTest(ITestOutputHelper testOutputHelper)
+ {
+ _testOutputHelper = testOutputHelper;
+ _pulsarContainer = new PulsarBuilder().Build();
+ _cts = new CancellationTokenSource(TimeSpan.FromMinutes(1));
+ }
+
+ public Task InitializeAsync()
+ {
+ return _pulsarContainer.StartAsync();
+ }
+
+ public Task DisposeAsync()
+ {
+ return _pulsarContainer.DisposeAsync().AsTask();
+ }
+
+ [Fact]
+ [Trait(nameof(DockerCli.DockerPlatform), nameof(DockerCli.DockerPlatform.Linux))]
+ public async Task PulsarContainer_WhenBrokerIsStarted_ShouldConnect()
+ {
+ // Given
+ await using var client = CreateClient();
+ var expected = new List<MessageId> { MessageId.Earliest };
+ await using var reader = CreateReader(client, MessageId.Earliest, await CreateTopic(_cts.Token));
+
+ // When
+ var actual = await reader.GetLastMessageIds(_cts.Token);
+
+ // Then
+ Assert.Equal(expected,actual);
+ }
+
+ private IReader<string> CreateReader(IPulsarClient pulsarClient, MessageId messageId, string topicName)
+ => pulsarClient.NewReader(Schema.String)
+ .StartMessageId(messageId)
+ .Topic(topicName)
+ .Create();
+
+ private static string CreateTopicName() => $"persistent://public/default/{Guid.NewGuid():N}";
+
+ private async Task CreateTopic(string topic, CancellationToken cancellationToken)
+ {
+ var arguments = $"bin/pulsar-admin topics create {topic}";
+
+ var result = await _pulsarContainer.ExecAsync(new[] { "/bin/bash", "-c", arguments }, cancellationToken);
+
+ if (result.ExitCode != 0)
+ throw new Exception($"Could not create the topic: {result.Stderr}");
+ }
+
+ private async Task<string> CreateTopic(CancellationToken cancellationToken)
+ {
+ var topic = CreateTopicName();
+ await CreateTopic(topic, cancellationToken);
+ return topic;
+ }
+
+ private IPulsarClient CreateClient()
+ => PulsarClient
+ .Builder()
+ .ExceptionHandler(context => _testOutputHelper.WriteLine($"PulsarClient got an exception: {context.Exception}"))
+ .ServiceUrl(new Uri(_pulsarContainer.GetPulsarBrokerUrl()))
+ .Build();
+}

tests/Testcontainers.Pulsar.Tests/Testcontainers.Pulsar.Tests.csproj

@@ -0,0 +1,19 @@
+<Project Sdk="Microsoft.NET.Sdk">
+ <PropertyGroup>
+ <TargetFramework>net8.0</TargetFramework>
+ <IsPackable>false</IsPackable>
+ <IsPublishable>false</IsPublishable>
+ <IsTestProject>true</IsTestProject>
+ </PropertyGroup>
+ <ItemGroup>
+ <PackageReference Include="DotPulsar"/>
+ <PackageReference Include="Microsoft.NET.Test.Sdk"/>
+ <PackageReference Include="xunit"/>
+ <PackageReference Include="xunit.runner.visualstudio"/>
+ <PackageReference Include="coverlet.collector"/>
+ </ItemGroup>
+ <ItemGroup>
+ <ProjectReference Include="../../src/Testcontainers.Pulsar/Testcontainers.Pulsar.csproj"/>
+ <ProjectReference Include="../Testcontainers.Commons/Testcontainers.Commons.csproj"/>
+ </ItemGroup>
+</Project>