agent-framework/dotnet/tests/Microsoft.Agents.AI.Hosting.AzureFunctions.IntegrationTests/SamplesValidation.cs

// Copyright (c) Microsoft. All rights reserved.

using System.Diagnostics;
using System.Reflection;
using System.Text;
using System.Text.Json;
using Microsoft.Extensions.Configuration;
using Microsoft.Extensions.Logging;
using ModelContextProtocol.Client;
using ModelContextProtocol.Protocol;

namespace Microsoft.Agents.AI.Hosting.AzureFunctions.IntegrationTests;

[Collection("Samples")]
[Trait("Category", "SampleValidation")]
public sealed class SamplesValidation(ITestOutputHelper outputHelper) : IAsyncLifetime
{
    private const string AzureFunctionsPort = "7071";
    private const string AzuritePort = "10000";
    private const string DtsPort = "8080";
    private const string RedisPort = "6379";

    private static readonly string s_dotnetTargetFramework = GetTargetFramework();
    private static readonly HttpClient s_sharedHttpClient = new();
    private static readonly IConfiguration s_configuration =
        new ConfigurationBuilder()
            .AddEnvironmentVariables()
            .AddUserSecrets(Assembly.GetExecutingAssembly())
            .Build();

    private static bool s_infrastructureStarted;
    private static readonly TimeSpan s_orchestrationTimeout = TimeSpan.FromMinutes(1);

    // In CI, `dotnet run` builds the Functions project from scratch before the host starts, so 60s is not enough.
    private static readonly TimeSpan s_functionsReadyTimeout = TimeSpan.FromSeconds(180);

    private static readonly string s_samplesPath = Path.GetFullPath(
        Path.Combine(AppDomain.CurrentDomain.BaseDirectory, "..", "..", "..", "..", "..", "samples", "04-hosting", "DurableAgents", "AzureFunctions"));

    private readonly ITestOutputHelper _outputHelper = outputHelper;

    async ValueTask IAsyncLifetime.InitializeAsync()
    {
        if (!s_infrastructureStarted)
        {
            await this.StartSharedInfrastructureAsync();
            s_infrastructureStarted = true;
        }
    }

    async ValueTask IAsyncDisposable.DisposeAsync()
    {
        // Nothing to clean up
        await Task.CompletedTask;
    }

    [Fact]
    public async Task SingleAgentSampleValidationAsync()
    {
        string samplePath = Path.Combine(s_samplesPath, "01_SingleAgent");
        await this.RunSampleTestAsync(samplePath, async (logs) =>
        {
            Uri startUri = new($"http://localhost:{AzureFunctionsPort}/api/agents/Joker/run");
            this._outputHelper.WriteLine($"Starting single agent orchestration via POST request to {startUri}...");

            // Test the agent endpoint as described in the README
            const string RequestBody = "Tell me a joke about a pirate.";
            using HttpContent content = new StringContent(RequestBody, Encoding.UTF8, "text/plain");

            using HttpResponseMessage response = await s_sharedHttpClient.PostAsync(startUri, content);

            // The response is expected to be a plain text response with the agent's reply (the joke)
            Assert.True(response.IsSuccessStatusCode, $"Agent request failed with status: {response.StatusCode}");
            Assert.Equal("text/plain", response.Content.Headers.ContentType?.MediaType);
            string responseText = await response.Content.ReadAsStringAsync();
            Assert.NotEmpty(responseText);
            this._outputHelper.WriteLine($"Agent run response: {responseText}");

            // The response headers should include the agent session ID, which can be used to continue the conversation.
            string? sessionId = response.Headers.GetValues("x-ms-thread-id")?.FirstOrDefault();
            Assert.NotNull(sessionId);
            Assert.NotEmpty(sessionId);

            this._outputHelper.WriteLine($"Agent session ID: {sessionId}");

            // Wait for up to 30 seconds to see if the agent response is available in the logs
            await this.WaitForConditionAsync(
                condition: () =>
                {
                    lock (logs)
                    {
                        bool exists = logs.Any(
                            log => log.Message.Contains("Response:") && log.Message.Contains(sessionId));
                        return Task.FromResult(exists);
                    }
                },
                message: "Agent response is available",
                timeout: TimeSpan.FromSeconds(30));
        });
    }

    [Fact]
    public async Task SingleAgentOrchestrationChainingSampleValidationAsync()
    {
        string samplePath = Path.Combine(s_samplesPath, "02_AgentOrchestration_Chaining");
        await this.RunSampleTestAsync(samplePath, async (logs) =>
        {
            Uri startUri = new($"http://localhost:{AzureFunctionsPort}/api/singleagent/run");
            this._outputHelper.WriteLine($"Starting single agent orchestration via POST request to {startUri}...");

            // Start the orchestration
            using HttpResponseMessage startResponse = await s_sharedHttpClient.PostAsync(startUri, content: null);

            Assert.True(
                startResponse.IsSuccessStatusCode,
                $"Start orchestration failed with status: {startResponse.StatusCode}");
            string startResponseText = await startResponse.Content.ReadAsStringAsync();
            JsonElement startResult = JsonElement.Parse(startResponseText);

            Assert.True(startResult.TryGetProperty("statusQueryGetUri", out JsonElement statusUriElement));
            Uri statusUri = new(statusUriElement.GetString()!);

            // Wait for orchestration to complete
            await this.WaitForOrchestrationCompletionAsync(statusUri);

            // Verify the final result
            using HttpResponseMessage statusResponse = await s_sharedHttpClient.GetAsync(statusUri);
            Assert.True(
                statusResponse.IsSuccessStatusCode,
                $"Status check failed with status: {statusResponse.StatusCode}");

            string statusText = await statusResponse.Content.ReadAsStringAsync();
            JsonElement statusResult = JsonElement.Parse(statusText);

            Assert.Equal("Completed", statusResult.GetProperty("runtimeStatus").GetString());
            Assert.True(statusResult.TryGetProperty("output", out JsonElement outputElement));
            string? output = outputElement.GetString();

            // Can't really validate the output since it's non-deterministic, but we can at least check it's non-empty
            Assert.NotNull(output);
            Assert.True(output.Length > 20, "Output is unexpectedly short");
        });
    }

