[tests] Add Kakfa playground app to tests

Author: radical

playground/kafka/Consumer/ConsumerWorker.cs

@@ -19,6 +19,10 @@ protected override Task ExecuteAsync(CancellationToken stoppingToken)
                 try
                 {
                     result = consumer.Consume(TimeSpan.FromSeconds(1));
+                    if (result is not null)
+                    {
+                        logger.LogInformation($"Consumed message [{result.Message?.Key}] = {result.Message?.Value}");
+                    }
                 }
                 catch (ConsumeException ex) when (ex.Error.Code == ErrorCode.UnknownTopicOrPart)
                 {

tests/Aspire.Playground.Tests/Aspire.Playground.Tests.csproj

@@ -42,6 +42,7 @@
     <ProjectReference Include="$(PlaygroundSourceDir)Qdrant/Qdrant.AppHost/Qdrant.AppHost.csproj" />
     <ProjectReference Include="$(PlaygroundSourceDir)SqlServerEndToEnd/SqlServerEndToEnd.AppHost/SqlServerEndToEnd.AppHost.csproj" />
     <ProjectReference Include="$(PlaygroundSourceDir)TestShop/TestShop.AppHost/TestShop.AppHost.csproj" />
+    <ProjectReference Include="$(PlaygroundSourceDir)kafka/KafkaBasic.AppHost/KafkaBasic.AppHost.csproj" />
     <ProjectReference Include="$(PlaygroundSourceDir)keycloak/Keycloak.AppHost/Keycloak.AppHost.csproj" />
     <ProjectReference Include="$(PlaygroundSourceDir)milvus/MilvusPlayground.AppHost/MilvusPlayground.AppHost.csproj" />
     <ProjectReference Include="$(PlaygroundSourceDir)withdockerfile/WithDockerfile.AppHost/WithDockerfile.AppHost.csproj" />

tests/Aspire.Playground.Tests/ProjectSpecificTests.cs

@@ -1,10 +1,12 @@
 // Licensed to the .NET Foundation under one or more agreements.
 // The .NET Foundation licenses this file to you under the MIT license.
 
+using Aspire.Hosting;
 using Aspire.Hosting.Tests.Utils;
 using SamplesIntegrationTests;
 using SamplesIntegrationTests.Infrastructure;
 using Xunit;
+using Xunit.Sdk;
 using Xunit.Abstractions;
 
 namespace Aspire.Playground.Tests;
@@ -21,10 +23,59 @@ public async Task WithDockerfileTest()
         await app.StartAsync();
         await app.WaitForResources().WaitAsync(TimeSpan.FromMinutes(2));
 
-        await app.WaitForTextAsync([$"I'm Batman. - Batman", "I am Iron Man. - Iron Man"])
+        await app.WaitForTextAsync($"I'm Batman. - Batman")
                 .WaitAsync(TimeSpan.FromMinutes(3));
 
         app.EnsureNoErrorsLogged();
         await app.StopAsync();
     }
+
+    [Fact]
+    public async Task KafkaTest()
+    {
+        var appHostPath = Directory.GetFiles(AppContext.BaseDirectory, "KafkaBasic.AppHost.dll").Single();
+        var appHost = await DistributedApplicationTestFactory.CreateAsync(appHostPath, _testOutput);
+        await using var app = await appHost.BuildAsync();
+
+        await app.StartAsync();
+        await app.WaitForResources().WaitAsync(TimeSpan.FromMinutes(2));
+
+        await WaitForAllTextAsync(app,
+            [
+                "Hello, World! 343",
+                "Received 1000 messges."
+            ],
+            timeoutSecs: 30);
+
+        app.EnsureNoErrorsLogged();
+        await app.StopAsync();
+    }
+
+    internal static async Task WaitForAllTextAsync(DistributedApplication app, IEnumerable<string> logTexts, string? resourceName = null, int timeoutSecs = -1)
+    {
+        var table = logTexts.ToList();
+        try
+        {
+            var task = app.WaitForTextAsync((log) =>
+            {
+                foreach (var text in table)
+                {
+                    if (log.Contains(text))
+                    {
+                        table.Remove(text);
+                    }
+                }
+
+                return table.Count == 0;
+            }, resourceName);
+
+            await (timeoutSecs > 0
+                    ? task.WaitAsync(TimeSpan.FromSeconds(timeoutSecs))
+                    : task);
+        }
+        catch (TimeoutException te)
+        {
+            throw new XunitException($"The following messages were not found: {string.Join("', '", table)}", te);
+        }
+    }
 }