    [Fact]
    public async Task MultiAgentOrchestrationConcurrentSampleValidationAsync()
    {
        string samplePath = Path.Combine(s_samplesPath, "03_AgentOrchestration_Concurrency");
        await this.RunSampleTestAsync(samplePath, async (logs) =>
        {
            // Start the multi-agent orchestration
            const string RequestBody = "What is temperature?";
            using HttpContent content = new StringContent(RequestBody, Encoding.UTF8, "text/plain");

            Uri startUri = new($"http://localhost:{AzureFunctionsPort}/api/multiagent/run");
            this._outputHelper.WriteLine($"Starting multi agent orchestration via POST request to {startUri}...");
            using HttpResponseMessage startResponse = await s_sharedHttpClient.PostAsync(startUri, content);

            Assert.True(startResponse.IsSuccessStatusCode, $"Start orchestration failed with status: {startResponse.StatusCode}");
            string startResponseText = await startResponse.Content.ReadAsStringAsync();
            JsonElement startResult = JsonElement.Parse(startResponseText);

            Assert.True(startResult.TryGetProperty("instanceId", out JsonElement instanceIdElement));
            Assert.True(startResult.TryGetProperty("statusQueryGetUri", out JsonElement statusUriElement));

            Uri statusUri = new(statusUriElement.GetString()!);

            // Wait for orchestration to complete
            await this.WaitForOrchestrationCompletionAsync(statusUri);

            // Verify the final result
            using HttpResponseMessage statusResponse = await s_sharedHttpClient.GetAsync(statusUri);
            Assert.True(statusResponse.IsSuccessStatusCode, $"Status check failed with status: {statusResponse.StatusCode}");

            string statusText = await statusResponse.Content.ReadAsStringAsync();
            JsonElement statusResult = JsonElement.Parse(statusText);

            Assert.Equal("Completed", statusResult.GetProperty("runtimeStatus").GetString());
            Assert.True(statusResult.TryGetProperty("output", out JsonElement outputElement));

            // Verify both physicist and chemist responses are present
            Assert.True(outputElement.TryGetProperty("physicist", out JsonElement physicistElement));
            Assert.True(outputElement.TryGetProperty("chemist", out JsonElement chemistElement));

            string physicistResponse = physicistElement.GetString()!;
            string chemistResponse = chemistElement.GetString()!;

            Assert.NotEmpty(physicistResponse);
            Assert.NotEmpty(chemistResponse);
            Assert.Contains("temperature", physicistResponse, StringComparison.OrdinalIgnoreCase);
            Assert.Contains("temperature", chemistResponse, StringComparison.OrdinalIgnoreCase);
        });
    }

    [Fact]
    public async Task MultiAgentOrchestrationConditionalsSampleValidationAsync()
    {
        string samplePath = Path.Combine(s_samplesPath, "04_AgentOrchestration_Conditionals");
        await this.RunSampleTestAsync(samplePath, async (logs) =>
        {
            // Test with legitimate email
            await this.TestSpamDetectionAsync("email-001",
                "Hi John, I hope you're doing well. I wanted to follow up on our meeting yesterday about the quarterly report. Could you please send me the updated figures by Friday? Thanks!",
                expectedSpam: false);

            // Test with spam email
            await this.TestSpamDetectionAsync("email-002",
                "URGENT! You've won $1,000,000! Click here now to claim your prize! Limited time offer! Don't miss out!",
                expectedSpam: true);
        });
    }

    [Fact]
    public async Task SingleAgentOrchestrationHITLSampleValidationAsync()
    {
        string samplePath = Path.Combine(s_samplesPath, "05_AgentOrchestration_HITL");

        await this.RunSampleTestAsync(samplePath, async (logs) =>
        {
            // Start the HITL orchestration with short timeout for testing
            // TODO: Add validation for the approval case
            object requestBody = new
            {
                topic = "The Future of Artificial Intelligence",
                max_review_attempts = 3,
                approval_timeout_hours = 0.001 // Very short timeout for testing
            };

            string jsonContent = JsonSerializer.Serialize(requestBody);
            using HttpContent content = new StringContent(jsonContent, Encoding.UTF8, "application/json");

            Uri startUri = new($"http://localhost:{AzureFunctionsPort}/api/hitl/run");
            this._outputHelper.WriteLine($"Starting HITL orchestration via POST request to {startUri}...");
            using HttpResponseMessage startResponse = await s_sharedHttpClient.PostAsync(startUri, content);

            Assert.True(
                startResponse.IsSuccessStatusCode,
                $"Start HITL orchestration failed with status: {startResponse.StatusCode}");
            string startResponseText = await startResponse.Content.ReadAsStringAsync();
            JsonElement startResult = JsonElement.Parse(startResponseText);

            Assert.True(startResult.TryGetProperty("statusQueryGetUri", out JsonElement statusUriElement));
            Uri statusUri = new(statusUriElement.GetString()!);

            // Wait for orchestration to complete (it should timeout due to short timeout)
            await this.WaitForOrchestrationCompletionAsync(statusUri);

            // Verify the final result
            using HttpResponseMessage statusResponse = await s_sharedHttpClient.GetAsync(statusUri);
            Assert.True(
                statusResponse.IsSuccessStatusCode,
                $"Status check failed with status: {statusResponse.StatusCode}");

            string statusText = await statusResponse.Content.ReadAsStringAsync();
            this._outputHelper.WriteLine($"HITL orchestration status text: {statusText}");

            JsonElement statusResult = JsonElement.Parse(statusText);

            // The orchestration should complete with a failed status due to timeout
            Assert.Equal("Failed", statusResult.GetProperty("runtimeStatus").GetString());
            Assert.True(statusResult.TryGetProperty("failureDetails", out JsonElement failureDetailsElement));
            Assert.True(failureDetailsElement.TryGetProperty("ErrorType", out JsonElement errorTypeElement));
            Assert.Equal("System.TimeoutException", errorTypeElement.GetString());
            Assert.True(failureDetailsElement.TryGetProperty("ErrorMessage", out JsonElement errorMessageElement));
            Assert.StartsWith("Human approval timed out", errorMessageElement.GetString());
        });
    }

    [Fact]
    public async Task LongRunningToolsSampleValidationAsync()
    {
        string samplePath = Path.Combine(s_samplesPath, "06_LongRunningTools");

        await this.RunSampleTestAsync(samplePath, async (logs) =>
        {
            // Test starting an agent that schedules a content generation orchestration
            const string Prompt = "Start a content generation workflow for the topic 'The Future of Artificial Intelligence'";
            using HttpContent messageContent = new StringContent(Prompt, Encoding.UTF8, "text/plain");

            Uri runAgentUri = new($"http://localhost:{AzureFunctionsPort}/api/agents/publisher/run");

            this._outputHelper.WriteLine($"Starting agent tool orchestration via POST request to {runAgentUri}...");
            using HttpResponseMessage startResponse = await s_sharedHttpClient.PostAsync(runAgentUri, messageContent);

            Assert.True(
                startResponse.IsSuccessStatusCode,
                $"Start agent request failed with status: {startResponse.StatusCode}");

            string startResponseText = await startResponse.Content.ReadAsStringAsync();
            this._outputHelper.WriteLine($"Agent response: {startResponseText}");

            // The response should be deserializable as an AgentResponse object and have a valid session ID
            startResponse.Headers.TryGetValues("x-ms-thread-id", out IEnumerable<string>? agentIdValues);
            string? sessionId = agentIdValues?.FirstOrDefault();
            Assert.NotNull(sessionId);
            Assert.NotEmpty(sessionId);

            // Wait for the orchestration to report that it's waiting for human approval
            await this.WaitForConditionAsync(
                condition: () =>
                {
                    // For now, we have to rely on the logs to check for the "NOTIFICATION" message that gets generated by the activity function.
                    // TODO: Synchronously prompt the agent for status
                    lock (logs)
                    {
                        bool exists = logs.Any(log => log.Message.Contains("NOTIFICATION: Please review the following content for approval"));
                        return Task.FromResult(exists);
                    }
                },
                message: "Orchestration is requesting human feedback",
                timeout: TimeSpan.FromSeconds(60));

            // Approve the content
            Uri approvalUri = new($"{runAgentUri}?thread_id={sessionId}");
            using HttpContent approvalContent = new StringContent("Approve the content", Encoding.UTF8, "text/plain");
            using HttpResponseMessage approvalResponse = await s_sharedHttpClient.PostAsync(approvalUri, approvalContent);
            Assert.True(approvalResponse.IsSuccessStatusCode, $"Approve content request failed with status: {approvalResponse.StatusCode}");

            // Wait for the publish notification to be logged
            await this.WaitForConditionAsync(
                condition: () =>
                {
                    lock (logs)
                    {
                        // TODO: Synchronously prompt the agent for status
                        bool exists = logs.Any(log => log.Message.Contains("PUBLISHING: Content has been published successfully"));
                        return Task.FromResult(exists);
                    }
                },
                message: "Content published notification is logged",
                timeout: TimeSpan.FromSeconds(60));

            // Verify the final orchestration status by asking the agent for the status
            Uri statusUri = new($"{runAgentUri}?thread_id={sessionId}");
            await this.WaitForConditionAsync(
                condition: async () =>
                {
                    this._outputHelper.WriteLine($"Checking status of orchestration at {statusUri}...");

                    using StringContent content = new("Get the status of the workflow", Encoding.UTF8, "text/plain");
                    using HttpResponseMessage statusResponse = await s_sharedHttpClient.PostAsync(statusUri, content);
                    Assert.True(
                        statusResponse.IsSuccessStatusCode,
                        $"Status check failed with status: {statusResponse.StatusCode}");
                    string statusText = await statusResponse.Content.ReadAsStringAsync();
                    this._outputHelper.WriteLine($"Status text: {statusText}");

                    bool isCompleted = statusText.Contains("Completed", StringComparison.OrdinalIgnoreCase);
                    bool hasContent = statusText.Contains(
                        "The Future of Artificial Intelligence",
                        StringComparison.OrdinalIgnoreCase);
                    return isCompleted && hasContent;
                },
                message: "Orchestration is completed",
                timeout: TimeSpan.FromSeconds(60));
        });
    }

    [Fact]
    public async Task AgentAsMcpToolAsync()
    {
        string samplePath = Path.Combine(s_samplesPath, "07_AgentAsMcpTool");
        await this.RunSampleTestAsync(samplePath, async (logs) =>
        {
            IClientTransport clientTransport = new HttpClientTransport(new()
            {
                Endpoint = new Uri($"http://localhost:{AzureFunctionsPort}/runtime/webhooks/mcp")
            });

            await using McpClient mcpClient = await McpClient.CreateAsync(clientTransport!);

            // Ensure the expected tools are present.
            IList<McpClientTool> tools = await mcpClient.ListToolsAsync();

            Assert.Single(tools, t => t.Name == "StockAdvisor");
            Assert.Single(tools, t => t.Name == "PlantAdvisor");

            // Invoke the tools to verify they work as expected.
            string stockPriceResponse = await this.InvokeMcpToolAsync(mcpClient, "StockAdvisor", "MSFT ATH");
            string plantSuggestionResponse = await this.InvokeMcpToolAsync(mcpClient, "PlantAdvisor", "Low light plant");
            Assert.NotEmpty(stockPriceResponse);
            Assert.NotEmpty(plantSuggestionResponse);

            // Wait for up to 30 seconds to see if the agent responses are available in the logs
            await this.WaitForConditionAsync(
                condition: () =>
                {
                    lock (logs)
                    {
                        bool expectedLogsPresent = logs.Count(log => log.Message.Contains("Response:")) >= 2;
                        return Task.FromResult(expectedLogsPresent);
                    }
                },
                message: "Agent response is available",
                timeout: TimeSpan.FromSeconds(30));
        });
    }

    [Fact]
    public async Task ReliableStreamingSampleValidationAsync()
    {
        string samplePath = Path.Combine(s_samplesPath, "08_ReliableStreaming");
        await this.RunSampleTestAsync(samplePath, async (logs) =>
        {
            Uri createUri = new($"http://localhost:{AzureFunctionsPort}/api/agent/create");
            this._outputHelper.WriteLine($"Starting reliable streaming agent via POST request to {createUri}...");

            // Test the agent endpoint with a simple prompt
            const string RequestBody = "Plan a 3-day trip to Seattle. Include daily activities.";
            using HttpContent content = new StringContent(RequestBody, Encoding.UTF8, "text/plain");
            using HttpRequestMessage request = new(HttpMethod.Post, createUri)
            {
                Content = content
            };
            request.Headers.Add("Accept", "text/plain");

            using HttpResponseMessage response = await s_sharedHttpClient.SendAsync(
                request,
                HttpCompletionOption.ResponseHeadersRead);

            // The response should be successful
            Assert.True(response.IsSuccessStatusCode, $"Agent request failed with status: {response.StatusCode}");
            Assert.Equal("text/plain", response.Content.Headers.ContentType?.MediaType);

            // The response headers should include the conversation ID
            string? conversationId = response.Headers.GetValues("x-conversation-id")?.FirstOrDefault();
            Assert.NotNull(conversationId);
            Assert.NotEmpty(conversationId);
            this._outputHelper.WriteLine($"Agent conversation ID: {conversationId}");

            // Read the streamed response
            using Stream responseStream = await response.Content.ReadAsStreamAsync();
            using StreamReader reader = new(responseStream);
            StringBuilder responseText = new();
            char[] buffer = new char[1024];
            int bytesRead;

            // Read for a reasonable amount of time to get some content
            using CancellationTokenSource readTimeout = new(TimeSpan.FromSeconds(30));
            try
            {
                while (!readTimeout.Token.IsCancellationRequested)
                {
                    bytesRead = await reader.ReadAsync(buffer, 0, buffer.Length);
                    if (bytesRead == 0)
                    {
                        // Check if we've received enough content
                        if (responseText.Length > 50)
                        {
                            break;
                        }
                        await Task.Delay(100, readTimeout.Token);
                        continue;
                    }

                    responseText.Append(buffer, 0, bytesRead);
                    if (responseText.Length > 200)
                    {
                        // We've received enough content to validate
                        break;
                    }
                }
            }
            catch (OperationCanceledException)
            {
                // Timeout is acceptable if we got some content
            }

            string responseContent = responseText.ToString();
            Assert.True(responseContent.Length > 0, "Expected to receive some streamed content");
            this._outputHelper.WriteLine($"Received {responseContent.Length} characters of streamed content");

            // Test resumption by calling the stream endpoint
            Uri streamUri = new($"http://localhost:{AzureFunctionsPort}/api/agent/stream/{conversationId}");
            this._outputHelper.WriteLine($"Testing stream resumption via GET request to {streamUri}...");

            using HttpRequestMessage streamRequest = new(HttpMethod.Get, streamUri);
            streamRequest.Headers.Add("Accept", "text/plain");

            using HttpResponseMessage streamResponse = await s_sharedHttpClient.SendAsync(
                streamRequest,
                HttpCompletionOption.ResponseHeadersRead);
            Assert.True(streamResponse.IsSuccessStatusCode, $"Stream request failed with status: {streamResponse.StatusCode}");
            Assert.Equal("text/plain", streamResponse.Content.Headers.ContentType?.MediaType);

            // Verify the conversation ID header is present
            string? resumedConversationId = streamResponse.Headers.GetValues("x-conversation-id")?.FirstOrDefault();
            Assert.Equal(conversationId, resumedConversationId);

            // Read some content from the resumed stream
            using Stream resumedStream = await streamResponse.Content.ReadAsStreamAsync();
            using StreamReader resumedReader = new(resumedStream);
            StringBuilder resumedText = new();

            using CancellationTokenSource resumedReadTimeout = new(TimeSpan.FromSeconds(10));
            try
            {
                while (!resumedReadTimeout.Token.IsCancellationRequested)
                {
                    bytesRead = await resumedReader.ReadAsync(buffer, 0, buffer.Length);
                    if (bytesRead == 0)
                    {
                        if (resumedText.Length > 50)
                        {
                            break;
                        }
                        await Task.Delay(100, resumedReadTimeout.Token);
                        continue;
                    }

                    resumedText.Append(buffer, 0, bytesRead);
                    if (resumedText.Length > 100)
                    {
                        break;
                    }
                }
            }
            catch (OperationCanceledException)
            {
                // Timeout is acceptable if we got some content
            }

            string resumedContent = resumedText.ToString();
            Assert.True(resumedContent.Length > 0, "Expected to receive some content from resumed stream");
            this._outputHelper.WriteLine($"Received {resumedContent.Length} characters from resumed stream");
        });
    }

    private async Task<string> InvokeMcpToolAsync(McpClient mcpClient, string toolName, string query)
    {
        this._outputHelper.WriteLine($"Invoking MCP tool '{toolName}'...");

        CallToolResult result = await mcpClient.CallToolAsync(
            toolName,
            arguments: new Dictionary<string, object?> { { "query", query } });

        string toolCallResult = ((TextContentBlock)result.Content[0]).Text;
        this._outputHelper.WriteLine($"MCP tool '{toolName}' response: {toolCallResult}");

        return toolCallResult;
    }

    private async Task TestSpamDetectionAsync(string emailId, string emailContent, bool expectedSpam)
    {
        object requestBody = new
        {
            email_id = emailId,
            email_content = emailContent
        };

        string jsonContent = JsonSerializer.Serialize(requestBody);
        using HttpContent content = new StringContent(jsonContent, Encoding.UTF8, "application/json");

        Uri startUri = new($"http://localhost:{AzureFunctionsPort}/api/spamdetection/run");
        this._outputHelper.WriteLine($"Starting spam detection orchestration via POST request to {startUri}...");
        using HttpResponseMessage startResponse = await s_sharedHttpClient.PostAsync(startUri, content);

        Assert.True(startResponse.IsSuccessStatusCode, $"Start orchestration failed with status: {startResponse.StatusCode}");
        string startResponseText = await startResponse.Content.ReadAsStringAsync();
        JsonElement startResult = JsonElement.Parse(startResponseText);

        Assert.True(startResult.TryGetProperty("statusQueryGetUri", out JsonElement statusUriElement));
        Uri statusUri = new(statusUriElement.GetString()!);

        // Wait for orchestration to complete
        await this.WaitForOrchestrationCompletionAsync(statusUri);

        // Verify the final result
        using HttpResponseMessage statusResponse = await s_sharedHttpClient.GetAsync(statusUri);
        Assert.True(statusResponse.IsSuccessStatusCode, $"Status check failed with status: {statusResponse.StatusCode}");

        string statusText = await statusResponse.Content.ReadAsStringAsync();
        JsonElement statusResult = JsonElement.Parse(statusText);

        Assert.Equal("Completed", statusResult.GetProperty("runtimeStatus").GetString());
        Assert.True(statusResult.TryGetProperty("output", out JsonElement outputElement));

        string output = outputElement.GetString()!;
        Assert.NotEmpty(output);

        if (expectedSpam)
        {
            Assert.Contains("spam", output, StringComparison.OrdinalIgnoreCase);
        }
        else
        {
            Assert.Contains("sent", output, StringComparison.OrdinalIgnoreCase);
        }
    }

    private async Task StartSharedInfrastructureAsync()
    {
        // Start Azurite if it's not already running
        if (!await this.IsAzuriteRunningAsync())
        {
            await this.StartDockerContainerAsync(
                containerName: "azurite",
                image: "mcr.microsoft.com/azure-storage/azurite",
                ports: ["-p", "10000:10000", "-p", "10001:10001", "-p", "10002:10002"]);

            // Wait for Azurite
            await this.WaitForConditionAsync(this.IsAzuriteRunningAsync, "Azurite is running", TimeSpan.FromSeconds(30));
        }

        // Start DTS emulator if it's not already running
        if (!await this.IsDtsEmulatorRunningAsync())
        {
            await this.StartDockerContainerAsync(
                containerName: "dts-emulator",
                image: "mcr.microsoft.com/dts/dts-emulator:latest",
                ports: ["-p", "8080:8080", "-p", "8082:8082"]);

            // Wait for DTS emulator
            await this.WaitForConditionAsync(
                condition: this.IsDtsEmulatorRunningAsync,
                message: "DTS emulator is running",
                timeout: TimeSpan.FromSeconds(30));
        }

        // Start Redis if it's not already running
        if (!await this.IsRedisRunningAsync())
        {
            await this.StartDockerContainerAsync(
                containerName: "redis",
                image: "redis:latest",
                ports: ["-p", "6379:6379"]);

            // Wait for Redis
            await this.WaitForConditionAsync(
                condition: this.IsRedisRunningAsync,
                message: "Redis is running",
                timeout: TimeSpan.FromSeconds(30));
        }
    }

    private async Task<bool> IsAzuriteRunningAsync()
    {
        this._outputHelper.WriteLine(
            $"Checking if Azurite is running at http://localhost:{AzuritePort}/devstoreaccount1...");

        try
        {
            using CancellationTokenSource timeoutCts = new(TimeSpan.FromSeconds(30));

            // Example output when pinging Azurite:
            // $ curl -i http://localhost:10000/devstoreaccount1?comp=list
            // HTTP/1.1 403 Server failed to authenticate the request.
            // Server: Azurite-Blob/3.34.0
            // x-ms-error-code: AuthorizationFailure
            // x-ms-request-id: 6cd21522-bb0f-40f6-962c-fa174f17aa30
            // content-type: application/xml
            // Date: Mon, 20 Oct 2025 23:52:02 GMT
            // Connection: keep-alive
            // Keep-Alive: timeout=5
            // Transfer-Encoding: chunked
            using HttpResponseMessage response = await s_sharedHttpClient.GetAsync(
                requestUri: new Uri($"http://localhost:{AzuritePort}/devstoreaccount1?comp=list"),
                cancellationToken: timeoutCts.Token);
            if (response.Headers.TryGetValues(
                "Server",
                out IEnumerable<string>? serverValues) && serverValues.Any(s => s.StartsWith("Azurite", StringComparison.OrdinalIgnoreCase)))
            {
                this._outputHelper.WriteLine($"Azurite is running, server: {string.Join(", ", serverValues)}");
                return true;
            }

            this._outputHelper.WriteLine($"Azurite is not running. Status code: {response.StatusCode}");
            return false;
        }
        catch (HttpRequestException ex)
        {
            this._outputHelper.WriteLine($"Azurite is not running: {ex.Message}");
            return false;
        }
    }

    private async Task<bool> IsDtsEmulatorRunningAsync()
    {
        this._outputHelper.WriteLine($"Checking if DTS emulator is running at http://localhost:{DtsPort}/healthz...");

        // DTS emulator doesn't support HTTP/1.1, so we need to use HTTP/2.0
        using HttpClient http2Client = new()
        {
            DefaultRequestVersion = new Version(2, 0),
            DefaultVersionPolicy = HttpVersionPolicy.RequestVersionExact
        };

        try
        {
            using CancellationTokenSource timeoutCts = new(TimeSpan.FromSeconds(30));
            using HttpResponseMessage response = await http2Client.GetAsync(new Uri($"http://localhost:{DtsPort}/healthz"), timeoutCts.Token);
            if (response.Content.Headers.ContentLength > 0)
            {
                string content = await response.Content.ReadAsStringAsync(timeoutCts.Token);
                this._outputHelper.WriteLine($"DTS emulator health check response: {content}");
            }

            if (response.IsSuccessStatusCode)
            {
                this._outputHelper.WriteLine("DTS emulator is running");
                return true;
            }

            this._outputHelper.WriteLine($"DTS emulator is not running. Status code: {response.StatusCode}");
            return false;
        }
        catch (HttpRequestException ex)
        {
            this._outputHelper.WriteLine($"DTS emulator is not running: {ex.Message}");
            return false;
        }
    }

    private async Task<bool> IsRedisRunningAsync()
    {
        this._outputHelper.WriteLine($"Checking if Redis is running at localhost:{RedisPort}...");

        try
        {
            using CancellationTokenSource timeoutCts = new(TimeSpan.FromSeconds(30));
            ProcessStartInfo startInfo = new()
            {
                FileName = "docker",
                Arguments = "exec redis redis-cli ping",
                UseShellExecute = false,
                RedirectStandardOutput = true,
                RedirectStandardError = true,
                CreateNoWindow = true
            };

            using Process process = new() { StartInfo = startInfo };
            if (!process.Start())
            {
                this._outputHelper.WriteLine("Failed to start docker exec command");
                return false;
            }

            string output = await process.StandardOutput.ReadToEndAsync(timeoutCts.Token);
            await process.WaitForExitAsync(timeoutCts.Token);

            if (process.ExitCode == 0 && output.Contains("PONG", StringComparison.OrdinalIgnoreCase))
            {
                this._outputHelper.WriteLine("Redis is running");
                return true;
            }

            this._outputHelper.WriteLine($"Redis is not running. Exit code: {process.ExitCode}, Output: {output}");
            return false;
        }
        catch (Exception ex)
        {
            this._outputHelper.WriteLine($"Redis is not running: {ex.Message}");
            return false;
        }
    }

    private async Task StartDockerContainerAsync(string containerName, string image, string[] ports)
    {
        // Stop existing container if it exists
        await this.RunCommandAsync("docker", ["stop", containerName]);
        await this.RunCommandAsync("docker", ["rm", containerName]);

        // Start new container
        List<string> args = ["run", "-d", "--name", containerName];
        args.AddRange(ports);
        args.Add(image);

        this._outputHelper.WriteLine(
            $"Starting new container: {containerName} with image: {image} and ports: {string.Join(", ", ports)}");
        await this.RunCommandAsync("docker", args.ToArray());
        this._outputHelper.WriteLine($"Container started: {containerName}");
    }

    private async Task WaitForConditionAsync(Func<Task<bool>> condition, string message, TimeSpan timeout)
    {
        this._outputHelper.WriteLine($"Waiting for '{message}'...");

        using CancellationTokenSource cancellationTokenSource = new(timeout);
        while (true)
        {
            if (await condition())
            {
                return;
            }

            try
            {
                await Task.Delay(TimeSpan.FromSeconds(1), cancellationTokenSource.Token);
            }
            catch (OperationCanceledException) when (cancellationTokenSource.IsCancellationRequested)
            {
                throw new TimeoutException($"Timeout waiting for '{message}'");
            }
        }
    }

    private async Task RunSampleTestAsync(string samplePath, Func<IReadOnlyList<OutputLog>, Task> testAction)
    {
        // Build the sample project first (it may not have been built as part of the solution)
        await this.BuildSampleAsync(samplePath);

        // Start the Azure Functions app
        List<OutputLog> logsContainer = [];
        using Process funcProcess = this.StartFunctionApp(samplePath, logsContainer);
        try
        {
            // Wait for the app to be ready
            await this.WaitForAzureFunctionsAsync();

            // Run the test
            await testAction(logsContainer);
        }
        finally
        {
            await this.StopProcessAsync(funcProcess);
        }
    }

    private sealed record OutputLog(DateTime Timestamp, LogLevel Level, string Message);

    private async Task BuildSampleAsync(string samplePath)
    {
        this._outputHelper.WriteLine($"Building sample at {samplePath}...");

        ProcessStartInfo buildInfo = new()
        {
            FileName = "dotnet",
            Arguments = $"build -f {s_dotnetTargetFramework}",
            WorkingDirectory = samplePath,
            UseShellExecute = false,
            RedirectStandardOutput = true,
            RedirectStandardError = true,
        };

        using Process buildProcess = new() { StartInfo = buildInfo };
        buildProcess.Start();

        // Read both streams asynchronously to avoid deadlocks from filled pipe buffers
        Task<string> stdoutTask = buildProcess.StandardOutput.ReadToEndAsync();
        Task<string> stderrTask = buildProcess.StandardError.ReadToEndAsync();
        await buildProcess.WaitForExitAsync();

        string stderr = await stderrTask;
        if (buildProcess.ExitCode != 0)
        {
            string stdout = await stdoutTask;
            throw new InvalidOperationException($"Failed to build sample at {samplePath}:\n{stdout}\n{stderr}");
        }

        this._outputHelper.WriteLine($"Build completed for {samplePath}.");
    }

    private Process StartFunctionApp(string samplePath, List<OutputLog> logs)
    {
        ProcessStartInfo startInfo = new()
        {
            FileName = "dotnet",
            Arguments = $"run --no-build -f {s_dotnetTargetFramework} --port {AzureFunctionsPort}",
            WorkingDirectory = samplePath,
            UseShellExecute = false,
            RedirectStandardOutput = true,
            RedirectStandardError = true,
        };

        string openAiEndpoint = s_configuration["AZURE_OPENAI_ENDPOINT"] ??
            throw new InvalidOperationException("The required AZURE_OPENAI_ENDPOINT env variable is not set.");
        string openAiDeployment = s_configuration["AZURE_OPENAI_DEPLOYMENT_NAME"] ??
            throw new InvalidOperationException("The required AZURE_OPENAI_DEPLOYMENT_NAME env variable is not set.");

        // Set required environment variables for the function app (see local.settings.json for required settings)
        startInfo.EnvironmentVariables["AZURE_OPENAI_ENDPOINT"] = openAiEndpoint;
        startInfo.EnvironmentVariables["AZURE_OPENAI_DEPLOYMENT_NAME"] = openAiDeployment;
        startInfo.EnvironmentVariables["DURABLE_TASK_SCHEDULER_CONNECTION_STRING"] =
            $"Endpoint=http://localhost:{DtsPort};TaskHub=default;Authentication=None";
        startInfo.EnvironmentVariables["AzureWebJobsStorage"] = "UseDevelopmentStorage=true";
        startInfo.EnvironmentVariables["REDIS_CONNECTION_STRING"] = $"localhost:{RedisPort}";

        Process process = new() { StartInfo = startInfo };

        // Capture the output and error streams
        process.ErrorDataReceived += (sender, e) =>
        {
            if (e.Data != null)
            {
                this._outputHelper.WriteLine($"[{startInfo.FileName}(err)]: {e.Data}");
                lock (logs)
                {
                    logs.Add(new OutputLog(DateTime.Now, LogLevel.Error, e.Data));
                }
            }
        };

        process.OutputDataReceived += (sender, e) =>
        {
            if (e.Data != null)
            {
                this._outputHelper.WriteLine($"[{startInfo.FileName}(out)]: {e.Data}");
                lock (logs)
                {
                    logs.Add(new OutputLog(DateTime.Now, LogLevel.Information, e.Data));
                }
            }
        };

        if (!process.Start())
        {
            throw new InvalidOperationException("Failed to start the function app");
        }

        process.BeginErrorReadLine();
        process.BeginOutputReadLine();

        return process;
    }

    private async Task WaitForAzureFunctionsAsync()
    {
        this._outputHelper.WriteLine(
            $"Waiting for Azure Functions Core Tools to be ready at http://localhost:{AzureFunctionsPort}/...");
        await this.WaitForConditionAsync(
            condition: async () =>
            {
                try
                {
                    using HttpRequestMessage request = new(HttpMethod.Head, $"http://localhost:{AzureFunctionsPort}/");
                    using HttpResponseMessage response = await s_sharedHttpClient.SendAsync(request);
                    this._outputHelper.WriteLine($"Azure Functions Core Tools response: {response.StatusCode}");
                    return response.IsSuccessStatusCode;
                }
                catch (HttpRequestException)
                {
                    // Expected when the app isn't yet ready
                    return false;
                }
            },
            message: "Azure Functions Core Tools is ready",
            timeout: s_functionsReadyTimeout);
    }

    private async Task WaitForOrchestrationCompletionAsync(Uri statusUri)
    {
        using CancellationTokenSource timeoutCts = new(s_orchestrationTimeout);
        while (true)
        {
            try
            {
                using HttpResponseMessage response = await s_sharedHttpClient.GetAsync(
                    statusUri,
                    timeoutCts.Token);
                if (response.IsSuccessStatusCode)
                {
                    string responseText = await response.Content.ReadAsStringAsync(timeoutCts.Token);
                    JsonElement result = JsonElement.Parse(responseText);

                    if (result.TryGetProperty("runtimeStatus", out JsonElement statusElement) &&
                        statusElement.GetString() is "Completed" or "Failed" or "Terminated")
                    {
                        return;
                    }
                }
            }
            catch (Exception ex) when (!timeoutCts.Token.IsCancellationRequested)
            {
                // Ignore errors and retry
                this._outputHelper.WriteLine($"Error waiting for orchestration completion: {ex}");
            }

            await Task.Delay(TimeSpan.FromSeconds(1), timeoutCts.Token);
        }
    }

    private async Task RunCommandAsync(string command, string[] args)
    {
        await this.RunCommandAsync(command, workingDirectory: null, args: args);
    }

    private async Task RunCommandAsync(string command, string? workingDirectory, string[] args)
    {
        ProcessStartInfo startInfo = new()
        {
            FileName = command,
            Arguments = string.Join(" ", args),
            WorkingDirectory = workingDirectory,
            UseShellExecute = false,
            RedirectStandardOutput = true,
            RedirectStandardError = true,
            CreateNoWindow = true
        };

        this._outputHelper.WriteLine($"Running command: {command} {string.Join(" ", args)}");

        using Process process = new() { StartInfo = startInfo };
        process.ErrorDataReceived += (sender, e) => this._outputHelper.WriteLine($"[{command}(err)]: {e.Data}");
        process.OutputDataReceived += (sender, e) => this._outputHelper.WriteLine($"[{command}(out)]: {e.Data}");
        if (!process.Start())
        {
            throw new InvalidOperationException("Failed to start the command");
        }
        process.BeginErrorReadLine();
        process.BeginOutputReadLine();

        using CancellationTokenSource cancellationTokenSource = new(TimeSpan.FromMinutes(1));
        await process.WaitForExitAsync(cancellationTokenSource.Token);

        this._outputHelper.WriteLine($"Command completed with exit code: {process.ExitCode}");
    }

    private async Task StopProcessAsync(Process process)
    {
        try
        {
            if (!process.HasExited)
            {
                this._outputHelper.WriteLine($"Killing process {process.ProcessName}#{process.Id}");
                process.Kill(entireProcessTree: true);

                using CancellationTokenSource timeoutCts = new(TimeSpan.FromSeconds(10));
                await process.WaitForExitAsync(timeoutCts.Token);
                this._outputHelper.WriteLine($"Process exited: {process.Id}");
            }
        }
        catch (Exception ex)
        {
            this._outputHelper.WriteLine($"Failed to stop process: {ex.Message}");
        }
    }

    private static string GetTargetFramework()
    {
        // Get the target framework by looking at the path of the current file. It should be something like /path/to/project/bin/Debug/net8.0/...
        string filePath = new Uri(typeof(SamplesValidation).Assembly.Location).LocalPath;
        string directory = Path.GetDirectoryName(filePath)!;
        string tfm = Path.GetFileName(directory);
        if (tfm.StartsWith("net", StringComparison.OrdinalIgnoreCase))
        {
            return tfm;
        }

        throw new InvalidOperationException($"Unable to find target framework in path: {filePath}");
    }
}