agent/agent-framework
2
0
Fork 0
mirror of https://github.com/microsoft/agent-framework.git synced 2026-06-16 21:04:09 +08:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'main' into copilot/fix-messageindex-cache-properties

Browse Source
This commit is contained in:
Chris
2026-03-17 13:56:32 -07:00
committed by GitHub
Unverified
parent 31375e5d0c 7c85f98c27
commit d4a185d60d
140 changed files with 13156 additions and 696 deletions
Download Patch File Download Diff File Expand all files Collapse all files
dotnet/tests/Microsoft.Agents.AI.DurableTask.IntegrationTests/ConsoleAppSamplesValidation.cs
+20 -432
View File
@@ -2,46 +2,30 @@
using System.Collections.Concurrent;
using System.Diagnostics;
using System.Reflection;
using System.Text;
using Microsoft.Extensions.Configuration;
using Microsoft.Extensions.Logging;
namespace Microsoft.Agents.AI.DurableTask.IntegrationTests;
/// <summary>
/// Integration tests for validating the durable agent console app samples
/// located in samples/Durable/Agents/ConsoleApps.
/// </summary>
[Collection("Samples")]
[Trait("Category", "SampleValidation")]
public sealed class ConsoleAppSamplesValidation(ITestOutputHelper outputHelper) : IAsyncLifetime
public sealed class ConsoleAppSamplesValidation(ITestOutputHelper outputHelper) : SamplesValidationBase(outputHelper)
{
private const string DtsPort = "8080";
private const string RedisPort = "6379";
private static readonly string s_dotnetTargetFramework = GetTargetFramework();
private static readonly IConfiguration s_configuration =
new ConfigurationBuilder()
.AddUserSecrets(Assembly.GetExecutingAssembly())
.AddEnvironmentVariables()
.Build();
private static bool s_infrastructureStarted;
private static readonly string s_samplesPath = Path.GetFullPath(
Path.Combine(AppDomain.CurrentDomain.BaseDirectory, "..", "..", "..", "..", "..", "samples", "04-hosting", "DurableAgents", "ConsoleApps"));
private readonly ITestOutputHelper _outputHelper = outputHelper;
/// <inheritdoc />
protected override string SamplesPath => s_samplesPath;
async ValueTask IAsyncLifetime.InitializeAsync()
{
if (!s_infrastructureStarted)
{
await this.StartSharedInfrastructureAsync();
s_infrastructureStarted = true;
}
}
/// <inheritdoc />
protected override bool RequiresRedis => true;
async ValueTask IAsyncDisposable.DisposeAsync()
/// <inheritdoc />
protected override void ConfigureAdditionalEnvironmentVariables(ProcessStartInfo startInfo, Action<string, string> setEnvVar)
{
// Nothing to clean up
await Task.CompletedTask;
setEnvVar("REDIS_CONNECTION_STRING", $"localhost:{RedisPort}");
}
[Fact]
@@ -474,7 +458,7 @@ public sealed class ConsoleAppSamplesValidation(ITestOutputHelper outputHelper)
// (streams can complete very quickly, so we need to interrupt early)
if (foundConversationStart && !interrupted && contentLinesBeforeInterrupt >= 2)
{
this._outputHelper.WriteLine($"Interrupting stream after {contentLinesBeforeInterrupt} content lines");
this.OutputHelper.WriteLine($"Interrupting stream after {contentLinesBeforeInterrupt} content lines");
interrupted = true;
interruptTime = DateTime.Now;
@@ -492,7 +476,7 @@ public sealed class ConsoleAppSamplesValidation(ITestOutputHelper outputHelper)
foundLastCursor = true;
// Send Enter again to resume
this._outputHelper.WriteLine("Resuming stream from last cursor");
this.OutputHelper.WriteLine("Resuming stream from last cursor");
await this.WriteInputAsync(process, string.Empty, testTimeoutCts.Token);
resumed = true;
}
@@ -520,7 +504,7 @@ public sealed class ConsoleAppSamplesValidation(ITestOutputHelper outputHelper)
if (timeSinceInterrupt < TimeSpan.FromSeconds(2))
{
// Continue reading for a bit more to catch the cancellation message
this._outputHelper.WriteLine("Stream completed naturally, but waiting for Last cursor message after interrupt...");
this.OutputHelper.WriteLine("Stream completed naturally, but waiting for Last cursor message after interrupt...");
continue;
}
}
@@ -535,7 +519,7 @@ public sealed class ConsoleAppSamplesValidation(ITestOutputHelper outputHelper)
// Stop once we've verified the interrupt/resume flow works
if (resumed && foundResumeMessage && contentLinesAfterResume >= 5)
{
this._outputHelper.WriteLine($"Successfully verified interrupt/resume: {contentLinesBeforeInterrupt} lines before, {contentLinesAfterResume} lines after");
this.OutputHelper.WriteLine($"Successfully verified interrupt/resume: {contentLinesBeforeInterrupt} lines before, {contentLinesAfterResume} lines after");
break;
}
}
@@ -546,7 +530,7 @@ public sealed class ConsoleAppSamplesValidation(ITestOutputHelper outputHelper)
TimeSpan timeSinceInterrupt = DateTime.Now - interruptTime.Value;
if (timeSinceInterrupt < TimeSpan.FromSeconds(3))
{
this._outputHelper.WriteLine("Waiting for Last cursor message after interrupt...");
this.OutputHelper.WriteLine("Waiting for Last cursor message after interrupt...");
using CancellationTokenSource waitCts = new(TimeSpan.FromSeconds(2));
try
{
@@ -557,7 +541,7 @@ public sealed class ConsoleAppSamplesValidation(ITestOutputHelper outputHelper)
foundLastCursor = true;
if (!resumed)
{
this._outputHelper.WriteLine("Resuming stream from last cursor");
this.OutputHelper.WriteLine("Resuming stream from last cursor");
await this.WriteInputAsync(process, string.Empty, testTimeoutCts.Token);
resumed = true;
}
@@ -575,7 +559,7 @@ public sealed class ConsoleAppSamplesValidation(ITestOutputHelper outputHelper)
catch (OperationCanceledException)
{
// Timeout - check if we got enough to verify the flow
this._outputHelper.WriteLine($"Read timeout reached. Interrupted: {interrupted}, Resumed: {resumed}, Content before: {contentLinesBeforeInterrupt}, Content after: {contentLinesAfterResume}");
this.OutputHelper.WriteLine($"Read timeout reached. Interrupted: {interrupted}, Resumed: {resumed}, Content before: {contentLinesBeforeInterrupt}, Content after: {contentLinesAfterResume}");
}
Assert.True(foundConversationStart, "Conversation start message not found.");
@@ -585,7 +569,7 @@ public sealed class ConsoleAppSamplesValidation(ITestOutputHelper outputHelper)
// but we should still verify we got the conversation started
if (!interrupted)
{
this._outputHelper.WriteLine("WARNING: Stream completed before interrupt could be sent. This may indicate the stream is too fast.");
this.OutputHelper.WriteLine("WARNING: Stream completed before interrupt could be sent. This may indicate the stream is too fast.");
}
Assert.True(interrupted, "Stream was not interrupted (may have completed too quickly).");
@@ -595,400 +579,4 @@ public sealed class ConsoleAppSamplesValidation(ITestOutputHelper outputHelper)
Assert.True(contentLinesAfterResume > 0, "No content received after resume (expected to continue from cursor, not restart).");
});
}
private static string GetTargetFramework()
{
string filePath = new Uri(typeof(ConsoleAppSamplesValidation).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}");
}
private async Task StartSharedInfrastructureAsync()
{
this._outputHelper.WriteLine("Starting shared infrastructure for console app samples...");
// Start DTS emulator
await this.StartDtsEmulatorAsync();
// Start Redis
await this.StartRedisAsync();
// Wait for infrastructure to be ready
await Task.Delay(TimeSpan.FromSeconds(5));
}
private async Task StartDtsEmulatorAsync()
{
// Start DTS emulator if it's not already running
if (!await this.IsDtsEmulatorRunningAsync())
{
this._outputHelper.WriteLine("Starting DTS emulator...");
await this.RunCommandAsync("docker", [
"run", "-d",
"--name", "dts-emulator",
"-p", $"{DtsPort}:8080",
"-e", "DTS_USE_DYNAMIC_TASK_HUBS=true",
"mcr.microsoft.com/dts/dts-emulator:latest"
]);
}
}
private async Task StartRedisAsync()
{
if (!await this.IsRedisRunningAsync())
{
this._outputHelper.WriteLine("Starting Redis...");
await this.RunCommandAsync("docker", [
"run", "-d",
"--name", "redis",
"-p", $"{RedisPort}:6379",
"redis:latest"
]);
}
}
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 RunSampleTestAsync(string samplePath, Func<Process, BlockingCollection<OutputLog>, Task> testAction)
{
// Build the sample project first (it may not have been built as part of the solution)
await this.BuildSampleAsync(samplePath);
// Generate a unique TaskHub name for this sample test to prevent cross-test interference
// when multiple tests run together and share the same DTS emulator.
string uniqueTaskHubName = $"sample-{Guid.NewGuid().ToString("N").Substring(0, 6)}";
// Start the console app
// Use BlockingCollection to safely read logs asynchronously captured from the process
using BlockingCollection<OutputLog> logsContainer = [];
using Process appProcess = this.StartConsoleApp(samplePath, logsContainer, uniqueTaskHubName);
try
{
// Run the test
await testAction(appProcess, logsContainer);
}
catch (OperationCanceledException e)
{
throw new TimeoutException("Core test logic timed out!", e);
}
finally
{
logsContainer.CompleteAdding();
await this.StopProcessAsync(appProcess);
}
}
private sealed record OutputLog(DateTime Timestamp, LogLevel Level, string Message);
/// <summary>
/// Writes a line to the process's stdin and flushes it.
/// Logs the input being sent for debugging purposes.
/// </summary>
private async Task WriteInputAsync(Process process, string input, CancellationToken cancellationToken)
{
this._outputHelper.WriteLine($"{DateTime.Now:HH:mm:ss.fff} [{process.ProcessName}(in)]: {input}");
await process.StandardInput.WriteLineAsync(input);
await process.StandardInput.FlushAsync(cancellationToken);
}
/// <summary>
/// Reads a line from the logs queue, filtering for Information level logs (stdout).
/// Returns null if the collection is completed and empty, or if cancellation is requested.
/// </summary>
private string? ReadLogLine(BlockingCollection<OutputLog> logs, CancellationToken cancellationToken)
{
try
{
while (!cancellationToken.IsCancellationRequested)
{
// Block until a log entry is available or cancellation is requested
// Take will throw OperationCanceledException if cancelled, or InvalidOperationException if collection is completed
OutputLog log = logs.Take(cancellationToken);
// Check for unhandled exceptions in the logs, which are never expected (but can happen)
if (log.Message.Contains("Unhandled exception"))
{
Assert.Fail("Console app encountered an unhandled exception.");
}
// Only return Information level logs (stdout), skip Error logs (stderr)
if (log.Level == LogLevel.Information)
{
return log.Message;
}
}
}
catch (OperationCanceledException)
{
// Cancellation requested
return null;
}
catch (InvalidOperationException)
{
// Collection is completed and empty
return null;
}
return null;
}
private async Task BuildSampleAsync(string samplePath)
{
this._outputHelper.WriteLine($"Building sample at {samplePath}...");
ProcessStartInfo buildInfo = new()
{
FileName = "dotnet",
Arguments = $"build --framework {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 StartConsoleApp(string samplePath, BlockingCollection<OutputLog> logs, string taskHubName)
{
ProcessStartInfo startInfo = new()
{
FileName = "dotnet",
Arguments = $"run --no-build --framework {s_dotnetTargetFramework}",
WorkingDirectory = samplePath,
UseShellExecute = false,
RedirectStandardOutput = true,
RedirectStandardError = true,
RedirectStandardInput = 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.");
void SetAndLogEnvironmentVariable(string key, string value)
{
this._outputHelper.WriteLine($"Setting environment variable for {startInfo.FileName} sub-process: {key}={value}");
startInfo.EnvironmentVariables[key] = value;
}
// Set required environment variables for the app
SetAndLogEnvironmentVariable("AZURE_OPENAI_ENDPOINT", openAiEndpoint);
SetAndLogEnvironmentVariable("AZURE_OPENAI_DEPLOYMENT_NAME", openAiDeployment);
SetAndLogEnvironmentVariable("DURABLE_TASK_SCHEDULER_CONNECTION_STRING",
$"Endpoint=http://localhost:{DtsPort};TaskHub={taskHubName};Authentication=None");
SetAndLogEnvironmentVariable("REDIS_CONNECTION_STRING", $"localhost:{RedisPort}");
Process process = new() { StartInfo = startInfo };
// Capture the output and error streams asynchronously
// These events fire asynchronously, so we add to the blocking collection which is thread-safe
process.ErrorDataReceived += (sender, e) =>
{
if (e.Data != null)
{
string logMessage = $"{DateTime.Now:HH:mm:ss.fff} [{startInfo.FileName}(err)]: {e.Data}";
this._outputHelper.WriteLine(logMessage);
Debug.WriteLine(logMessage);
try
{
logs.Add(new OutputLog(DateTime.Now, LogLevel.Error, e.Data));
}
catch (InvalidOperationException)
{
// Collection is completed, ignore
}
}
};
process.OutputDataReceived += (sender, e) =>
{
if (e.Data != null)
{
string logMessage = $"{DateTime.Now:HH:mm:ss.fff} [{startInfo.FileName}(out)]: {e.Data}";
this._outputHelper.WriteLine(logMessage);
Debug.WriteLine(logMessage);
try
{
logs.Add(new OutputLog(DateTime.Now, LogLevel.Information, e.Data));
}
catch (InvalidOperationException)
{
// Collection is completed, ignore
}
}
};
if (!process.Start())
{
throw new InvalidOperationException("Failed to start the console app");
}
process.BeginErrorReadLine();
process.BeginOutputReadLine();
return process;
}
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($"{DateTime.Now:HH:mm:ss.fff} 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($"{DateTime.Now:HH:mm:ss.fff} Process exited: {process.Id}");
}
}
catch (Exception ex)
{
this._outputHelper.WriteLine($"{DateTime.Now:HH:mm:ss.fff} Failed to stop process: {ex.Message}");
}
}
private CancellationTokenSource CreateTestTimeoutCts(TimeSpan? timeout = null)
{
TimeSpan testTimeout = Debugger.IsAttached ? TimeSpan.FromMinutes(5) : timeout ?? TimeSpan.FromSeconds(60);
return new CancellationTokenSource(testTimeout);
}
}
dotnet/tests/Microsoft.Agents.AI.DurableTask.IntegrationTests/SamplesValidationBase.cs
+451
View File
@@ -0,0 +1,451 @@
// Copyright (c) Microsoft. All rights reserved.
using System.Collections.Concurrent;
using System.Diagnostics;
using System.Reflection;
using Microsoft.Extensions.Configuration;
using Microsoft.Extensions.Logging;
namespace Microsoft.Agents.AI.DurableTask.IntegrationTests;
/// <summary>
/// Base class for sample validation integration tests providing shared infrastructure
/// setup and utility methods for running console app samples.
/// </summary>
public abstract class SamplesValidationBase : IAsyncLifetime
{
protected const string DtsPort = "8080";
protected const string RedisPort = "6379";
protected static readonly string DotnetTargetFramework = GetTargetFramework();
protected static readonly IConfiguration Configuration =
new ConfigurationBuilder()
.AddUserSecrets(Assembly.GetExecutingAssembly())
.AddEnvironmentVariables()
.Build();
// Semaphores for thread-safe initialization of shared infrastructure.
// xUnit may run tests in parallel, so we need to ensure that DTS emulator and Redis
// are started only once across all test instances. Using SemaphoreSlim allows async-safe
// locking, and the double-check pattern (check flag, acquire lock, check flag again)
// minimizes lock contention after initialization is complete.
private static readonly SemaphoreSlim s_dtsInitLock = new(1, 1);
private static readonly SemaphoreSlim s_redisInitLock = new(1, 1);
private static bool s_dtsInfrastructureStarted;
private static bool s_redisInfrastructureStarted;
protected SamplesValidationBase(ITestOutputHelper outputHelper)
{
this.OutputHelper = outputHelper;
}
/// <summary>
/// Gets the test output helper for logging.
/// </summary>
protected ITestOutputHelper OutputHelper { get; }
/// <summary>
/// Gets the base path to the samples directory for this test class.
/// </summary>
protected abstract string SamplesPath { get; }
/// <summary>
/// Gets whether this test class requires Redis infrastructure.
/// </summary>
protected virtual bool RequiresRedis => false;
/// <summary>
/// Gets the task hub name prefix for this test class.
/// </summary>
protected virtual string TaskHubPrefix => "sample";
/// <inheritdoc />
public async ValueTask InitializeAsync()
{
await EnsureDtsInfrastructureStartedAsync(this.OutputHelper, this.StartDtsEmulatorAsync);
if (this.RequiresRedis)
{
await EnsureRedisInfrastructureStartedAsync(this.OutputHelper, this.StartRedisAsync);
}
await Task.Delay(TimeSpan.FromSeconds(5));
}
/// <summary>
/// Ensures DTS infrastructure is started exactly once across all test instances.
/// Static method writes to static field to avoid the code smell of instance methods modifying shared state.
/// </summary>
private static async Task EnsureDtsInfrastructureStartedAsync(ITestOutputHelper outputHelper, Func<Task> startAction)
{
if (s_dtsInfrastructureStarted)
{
return;
}
await s_dtsInitLock.WaitAsync();
try
{
if (!s_dtsInfrastructureStarted)
{
outputHelper.WriteLine("Starting shared DTS infrastructure...");
await startAction();
s_dtsInfrastructureStarted = true;
}
}
finally
{
s_dtsInitLock.Release();
}
}
/// <summary>
/// Ensures Redis infrastructure is started exactly once across all test instances.
/// Static method writes to static field to avoid the code smell of instance methods modifying shared state.
/// </summary>
private static async Task EnsureRedisInfrastructureStartedAsync(ITestOutputHelper outputHelper, Func<Task> startAction)
{
if (s_redisInfrastructureStarted)
{
return;
}
await s_redisInitLock.WaitAsync();
try
{
if (!s_redisInfrastructureStarted)
{
outputHelper.WriteLine("Starting shared Redis infrastructure...");
await startAction();
s_redisInfrastructureStarted = true;
}
}
finally
{
s_redisInitLock.Release();
}
}
/// <inheritdoc />
public ValueTask DisposeAsync()
{
GC.SuppressFinalize(this);
return default;
}
protected sealed record OutputLog(DateTime Timestamp, LogLevel Level, string Message);
/// <summary>
/// Runs a sample test by starting the console app and executing the provided test action.
/// </summary>
protected async Task RunSampleTestAsync(string samplePath, Func<Process, BlockingCollection<OutputLog>, Task> testAction)
{
string uniqueTaskHubName = $"{this.TaskHubPrefix}-{Guid.NewGuid():N}"[..^26];
using BlockingCollection<OutputLog> logsContainer = [];
using Process appProcess = this.StartConsoleApp(samplePath, logsContainer, uniqueTaskHubName);
try
{
await testAction(appProcess, logsContainer);
}
catch (OperationCanceledException e)
{
throw new TimeoutException("Core test logic timed out!", e);
}
finally
{
logsContainer.CompleteAdding();
await this.StopProcessAsync(appProcess);
}
}
/// <summary>
/// Writes a line to the process's stdin and flushes it.
/// </summary>
protected async Task WriteInputAsync(Process process, string input, CancellationToken cancellationToken)
{
this.OutputHelper.WriteLine($"{DateTime.Now:HH:mm:ss.fff} [{process.ProcessName}(in)]: {input}");
await process.StandardInput.WriteLineAsync(input);
await process.StandardInput.FlushAsync(cancellationToken);
}
/// <summary>
/// Reads the next Information-level log line from the queue.
/// Returns null if cancelled or collection is completed.
/// </summary>
protected string? ReadLogLine(BlockingCollection<OutputLog> logs, CancellationToken cancellationToken)
{
try
{
while (!cancellationToken.IsCancellationRequested)
{
OutputLog log = logs.Take(cancellationToken);
if (log.Message.Contains("Unhandled exception"))
{
Assert.Fail("Console app encountered an unhandled exception.");
}
if (log.Level == LogLevel.Information)
{
return log.Message;
}
}
}
catch (OperationCanceledException)
{
return null;
}
catch (InvalidOperationException)
{
return null;
}
return null;
}
/// <summary>
/// Creates a cancellation token source with the specified timeout for test operations.
/// </summary>
protected CancellationTokenSource CreateTestTimeoutCts(TimeSpan? timeout = null)
{
TimeSpan testTimeout = Debugger.IsAttached ? TimeSpan.FromMinutes(5) : timeout ?? TimeSpan.FromSeconds(60);
return new CancellationTokenSource(testTimeout);
}
/// <summary>
/// Allows derived classes to set additional environment variables for the console app process.
/// </summary>
protected virtual void ConfigureAdditionalEnvironmentVariables(ProcessStartInfo startInfo, Action<string, string> setEnvVar)
{
}
private static string GetTargetFramework()
{
string filePath = new Uri(typeof(SamplesValidationBase).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}");
}
private async Task StartDtsEmulatorAsync()
{
if (!await this.IsDtsEmulatorRunningAsync())
{
this.OutputHelper.WriteLine("Starting DTS emulator...");
await this.RunCommandAsync("docker", "run", "-d",
"--name", "dts-emulator",
"-p", $"{DtsPort}:8080",
"-e", "DTS_USE_DYNAMIC_TASK_HUBS=true",
"mcr.microsoft.com/dts/dts-emulator:latest");
}
}
private async Task StartRedisAsync()
{
if (!await this.IsRedisRunningAsync())
{
this.OutputHelper.WriteLine("Starting Redis...");
await this.RunCommandAsync("docker", "run", "-d",
"--name", "redis",
"-p", $"{RedisPort}:6379",
"redis:latest");
}
}
private async Task<bool> IsDtsEmulatorRunningAsync()
{
this.OutputHelper.WriteLine($"Checking if DTS emulator is running at http://localhost:{DtsPort}/healthz...");
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}");
}
bool isRunning = response.IsSuccessStatusCode;
this.OutputHelper.WriteLine(isRunning ? "DTS emulator is running" : $"DTS emulator not running. Status: {response.StatusCode}");
return isRunning;
}
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);
bool isRunning = process.ExitCode == 0 && output.Contains("PONG", StringComparison.OrdinalIgnoreCase);
this.OutputHelper.WriteLine(isRunning ? "Redis is running" : $"Redis not running. Exit: {process.ExitCode}, Output: {output}");
return isRunning;
}
catch (Exception ex)
{
this.OutputHelper.WriteLine($"Redis is not running: {ex.Message}");
return false;
}
}
private Process StartConsoleApp(string samplePath, BlockingCollection<OutputLog> logs, string taskHubName)
{
ProcessStartInfo startInfo = new()
{
FileName = "dotnet",
Arguments = $"run --framework {DotnetTargetFramework}",
WorkingDirectory = samplePath,
UseShellExecute = false,
RedirectStandardOutput = true,
RedirectStandardError = true,
RedirectStandardInput = true,
};
string openAiEndpoint = Configuration["AZURE_OPENAI_ENDPOINT"] ??
throw new InvalidOperationException("The required AZURE_OPENAI_ENDPOINT env variable is not set.");
string openAiDeployment = Configuration["AZURE_OPENAI_CHAT_DEPLOYMENT_NAME"] ??
throw new InvalidOperationException("The required AZURE_OPENAI_CHAT_DEPLOYMENT_NAME env variable is not set.");
void SetAndLogEnvironmentVariable(string key, string value)
{
this.OutputHelper.WriteLine($"Setting environment variable for {startInfo.FileName} sub-process: {key}={value}");
startInfo.EnvironmentVariables[key] = value;
}
SetAndLogEnvironmentVariable("AZURE_OPENAI_ENDPOINT", openAiEndpoint);
SetAndLogEnvironmentVariable("AZURE_OPENAI_DEPLOYMENT", openAiDeployment);
SetAndLogEnvironmentVariable("DURABLE_TASK_SCHEDULER_CONNECTION_STRING",
$"Endpoint=http://localhost:{DtsPort};TaskHub={taskHubName};Authentication=None");
this.ConfigureAdditionalEnvironmentVariables(startInfo, SetAndLogEnvironmentVariable);
Process process = new() { StartInfo = startInfo };
process.ErrorDataReceived += (sender, e) => this.HandleProcessOutput(e.Data, startInfo.FileName, "err", LogLevel.Error, logs);
process.OutputDataReceived += (sender, e) => this.HandleProcessOutput(e.Data, startInfo.FileName, "out", LogLevel.Information, logs);
if (!process.Start())
{
throw new InvalidOperationException("Failed to start the console app");
}
process.BeginErrorReadLine();
process.BeginOutputReadLine();
return process;
}
private void HandleProcessOutput(string? data, string processName, string stream, LogLevel level, BlockingCollection<OutputLog> logs)
{
if (data is null)
{
return;
}
string logMessage = $"{DateTime.Now:HH:mm:ss.fff} [{processName}({stream})]: {data}";
this.OutputHelper.WriteLine(logMessage);
Debug.WriteLine(logMessage);
try
{
logs.Add(new OutputLog(DateTime.Now, level, data));
}
catch (InvalidOperationException)
{
// Collection completed
}
}
private async Task RunCommandAsync(string command, params string[] args)
{
ProcessStartInfo startInfo = new()
{
FileName = command,
Arguments = string.Join(" ", args),
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 cts = new(TimeSpan.FromMinutes(1));
await process.WaitForExitAsync(cts.Token);
this.OutputHelper.WriteLine($"Command completed with exit code: {process.ExitCode}");
}
private async Task StopProcessAsync(Process process)
{
try
{
if (!process.HasExited)
{
this.OutputHelper.WriteLine($"{DateTime.Now:HH:mm:ss.fff} Killing process {process.ProcessName}#{process.Id}");
process.Kill(entireProcessTree: true);
using CancellationTokenSource cts = new(TimeSpan.FromSeconds(10));
await process.WaitForExitAsync(cts.Token);
this.OutputHelper.WriteLine($"{DateTime.Now:HH:mm:ss.fff} Process exited: {process.Id}");
}
}
catch (Exception ex)
{
this.OutputHelper.WriteLine($"{DateTime.Now:HH:mm:ss.fff} Failed to stop process: {ex.Message}");
}
}
}
dotnet/tests/Microsoft.Agents.AI.DurableTask.IntegrationTests/WorkflowConsoleAppSamplesValidation.cs
+566
View File
@@ -0,0 +1,566 @@
// Copyright (c) Microsoft. All rights reserved.
namespace Microsoft.Agents.AI.DurableTask.IntegrationTests;
/// <summary>
/// Integration tests for validating the durable workflow console app samples
/// located in samples/04-hosting/DurableWorkflows/ConsoleApps.
/// </summary>
[Collection("Samples")]
[Trait("Category", "SampleValidation")]
public sealed class WorkflowConsoleAppSamplesValidation(ITestOutputHelper outputHelper) : SamplesValidationBase(outputHelper)
{
// In CI, `dotnet run` builds samples from scratch and LLM calls add latency, so 60s is not enough.
private static readonly TimeSpan s_testTimeout = TimeSpan.FromSeconds(180);
private static readonly string s_samplesPath = Path.GetFullPath(
Path.Combine(AppDomain.CurrentDomain.BaseDirectory, "..", "..", "..", "..", "..", "samples", "04-hosting", "DurableWorkflows", "ConsoleApps"));
/// <inheritdoc />
protected override string SamplesPath => s_samplesPath;
/// <inheritdoc />
protected override string TaskHubPrefix => "workflow";
[Fact]
public async Task SequentialWorkflowSampleValidationAsync()
{
using CancellationTokenSource testTimeoutCts = this.CreateTestTimeoutCts(s_testTimeout);
string samplePath = Path.Combine(s_samplesPath, "01_SequentialWorkflow");
await this.RunSampleTestAsync(samplePath, async (process, logs) =>
{
bool inputSent = false;
bool workflowCompleted = false;
bool foundOrderLookup = false;
bool foundOrderCancel = false;
bool foundSendEmail = false;
string? line;
while ((line = this.ReadLogLine(logs, testTimeoutCts.Token)) != null)
{
if (!inputSent && line.Contains("Enter an order ID", StringComparison.OrdinalIgnoreCase))
{
await this.WriteInputAsync(process, "12345", testTimeoutCts.Token);
inputSent = true;
}
if (inputSent)
{
foundOrderLookup |= line.Contains("[Activity] OrderLookup:", StringComparison.Ordinal);
foundOrderCancel |= line.Contains("[Activity] OrderCancel:", StringComparison.Ordinal);
foundSendEmail |= line.Contains("[Activity] SendEmail:", StringComparison.Ordinal);
if (line.Contains("Workflow completed. Cancellation email sent for order 12345", StringComparison.OrdinalIgnoreCase))
{
workflowCompleted = true;
break;
}
}
this.AssertNoError(line);
}
Assert.True(inputSent, "Input was not sent to the workflow.");
Assert.True(foundOrderLookup, "OrderLookup executor log entry not found.");
Assert.True(foundOrderCancel, "OrderCancel executor log entry not found.");
Assert.True(foundSendEmail, "SendEmail executor log entry not found.");
Assert.True(workflowCompleted, "Workflow did not complete successfully.");
await this.WriteInputAsync(process, "exit", testTimeoutCts.Token);
});
}
[Fact]
public async Task ConcurrentWorkflowSampleValidationAsync()
{
using CancellationTokenSource testTimeoutCts = this.CreateTestTimeoutCts(s_testTimeout);
string samplePath = Path.Combine(s_samplesPath, "02_ConcurrentWorkflow");
await this.RunSampleTestAsync(samplePath, async (process, logs) =>
{
bool inputSent = false;
bool workflowCompleted = false;
bool foundParseQuestion = false;
bool foundAggregator = false;
bool foundAggregatorReceived2Responses = false;
string? line;
while ((line = this.ReadLogLine(logs, testTimeoutCts.Token)) != null)
{
if (!inputSent && line.Contains("Enter a science question", StringComparison.OrdinalIgnoreCase))
{
await this.WriteInputAsync(process, "What is gravity?", testTimeoutCts.Token);
inputSent = true;
}
if (inputSent)
{
foundParseQuestion |= line.Contains("[ParseQuestion]", StringComparison.Ordinal);
foundAggregator |= line.Contains("[Aggregator]", StringComparison.Ordinal);
foundAggregatorReceived2Responses |= line.Contains("Received 2 AI agent responses", StringComparison.Ordinal);
if (line.Contains("Aggregation complete", StringComparison.OrdinalIgnoreCase))
{
workflowCompleted = true;
break;
}
}
this.AssertNoError(line);
}
Assert.True(inputSent, "Input was not sent to the workflow.");
Assert.True(foundParseQuestion, "ParseQuestion executor log entry not found.");
Assert.True(foundAggregator, "Aggregator executor log entry not found.");
Assert.True(foundAggregatorReceived2Responses, "Aggregator did not receive 2 AI agent responses.");
Assert.True(workflowCompleted, "Workflow did not complete successfully.");
await this.WriteInputAsync(process, "exit", testTimeoutCts.Token);
});
}
[Fact]
public async Task ConditionalEdgesWorkflowSampleValidationAsync()
{
using CancellationTokenSource testTimeoutCts = this.CreateTestTimeoutCts(s_testTimeout);
string samplePath = Path.Combine(s_samplesPath, "03_ConditionalEdges");
await this.RunSampleTestAsync(samplePath, async (process, logs) =>
{
bool validOrderSent = false;
bool blockedOrderSent = false;
bool validOrderCompleted = false;
bool blockedOrderCompleted = false;
string? line;
while ((line = this.ReadLogLine(logs, testTimeoutCts.Token)) != null)
{
// Send a valid order first (no 'B' in ID)
if (!validOrderSent && line.Contains("Enter an order ID", StringComparison.OrdinalIgnoreCase))
{
await this.WriteInputAsync(process, "12345", testTimeoutCts.Token);
validOrderSent = true;
}
// Check valid order completed (routed to PaymentProcessor)
if (validOrderSent && !validOrderCompleted &&
line.Contains("PaymentReferenceNumber", StringComparison.OrdinalIgnoreCase))
{
validOrderCompleted = true;
// Send a blocked order (contains 'B')
await this.WriteInputAsync(process, "ORDER-B-999", testTimeoutCts.Token);
blockedOrderSent = true;
}
// Check blocked order completed (routed to NotifyFraud)
if (blockedOrderSent && line.Contains("flagged as fraudulent", StringComparison.OrdinalIgnoreCase))
{
blockedOrderCompleted = true;
break;
}
this.AssertNoError(line);
}
Assert.True(validOrderSent, "Valid order input was not sent.");
Assert.True(validOrderCompleted, "Valid order did not complete (PaymentProcessor path).");
Assert.True(blockedOrderSent, "Blocked order input was not sent.");
Assert.True(blockedOrderCompleted, "Blocked order did not complete (NotifyFraud path).");
await this.WriteInputAsync(process, "exit", testTimeoutCts.Token);
});
}
private void AssertNoError(string line)
{
if (line.Contains("Failed:", StringComparison.OrdinalIgnoreCase) ||
line.Contains("Error:", StringComparison.OrdinalIgnoreCase))
{
Assert.Fail($"Workflow failed: {line}");
}
}
[Fact]
public async Task WorkflowEventsSampleValidationAsync()
{
using CancellationTokenSource testTimeoutCts = this.CreateTestTimeoutCts(s_testTimeout);
string samplePath = Path.Combine(s_samplesPath, "05_WorkflowEvents");
await this.RunSampleTestAsync(samplePath, async (process, logs) =>
{
bool inputSent = false;
bool foundStartedRun = false;
bool foundExecutorInvoked = false;
bool foundExecutorCompleted = false;
bool foundLookupStarted = false;
bool foundOrderFound = false;
bool foundCancelProgress = false;
bool foundOrderCancelled = false;
bool foundEmailSent = false;
bool foundYieldedOutput = false;
bool foundWorkflowCompleted = false;
bool foundCompletionResult = false;
List<string> eventLines = [];
string? line;
while ((line = this.ReadLogLine(logs, testTimeoutCts.Token)) != null)
{
if (!inputSent && line.Contains("Enter order ID", StringComparison.OrdinalIgnoreCase))
{
await this.WriteInputAsync(process, "12345", testTimeoutCts.Token);
inputSent = true;
}
if (inputSent)
{
foundStartedRun |= line.Contains("Started run:", StringComparison.Ordinal);
foundExecutorInvoked |= line.Contains("ExecutorInvokedEvent", StringComparison.Ordinal);
foundExecutorCompleted |= line.Contains("ExecutorCompletedEvent", StringComparison.Ordinal);
foundLookupStarted |= line.Contains("[Lookup] Looking up order", StringComparison.Ordinal);
foundOrderFound |= line.Contains("[Lookup] Found:", StringComparison.Ordinal);
foundCancelProgress |= line.Contains("[Cancel]", StringComparison.Ordinal) && line.Contains('%');
foundOrderCancelled |= line.Contains("[Cancel] Done", StringComparison.Ordinal);
foundEmailSent |= line.Contains("[Email] Sent to", StringComparison.Ordinal);
foundYieldedOutput |= line.Contains("[Output]", StringComparison.Ordinal);
foundWorkflowCompleted |= line.Contains("DurableWorkflowCompletedEvent", StringComparison.Ordinal);
if (line.Contains("Completed:", StringComparison.Ordinal))
{
foundCompletionResult = line.Contains("12345", StringComparison.Ordinal);
break;
}
// Collect event lines for ordering verification
if (line.Contains("[Lookup]", StringComparison.Ordinal)
|| line.Contains("[Cancel]", StringComparison.Ordinal)
|| line.Contains("[Email]", StringComparison.Ordinal)
|| line.Contains("[Output]", StringComparison.Ordinal))
{
eventLines.Add(line);
}
}
this.AssertNoError(line);
}
Assert.True(inputSent, "Input was not sent to the workflow.");
Assert.True(foundStartedRun, "Streaming run was not started.");
Assert.True(foundExecutorInvoked, "ExecutorInvokedEvent not found in stream.");
Assert.True(foundExecutorCompleted, "ExecutorCompletedEvent not found in stream.");
Assert.True(foundLookupStarted, "OrderLookupStartedEvent not found in stream.");
Assert.True(foundOrderFound, "OrderFoundEvent not found in stream.");
Assert.True(foundCancelProgress, "CancellationProgressEvent not found in stream.");
Assert.True(foundOrderCancelled, "OrderCancelledEvent not found in stream.");
Assert.True(foundEmailSent, "EmailSentEvent not found in stream.");
Assert.True(foundYieldedOutput, "WorkflowOutputEvent not found in stream.");
Assert.True(foundWorkflowCompleted, "DurableWorkflowCompletedEvent not found in stream.");
Assert.True(foundCompletionResult, "Completion result does not contain the order ID.");
// Verify event ordering: lookup events appear before cancel events, which appear before email events
int lastLookupIndex = eventLines.FindLastIndex(l => l.Contains("[Lookup]", StringComparison.Ordinal));
int firstCancelIndex = eventLines.FindIndex(l => l.Contains("[Cancel]", StringComparison.Ordinal));
int lastCancelIndex = eventLines.FindLastIndex(l => l.Contains("[Cancel]", StringComparison.Ordinal));
int firstEmailIndex = eventLines.FindIndex(l => l.Contains("[Email]", StringComparison.Ordinal));
if (lastLookupIndex >= 0 && firstCancelIndex >= 0)
{
Assert.True(lastLookupIndex < firstCancelIndex, "Lookup events should appear before cancel events.");
}
if (lastCancelIndex >= 0 && firstEmailIndex >= 0)
{
Assert.True(lastCancelIndex < firstEmailIndex, "Cancel events should appear before email events.");
}
await this.WriteInputAsync(process, "exit", testTimeoutCts.Token);
});
}
[Fact]
public async Task WorkflowSharedStateSampleValidationAsync()
{
using CancellationTokenSource testTimeoutCts = this.CreateTestTimeoutCts(s_testTimeout);
string samplePath = Path.Combine(s_samplesPath, "06_WorkflowSharedState");
await this.RunSampleTestAsync(samplePath, async (process, logs) =>
{
bool inputSent = false;
bool foundStartedRun = false;
bool foundValidateOutput = false;
bool foundEnrichOutput = false;
bool foundPaymentOutput = false;
bool foundInvoiceOutput = false;
bool foundTaxCalculation = false;
bool foundAuditTrail = false;
bool foundWorkflowCompleted = false;
List<string> outputLines = [];
string? line;
while ((line = this.ReadLogLine(logs, testTimeoutCts.Token)) != null)
{
if (!inputSent && line.Contains("Enter an order ID", StringComparison.OrdinalIgnoreCase))
{
await this.WriteInputAsync(process, "ORD-001", testTimeoutCts.Token);
inputSent = true;
}
if (inputSent)
{
foundStartedRun |= line.Contains("Started run:", StringComparison.Ordinal);
if (line.Contains("[Output]", StringComparison.Ordinal))
{
foundValidateOutput |= line.Contains("ValidateOrder:", StringComparison.Ordinal) && line.Contains("validated", StringComparison.OrdinalIgnoreCase);
foundEnrichOutput |= line.Contains("EnrichOrder:", StringComparison.Ordinal) && line.Contains("enriched", StringComparison.OrdinalIgnoreCase);
foundPaymentOutput |= line.Contains("ProcessPayment:", StringComparison.Ordinal) && line.Contains("Payment processed", StringComparison.OrdinalIgnoreCase);
foundInvoiceOutput |= line.Contains("GenerateInvoice:", StringComparison.Ordinal) && line.Contains("Invoice complete", StringComparison.OrdinalIgnoreCase);
// Verify shared state: tax rate was read by ProcessPayment
foundTaxCalculation |= line.Contains("tax:", StringComparison.OrdinalIgnoreCase);
// Verify shared state: audit trail was accumulated across executors
foundAuditTrail |= line.Contains("Audit trail:", StringComparison.Ordinal)
&& line.Contains("ValidateOrder", StringComparison.Ordinal)
&& line.Contains("EnrichOrder", StringComparison.Ordinal)
&& line.Contains("ProcessPayment", StringComparison.Ordinal);
outputLines.Add(line);
}
foundWorkflowCompleted |= line.Contains("DurableWorkflowCompletedEvent", StringComparison.Ordinal)
|| line.Contains("Completed:", StringComparison.Ordinal);
if (line.Contains("Completed:", StringComparison.Ordinal))
{
break;
}
}
this.AssertNoError(line);
}
Assert.True(inputSent, "Input was not sent to the workflow.");
Assert.True(foundStartedRun, "Streaming run was not started.");
Assert.True(foundValidateOutput, "ValidateOrder output not found in stream.");
Assert.True(foundEnrichOutput, "EnrichOrder output not found in stream.");
Assert.True(foundPaymentOutput, "ProcessPayment output not found in stream.");
Assert.True(foundInvoiceOutput, "GenerateInvoice output not found in stream.");
Assert.True(foundTaxCalculation, "Tax calculation (shared state read) not found.");
Assert.True(foundAuditTrail, "Audit trail (shared state accumulation) not found.");
Assert.True(foundWorkflowCompleted, "Workflow completion not found in stream.");
// Verify output ordering: ValidateOrder -> EnrichOrder -> ProcessPayment -> GenerateInvoice
int validateIndex = outputLines.FindIndex(l => l.Contains("ValidateOrder:", StringComparison.Ordinal) && l.Contains("validated", StringComparison.OrdinalIgnoreCase));
int enrichIndex = outputLines.FindIndex(l => l.Contains("EnrichOrder:", StringComparison.Ordinal));
int paymentIndex = outputLines.FindIndex(l => l.Contains("ProcessPayment:", StringComparison.Ordinal));
int invoiceIndex = outputLines.FindIndex(l => l.Contains("GenerateInvoice:", StringComparison.Ordinal));
if (validateIndex >= 0 && enrichIndex >= 0)
{
Assert.True(validateIndex < enrichIndex, "ValidateOrder output should appear before EnrichOrder.");
}
if (enrichIndex >= 0 && paymentIndex >= 0)
{
Assert.True(enrichIndex < paymentIndex, "EnrichOrder output should appear before ProcessPayment.");
}
if (paymentIndex >= 0 && invoiceIndex >= 0)
{
Assert.True(paymentIndex < invoiceIndex, "ProcessPayment output should appear before GenerateInvoice.");
}
await this.WriteInputAsync(process, "exit", testTimeoutCts.Token);
});
}
[Fact]
public async Task SubWorkflowsSampleValidationAsync()
{
using CancellationTokenSource testTimeoutCts = this.CreateTestTimeoutCts(s_testTimeout);
string samplePath = Path.Combine(s_samplesPath, "07_SubWorkflows");
await this.RunSampleTestAsync(samplePath, async (process, logs) =>
{
bool inputSent = false;
bool foundOrderReceived = false;
bool foundValidatePayment = false;
bool foundAnalyzePatterns = false;
bool foundCalculateRiskScore = false;
bool foundChargePayment = false;
bool foundSelectCarrier = false;
bool foundCreateShipment = false;
bool foundOrderCompleted = false;
bool foundFraudRiskEvent = false;
bool workflowCompleted = false;
string? line;
while ((line = this.ReadLogLine(logs, testTimeoutCts.Token)) != null)
{
if (!inputSent && line.Contains("Enter an order ID", StringComparison.OrdinalIgnoreCase))
{
await this.WriteInputAsync(process, "ORD-001", testTimeoutCts.Token);
inputSent = true;
}
if (inputSent)
{
// Main workflow executors
foundOrderReceived |= line.Contains("[OrderReceived]", StringComparison.Ordinal);
foundOrderCompleted |= line.Contains("[OrderCompleted]", StringComparison.Ordinal);
// Payment sub-workflow executors
foundValidatePayment |= line.Contains("[Payment/ValidatePayment]", StringComparison.Ordinal);
foundChargePayment |= line.Contains("[Payment/ChargePayment]", StringComparison.Ordinal);
// FraudCheck sub-sub-workflow executors (nested inside Payment)
foundAnalyzePatterns |= line.Contains("[Payment/FraudCheck/AnalyzePatterns]", StringComparison.Ordinal);
foundCalculateRiskScore |= line.Contains("[Payment/FraudCheck/CalculateRiskScore]", StringComparison.Ordinal);
// Shipping sub-workflow executors
foundSelectCarrier |= line.Contains("[Shipping/SelectCarrier]", StringComparison.Ordinal);
foundCreateShipment |= line.Contains("[Shipping/CreateShipment]", StringComparison.Ordinal);
// Custom event from nested sub-workflow (streamed to client)
foundFraudRiskEvent |= line.Contains("[Event from sub-workflow] FraudRiskAssessedEvent", StringComparison.Ordinal);
if (line.Contains("Order completed", StringComparison.OrdinalIgnoreCase))
{
workflowCompleted = true;
break;
}
}
this.AssertNoError(line);
}
Assert.True(inputSent, "Input was not sent to the workflow.");
Assert.True(foundOrderReceived, "OrderReceived executor log not found.");
Assert.True(foundValidatePayment, "Payment/ValidatePayment executor log not found.");
Assert.True(foundAnalyzePatterns, "Payment/FraudCheck/AnalyzePatterns executor log not found.");
Assert.True(foundCalculateRiskScore, "Payment/FraudCheck/CalculateRiskScore executor log not found.");
Assert.True(foundChargePayment, "Payment/ChargePayment executor log not found.");
Assert.True(foundSelectCarrier, "Shipping/SelectCarrier executor log not found.");
Assert.True(foundCreateShipment, "Shipping/CreateShipment executor log not found.");
Assert.True(foundOrderCompleted, "OrderCompleted executor log not found.");
Assert.True(foundFraudRiskEvent, "FraudRiskAssessedEvent from nested sub-workflow not found.");
Assert.True(workflowCompleted, "Workflow did not complete successfully.");
await this.WriteInputAsync(process, "exit", testTimeoutCts.Token);
});
}
[Fact]
public async Task WorkflowHITLSampleValidationAsync()
{
using CancellationTokenSource testTimeoutCts = this.CreateTestTimeoutCts(s_testTimeout);
string samplePath = Path.Combine(s_samplesPath, "08_WorkflowHITL");
await this.RunSampleTestAsync(samplePath, (process, logs) =>
{
bool foundStarted = false;
bool foundManagerApprovalPause = false;
bool foundManagerApprovalInput = false;
bool foundManagerResponseSent = false;
bool foundBudgetApprovalPause = false;
bool foundBudgetResponseSent = false;
bool foundComplianceApprovalPause = false;
bool foundComplianceResponseSent = false;
bool foundWorkflowCompleted = false;
string? line;
while ((line = this.ReadLogLine(logs, testTimeoutCts.Token)) != null)
{
foundStarted |= line.Contains("Starting expense reimbursement workflow", StringComparison.Ordinal);
foundManagerApprovalPause |= line.Contains("Workflow paused at RequestPort: ManagerApproval", StringComparison.Ordinal);
foundManagerApprovalInput |= line.Contains("Approval for: Jerry", StringComparison.Ordinal);
foundManagerResponseSent |= line.Contains("Response sent: Approved=True", StringComparison.Ordinal) && foundManagerApprovalPause && !foundBudgetApprovalPause && !foundComplianceApprovalPause;
foundBudgetApprovalPause |= line.Contains("Workflow paused at RequestPort: BudgetApproval", StringComparison.Ordinal);
foundBudgetResponseSent |= line.Contains("Response sent: Approved=True", StringComparison.Ordinal) && foundBudgetApprovalPause;
foundComplianceApprovalPause |= line.Contains("Workflow paused at RequestPort: ComplianceApproval", StringComparison.Ordinal);
foundComplianceResponseSent |= line.Contains("Response sent: Approved=True", StringComparison.Ordinal) && foundComplianceApprovalPause;
if (line.Contains("Workflow completed: Expense reimbursed at", StringComparison.Ordinal))
{
foundWorkflowCompleted = true;
break;
}
this.AssertNoError(line);
}
Assert.True(foundStarted, "Workflow start message not found.");
Assert.True(foundManagerApprovalPause, "Manager approval pause not found.");
Assert.True(foundManagerApprovalInput, "Manager approval input (Jerry) not found.");
Assert.True(foundManagerResponseSent, "Manager approval response not sent.");
Assert.True(foundBudgetApprovalPause, "Budget approval pause not found.");
Assert.True(foundBudgetResponseSent, "Budget approval response not sent.");
Assert.True(foundComplianceApprovalPause, "Compliance approval pause not found.");
Assert.True(foundComplianceResponseSent, "Compliance approval response not sent.");
Assert.True(foundWorkflowCompleted, "Workflow did not complete successfully.");
return Task.CompletedTask;
});
}
[Fact]
public async Task WorkflowAndAgentsSampleValidationAsync()
{
using CancellationTokenSource testTimeoutCts = this.CreateTestTimeoutCts(s_testTimeout);
string samplePath = Path.Combine(s_samplesPath, "04_WorkflowAndAgents");
await this.RunSampleTestAsync(samplePath, (process, logs) =>
{
// Arrange
bool foundDemo1 = false;
bool foundBiologistResponse = false;
bool foundChemistResponse = false;
bool foundDemo2 = false;
bool foundPhysicsWorkflow = false;
bool foundDemo3 = false;
bool foundExpertTeamWorkflow = false;
bool foundDemo4 = false;
bool foundChemistryWorkflow = false;
bool allDemosCompleted = false;
// Act
string? line;
while ((line = this.ReadLogLine(logs, testTimeoutCts.Token)) != null)
{
foundDemo1 |= line.Contains("DEMO 1:", StringComparison.Ordinal);
foundBiologistResponse |= line.Contains("Biologist:", StringComparison.Ordinal);
foundChemistResponse |= line.Contains("Chemist:", StringComparison.Ordinal);
foundDemo2 |= line.Contains("DEMO 2:", StringComparison.Ordinal);
foundPhysicsWorkflow |= line.Contains("PhysicsExpertReview", StringComparison.Ordinal);
foundDemo3 |= line.Contains("DEMO 3:", StringComparison.Ordinal);
foundExpertTeamWorkflow |= line.Contains("ExpertTeamReview", StringComparison.Ordinal);
foundDemo4 |= line.Contains("DEMO 4:", StringComparison.Ordinal);
foundChemistryWorkflow |= line.Contains("ChemistryExpertReview", StringComparison.Ordinal);
if (line.Contains("All demos completed", StringComparison.OrdinalIgnoreCase))
{
allDemosCompleted = true;
break;
}
this.AssertNoError(line);
}
// Assert
Assert.True(foundDemo1, "DEMO 1 (Direct Agent Conversation) not found.");
Assert.True(foundBiologistResponse, "Biologist agent response not found.");
Assert.True(foundChemistResponse, "Chemist agent response not found.");
Assert.True(foundDemo2, "DEMO 2 (Single-Agent Workflow) not found.");
Assert.True(foundPhysicsWorkflow, "PhysicsExpertReview workflow not found.");
Assert.True(foundDemo3, "DEMO 3 (Multi-Agent Workflow) not found.");
Assert.True(foundExpertTeamWorkflow, "ExpertTeamReview workflow not found.");
Assert.True(foundDemo4, "DEMO 4 (Chemistry Workflow) not found.");
Assert.True(foundChemistryWorkflow, "ChemistryExpertReview workflow not found.");
Assert.True(allDemosCompleted, "Sample did not complete all demos successfully.");
return Task.CompletedTask;
});
}
}
dotnet/tests/Microsoft.Agents.AI.DurableTask.UnitTests/Microsoft.Agents.AI.DurableTask.UnitTests.csproj
+1
View File
@@ -7,6 +7,7 @@
<ItemGroup>
<ProjectReference Include="..\..\src\Microsoft.Agents.AI.DurableTask\Microsoft.Agents.AI.DurableTask.csproj" />
<ProjectReference Include="..\..\src\Microsoft.Agents.AI.Workflows\Microsoft.Agents.AI.Workflows.csproj" />
</ItemGroup>
</Project>
dotnet/tests/Microsoft.Agents.AI.DurableTask.UnitTests/Workflows/DurableActivityExecutorTests.cs
+235
View File
@@ -0,0 +1,235 @@
// Copyright (c) Microsoft. All rights reserved.
using System.Text.Json;
using Microsoft.Agents.AI.DurableTask.Workflows;
namespace Microsoft.Agents.AI.DurableTask.UnitTests.Workflows;
public sealed class DurableActivityExecutorTests
{
private static readonly JsonSerializerOptions s_camelCaseOptions = new()
{
PropertyNamingPolicy = JsonNamingPolicy.CamelCase,
PropertyNameCaseInsensitive = true
};
#region DeserializeInput
[Fact]
public void DeserializeInput_StringType_ReturnsInputAsIs()
{
// Arrange
const string Input = "hello world";
// Act
object result = DurableActivityExecutor.DeserializeInput(Input, typeof(string));
// Assert
Assert.Equal("hello world", result);
}
[Fact]
public void DeserializeInput_SimpleObject_DeserializesCorrectly()
{
// Arrange
string input = JsonSerializer.Serialize(new TestRecord("EXP-001", 100.50m), s_camelCaseOptions);
// Act
object result = DurableActivityExecutor.DeserializeInput(input, typeof(TestRecord));
// Assert
TestRecord record = Assert.IsType<TestRecord>(result);
Assert.Equal("EXP-001", record.Id);
Assert.Equal(100.50m, record.Amount);
}
[Fact]
public void DeserializeInput_StringArray_DeserializesDirectly()
{
// Arrange
string input = JsonSerializer.Serialize((string[])["a", "b", "c"]);
// Act
object result = DurableActivityExecutor.DeserializeInput(input, typeof(string[]));
// Assert
string[] array = Assert.IsType<string[]>(result);
Assert.Equal(["a", "b", "c"], array);
}
[Fact]
public void DeserializeInput_TypedArrayFromFanIn_DeserializesEachElement()
{
// Arrange — fan-in produces a JSON array of serialized strings
TestRecord r1 = new("EXP-001", 100m);
TestRecord r2 = new("EXP-002", 200m);
string[] serializedElements =
[
JsonSerializer.Serialize(r1, s_camelCaseOptions),
JsonSerializer.Serialize(r2, s_camelCaseOptions)
];
string input = JsonSerializer.Serialize(serializedElements);
// Act
object result = DurableActivityExecutor.DeserializeInput(input, typeof(TestRecord[]));
// Assert
TestRecord[] records = Assert.IsType<TestRecord[]>(result);
Assert.Equal(2, records.Length);
Assert.Equal("EXP-001", records[0].Id);
Assert.Equal(100m, records[0].Amount);
Assert.Equal("EXP-002", records[1].Id);
Assert.Equal(200m, records[1].Amount);
}
[Fact]
public void DeserializeInput_TypedArrayWithSingleElement_DeserializesCorrectly()
{
// Arrange
TestRecord r1 = new("EXP-001", 50m);
string[] serializedElements = [JsonSerializer.Serialize(r1, s_camelCaseOptions)];
string input = JsonSerializer.Serialize(serializedElements);
// Act
object result = DurableActivityExecutor.DeserializeInput(input, typeof(TestRecord[]));
// Assert
TestRecord[] records = Assert.IsType<TestRecord[]>(result);
Assert.Single(records);
Assert.Equal("EXP-001", records[0].Id);
}
[Fact]
public void DeserializeInput_TypedArrayWithNullElement_ThrowsInvalidOperationException()
{
// Arrange — one element is "null"
string input = JsonSerializer.Serialize((string[])["null"]);
// Act & Assert
Assert.Throws<InvalidOperationException>(
() => DurableActivityExecutor.DeserializeInput(input, typeof(TestRecord[])));
}
[Fact]
public void DeserializeInput_InvalidJson_ThrowsJsonException()
{
// Arrange
const string Input = "not valid json";
// Act & Assert
Assert.ThrowsAny<JsonException>(
() => DurableActivityExecutor.DeserializeInput(Input, typeof(TestRecord)));
}
#endregion
#region ResolveInputType
[Fact]
public void ResolveInputType_NullTypeName_ReturnsFirstSupportedType()
{
// Arrange
HashSet<Type> supportedTypes = [typeof(TestRecord), typeof(string)];
// Act
Type result = DurableActivityExecutor.ResolveInputType(null, supportedTypes);
// Assert
Assert.Equal(typeof(TestRecord), result);
}
[Fact]
public void ResolveInputType_EmptyTypeName_ReturnsFirstSupportedType()
{
// Arrange
HashSet<Type> supportedTypes = [typeof(TestRecord)];
// Act
Type result = DurableActivityExecutor.ResolveInputType(string.Empty, supportedTypes);
// Assert
Assert.Equal(typeof(TestRecord), result);
}
[Fact]
public void ResolveInputType_EmptySupportedTypes_DefaultsToString()
{
// Arrange
HashSet<Type> supportedTypes = [];
// Act
Type result = DurableActivityExecutor.ResolveInputType(null, supportedTypes);
// Assert
Assert.Equal(typeof(string), result);
}
[Fact]
public void ResolveInputType_MatchesByFullName()
{
// Arrange
HashSet<Type> supportedTypes = [typeof(TestRecord)];
// Act
Type result = DurableActivityExecutor.ResolveInputType(typeof(TestRecord).FullName, supportedTypes);
// Assert
Assert.Equal(typeof(TestRecord), result);
}
[Fact]
public void ResolveInputType_MatchesByName()
{
// Arrange
HashSet<Type> supportedTypes = [typeof(TestRecord)];
// Act
Type result = DurableActivityExecutor.ResolveInputType("TestRecord", supportedTypes);
// Assert
Assert.Equal(typeof(TestRecord), result);
}
[Fact]
public void ResolveInputType_StringArrayFallsBackToSupportedType()
{
// Arrange — fan-in sends string[] but executor expects TestRecord[]
HashSet<Type> supportedTypes = [typeof(TestRecord[])];
// Act
Type result = DurableActivityExecutor.ResolveInputType(typeof(string[]).FullName, supportedTypes);
// Assert
Assert.Equal(typeof(TestRecord[]), result);
}
[Fact]
public void ResolveInputType_StringFallsBackToSupportedType()
{
// Arrange — executor doesn't support string
HashSet<Type> supportedTypes = [typeof(TestRecord)];
// Act
Type result = DurableActivityExecutor.ResolveInputType(typeof(string).FullName, supportedTypes);
// Assert
Assert.Equal(typeof(TestRecord), result);
}
[Fact]
public void ResolveInputType_StringArrayRetainedWhenSupported()
{
// Arrange — executor explicitly supports string[]
HashSet<Type> supportedTypes = [typeof(string[])];
// Act
Type result = DurableActivityExecutor.ResolveInputType(typeof(string[]).FullName, supportedTypes);
// Assert
Assert.Equal(typeof(string[]), result);
}
#endregion
private sealed record TestRecord(string Id, decimal Amount);
}
dotnet/tests/Microsoft.Agents.AI.DurableTask.UnitTests/Workflows/DurableStreamingWorkflowRunTests.cs
+765
View File
@@ -0,0 +1,765 @@
// Copyright (c) Microsoft. All rights reserved.
using System.Text.Json;
using Microsoft.Agents.AI.DurableTask.Workflows;
using Microsoft.Agents.AI.Workflows;
using Microsoft.DurableTask;
using Microsoft.DurableTask.Client;
using Moq;
namespace Microsoft.Agents.AI.DurableTask.UnitTests.Workflows;
public sealed class DurableStreamingWorkflowRunTests
{
private const string InstanceId = "test-instance-123";
private const string WorkflowTestName = "TestWorkflow";
private static Workflow CreateTestWorkflow() =>
new WorkflowBuilder(new FunctionExecutor<string>("start", (_, _, _) => default))
.WithName(WorkflowTestName)
.Build();
private static OrchestrationMetadata CreateMetadata(
OrchestrationRuntimeStatus status,
string? serializedCustomStatus = null,
string? serializedOutput = null,
TaskFailureDetails? failureDetails = null)
{
return new OrchestrationMetadata(WorkflowTestName, InstanceId)
{
RuntimeStatus = status,
SerializedCustomStatus = serializedCustomStatus,
SerializedOutput = serializedOutput,
FailureDetails = failureDetails,
};
}
private static string SerializeCustomStatus(List<string> events)
{
DurableWorkflowLiveStatus status = new() { Events = events };
return JsonSerializer.Serialize(status, DurableSerialization.Options);
}
private static string SerializeCustomStatusWithPendingEvents(
List<string> events,
List<PendingRequestPortStatus> pendingEvents)
{
DurableWorkflowLiveStatus status = new() { Events = events, PendingEvents = pendingEvents };
return JsonSerializer.Serialize(status, DurableSerialization.Options);
}
private static Workflow CreateTestWorkflowWithRequestPort(string requestPortId)
{
FunctionExecutor<string> start = new("start", (_, _, _) => default);
RequestPort<string, string> requestPort = RequestPort.Create<string, string>(requestPortId);
FunctionExecutor<string> end = new("end", (_, _, _) => default);
return new WorkflowBuilder(start)
.WithName(WorkflowTestName)
.AddEdge(start, requestPort)
.AddEdge(requestPort, end)
.Build();
}
private static string SerializeWorkflowResult(string? result, List<string> events)
{
DurableWorkflowResult workflowResult = new() { Result = result, Events = events };
return JsonSerializer.Serialize(workflowResult, DurableWorkflowJsonContext.Default.DurableWorkflowResult);
}
private static string SerializeEvent(WorkflowEvent evt)
{
Type eventType = evt.GetType();
TypedPayload wrapper = new()
{
TypeName = eventType.AssemblyQualifiedName,
Data = JsonSerializer.Serialize(evt, eventType, DurableSerialization.Options)
};
return JsonSerializer.Serialize(wrapper, DurableWorkflowJsonContext.Default.TypedPayload);
}
#region Constructor and Properties
[Fact]
public void Constructor_SetsRunIdAndWorkflowName()
{
// Arrange
Mock<DurableTaskClient> mockClient = new("test");
// Act
DurableStreamingWorkflowRun run = new(mockClient.Object, InstanceId, CreateTestWorkflow());
// Assert
Assert.Equal(InstanceId, run.RunId);
Assert.Equal(WorkflowTestName, run.WorkflowName);
}
[Fact]
public void Constructor_NoWorkflowName_SetsEmptyString()
{
// Arrange
Mock<DurableTaskClient> mockClient = new("test");
Workflow workflow = new WorkflowBuilder(new FunctionExecutor<string>("start", (_, _, _) => default)).Build();
// Act
DurableStreamingWorkflowRun run = new(mockClient.Object, InstanceId, workflow);
// Assert
Assert.Equal(string.Empty, run.WorkflowName);
}
#endregion
#region GetStatusAsync
[Theory]
[InlineData(OrchestrationRuntimeStatus.Pending, DurableRunStatus.Pending)]
[InlineData(OrchestrationRuntimeStatus.Running, DurableRunStatus.Running)]
[InlineData(OrchestrationRuntimeStatus.Completed, DurableRunStatus.Completed)]
[InlineData(OrchestrationRuntimeStatus.Failed, DurableRunStatus.Failed)]
[InlineData(OrchestrationRuntimeStatus.Terminated, DurableRunStatus.Terminated)]
[InlineData(OrchestrationRuntimeStatus.Suspended, DurableRunStatus.Suspended)]
public async Task GetStatusAsync_MapsRuntimeStatusCorrectlyAsync(
OrchestrationRuntimeStatus runtimeStatus,
DurableRunStatus expectedStatus)
{
// Arrange
Mock<DurableTaskClient> mockClient = new("test");
mockClient.Setup(c => c.GetInstanceAsync(InstanceId, false, It.IsAny<CancellationToken>()))
.ReturnsAsync(CreateMetadata(runtimeStatus));
DurableStreamingWorkflowRun run = new(mockClient.Object, InstanceId, CreateTestWorkflow());
// Act
DurableRunStatus status = await run.GetStatusAsync();
// Assert
Assert.Equal(expectedStatus, status);
}
[Fact]
public async Task GetStatusAsync_InstanceNotFound_ReturnsNotFoundAsync()
{
// Arrange
Mock<DurableTaskClient> mockClient = new("test");
mockClient.Setup(c => c.GetInstanceAsync(InstanceId, false, It.IsAny<CancellationToken>()))
.ReturnsAsync((OrchestrationMetadata?)null);
DurableStreamingWorkflowRun run = new(mockClient.Object, InstanceId, CreateTestWorkflow());
// Act
DurableRunStatus status = await run.GetStatusAsync();
// Assert
Assert.Equal(DurableRunStatus.NotFound, status);
}
#endregion
#region WatchStreamAsync
[Fact]
public async Task WatchStreamAsync_InstanceNotFound_YieldsNoEventsAsync()
{
// Arrange
Mock<DurableTaskClient> mockClient = new("test");
mockClient.Setup(c => c.GetInstanceAsync(InstanceId, true, It.IsAny<CancellationToken>()))
.ReturnsAsync((OrchestrationMetadata?)null);
DurableStreamingWorkflowRun run = new(mockClient.Object, InstanceId, CreateTestWorkflow());
// Act
List<WorkflowEvent> events = [];
await foreach (WorkflowEvent evt in run.WatchStreamAsync())
{
events.Add(evt);
}
// Assert
Assert.Empty(events);
}
[Fact]
public async Task WatchStreamAsync_CompletedWithResult_YieldsCompletedEventAsync()
{
// Arrange
string serializedOutput = SerializeWorkflowResult("done", []);
Mock<DurableTaskClient> mockClient = new("test");
mockClient.Setup(c => c.GetInstanceAsync(InstanceId, true, It.IsAny<CancellationToken>()))
.ReturnsAsync(CreateMetadata(OrchestrationRuntimeStatus.Completed, serializedOutput: serializedOutput));
DurableStreamingWorkflowRun run = new(mockClient.Object, InstanceId, CreateTestWorkflow());
// Act
List<WorkflowEvent> events = [];
await foreach (WorkflowEvent evt in run.WatchStreamAsync())
{
events.Add(evt);
}
// Assert
Assert.Single(events);
DurableWorkflowCompletedEvent completedEvent = Assert.IsType<DurableWorkflowCompletedEvent>(events[0]);
Assert.Equal("done", completedEvent.Data);
}
[Fact]
public async Task WatchStreamAsync_CompletedWithEventsInOutput_YieldsEventsAndCompletionAsync()
{
// Arrange
DurableHaltRequestedEvent haltEvent = new("executor-1");
string serializedEvent = SerializeEvent(haltEvent);
string serializedOutput = SerializeWorkflowResult("result", [serializedEvent]);
Mock<DurableTaskClient> mockClient = new("test");
mockClient.Setup(c => c.GetInstanceAsync(InstanceId, true, It.IsAny<CancellationToken>()))
.ReturnsAsync(CreateMetadata(OrchestrationRuntimeStatus.Completed, serializedOutput: serializedOutput));
DurableStreamingWorkflowRun run = new(mockClient.Object, InstanceId, CreateTestWorkflow());
// Act
List<WorkflowEvent> events = [];
await foreach (WorkflowEvent evt in run.WatchStreamAsync())
{
events.Add(evt);
}
// Assert
Assert.Equal(2, events.Count);
DurableHaltRequestedEvent haltResult = Assert.IsType<DurableHaltRequestedEvent>(events[0]);
Assert.Equal("executor-1", haltResult.ExecutorId);
DurableWorkflowCompletedEvent completedResult = Assert.IsType<DurableWorkflowCompletedEvent>(events[1]);
Assert.Equal("result", completedResult.Result);
}
[Fact]
public async Task WatchStreamAsync_CompletedWithoutWrapper_YieldsFailedEventAsync()
{
// Arrange — output not wrapped in DurableWorkflowResult (indicates a bug)
Mock<DurableTaskClient> mockClient = new("test");
mockClient.Setup(c => c.GetInstanceAsync(InstanceId, true, It.IsAny<CancellationToken>()))
.ReturnsAsync(CreateMetadata(OrchestrationRuntimeStatus.Completed, serializedOutput: "\"raw output\""));
DurableStreamingWorkflowRun run = new(mockClient.Object, InstanceId, CreateTestWorkflow());
// Act
List<WorkflowEvent> events = [];
await foreach (WorkflowEvent evt in run.WatchStreamAsync())
{
events.Add(evt);
}
// Assert — yields a failed event with diagnostic message instead of crashing
Assert.Single(events);
DurableWorkflowFailedEvent failedEvent = Assert.IsType<DurableWorkflowFailedEvent>(events[0]);
Assert.Contains("could not be parsed", failedEvent.ErrorMessage);
}
[Fact]
public async Task WatchStreamAsync_Failed_YieldsFailedEventAsync()
{
// Arrange
Mock<DurableTaskClient> mockClient = new("test");
TaskFailureDetails failureDetails = new("ErrorType", "Something went wrong", null, null, null);
mockClient.Setup(c => c.GetInstanceAsync(InstanceId, true, It.IsAny<CancellationToken>()))
.ReturnsAsync(CreateMetadata(
OrchestrationRuntimeStatus.Failed,
failureDetails: failureDetails));
DurableStreamingWorkflowRun run = new(mockClient.Object, InstanceId, CreateTestWorkflow());
// Act
List<WorkflowEvent> events = [];
await foreach (WorkflowEvent evt in run.WatchStreamAsync())
{
events.Add(evt);
}
// Assert
Assert.Single(events);
DurableWorkflowFailedEvent failedEvent = Assert.IsType<DurableWorkflowFailedEvent>(events[0]);
Assert.Equal("Something went wrong", failedEvent.ErrorMessage);
Assert.NotNull(failedEvent.FailureDetails);
Assert.Equal("ErrorType", failedEvent.FailureDetails.ErrorType);
Assert.Equal("Something went wrong", failedEvent.FailureDetails.ErrorMessage);
}
[Fact]
public async Task WatchStreamAsync_FailedWithNoDetails_YieldsDefaultMessageAsync()
{
// Arrange
Mock<DurableTaskClient> mockClient = new("test");
mockClient.Setup(c => c.GetInstanceAsync(InstanceId, true, It.IsAny<CancellationToken>()))
.ReturnsAsync(CreateMetadata(OrchestrationRuntimeStatus.Failed));
DurableStreamingWorkflowRun run = new(mockClient.Object, InstanceId, CreateTestWorkflow());
// Act
List<WorkflowEvent> events = [];
await foreach (WorkflowEvent evt in run.WatchStreamAsync())
{
events.Add(evt);
}
// Assert
Assert.Single(events);
DurableWorkflowFailedEvent failedEvent = Assert.IsType<DurableWorkflowFailedEvent>(events[0]);
Assert.Equal("Workflow execution failed.", failedEvent.ErrorMessage);
Assert.Null(failedEvent.FailureDetails);
}
[Fact]
public async Task WatchStreamAsync_Terminated_YieldsFailedEventAsync()
{
// Arrange
Mock<DurableTaskClient> mockClient = new("test");
mockClient.Setup(c => c.GetInstanceAsync(InstanceId, true, It.IsAny<CancellationToken>()))
.ReturnsAsync(CreateMetadata(OrchestrationRuntimeStatus.Terminated));
DurableStreamingWorkflowRun run = new(mockClient.Object, InstanceId, CreateTestWorkflow());
// Act
List<WorkflowEvent> events = [];
await foreach (WorkflowEvent evt in run.WatchStreamAsync())
{
events.Add(evt);
}
// Assert
Assert.Single(events);
DurableWorkflowFailedEvent failedEvent = Assert.IsType<DurableWorkflowFailedEvent>(events[0]);
Assert.Equal("Workflow was terminated.", failedEvent.ErrorMessage);
Assert.Null(failedEvent.FailureDetails);
}
[Fact]
public async Task WatchStreamAsync_EventsInCustomStatus_YieldsEventsBeforeCompletionAsync()
{
// Arrange
DurableHaltRequestedEvent haltEvent = new("exec-1");
string serializedEvent = SerializeEvent(haltEvent);
string customStatus = SerializeCustomStatus([serializedEvent]);
string serializedOutput = SerializeWorkflowResult("final", []);
int callCount = 0;
Mock<DurableTaskClient> mockClient = new("test");
mockClient.Setup(c => c.GetInstanceAsync(InstanceId, true, It.IsAny<CancellationToken>()))
.ReturnsAsync(() =>
{
callCount++;
if (callCount == 1)
{
return CreateMetadata(OrchestrationRuntimeStatus.Running, serializedCustomStatus: customStatus);
}
return CreateMetadata(OrchestrationRuntimeStatus.Completed, serializedOutput: serializedOutput);
});
DurableStreamingWorkflowRun run = new(mockClient.Object, InstanceId, CreateTestWorkflow());
// Act
List<WorkflowEvent> events = [];
await foreach (WorkflowEvent evt in run.WatchStreamAsync())
{
events.Add(evt);
}
// Assert
Assert.Equal(2, events.Count);
DurableHaltRequestedEvent haltResult = Assert.IsType<DurableHaltRequestedEvent>(events[0]);
Assert.Equal("exec-1", haltResult.ExecutorId);
DurableWorkflowCompletedEvent completedResult = Assert.IsType<DurableWorkflowCompletedEvent>(events[1]);
Assert.Equal("final", completedResult.Result);
}
[Fact]
public async Task WatchStreamAsync_IncrementalEvents_YieldsOnlyNewEventsPerPollAsync()
{
// Arrange — simulate 3 poll cycles where events accumulate in custom status,
// then a final completion poll. This validates:
// 1. Events arriving across multiple poll cycles are yielded incrementally
// 2. Already-seen events are not re-yielded (lastReadEventIndex dedup)
// 3. Completion event follows all streamed events
DurableHaltRequestedEvent event1 = new("executor-1");
DurableHaltRequestedEvent event2 = new("executor-2");
DurableHaltRequestedEvent event3 = new("executor-3");
string serializedEvent1 = SerializeEvent(event1);
string serializedEvent2 = SerializeEvent(event2);
string serializedEvent3 = SerializeEvent(event3);
// Poll 1: 1 event in custom status
string customStatus1 = SerializeCustomStatus([serializedEvent1]);
// Poll 2: same event + 1 new event (accumulating list)
string customStatus2 = SerializeCustomStatus([serializedEvent1, serializedEvent2]);
// Poll 3: all 3 events accumulated
string customStatus3 = SerializeCustomStatus([serializedEvent1, serializedEvent2, serializedEvent3]);
// Poll 4: completed, all events also in output
string serializedOutput = SerializeWorkflowResult("done", [serializedEvent1, serializedEvent2, serializedEvent3]);
int callCount = 0;
Mock<DurableTaskClient> mockClient = new("test");
mockClient.Setup(c => c.GetInstanceAsync(InstanceId, true, It.IsAny<CancellationToken>()))
.ReturnsAsync(() =>
{
callCount++;
return callCount switch
{
1 => CreateMetadata(OrchestrationRuntimeStatus.Running, serializedCustomStatus: customStatus1),
2 => CreateMetadata(OrchestrationRuntimeStatus.Running, serializedCustomStatus: customStatus2),
3 => CreateMetadata(OrchestrationRuntimeStatus.Running, serializedCustomStatus: customStatus3),
_ => CreateMetadata(OrchestrationRuntimeStatus.Completed, serializedOutput: serializedOutput),
};
});
DurableStreamingWorkflowRun run = new(mockClient.Object, InstanceId, CreateTestWorkflow());
// Act
List<WorkflowEvent> events = [];
await foreach (WorkflowEvent evt in run.WatchStreamAsync())
{
events.Add(evt);
}
// Assert — exactly 4 events: 3 incremental halt events + 1 completion
Assert.Equal(4, events.Count);
DurableHaltRequestedEvent halt1 = Assert.IsType<DurableHaltRequestedEvent>(events[0]);
DurableHaltRequestedEvent halt2 = Assert.IsType<DurableHaltRequestedEvent>(events[1]);
DurableHaltRequestedEvent halt3 = Assert.IsType<DurableHaltRequestedEvent>(events[2]);
Assert.Equal("executor-1", halt1.ExecutorId);
Assert.Equal("executor-2", halt2.ExecutorId);
Assert.Equal("executor-3", halt3.ExecutorId);
DurableWorkflowCompletedEvent completed = Assert.IsType<DurableWorkflowCompletedEvent>(events[3]);
Assert.Equal("done", completed.Data);
}
[Fact]
public async Task WatchStreamAsync_NoNewEventsOnRepoll_DoesNotDuplicateAsync()
{
// Arrange — simulate polling where custom status doesn't change between polls,
// validating that events are not duplicated when the list is unchanged.
DurableHaltRequestedEvent event1 = new("executor-1");
string serializedEvent1 = SerializeEvent(event1);
string customStatus = SerializeCustomStatus([serializedEvent1]);
string serializedOutput = SerializeWorkflowResult("result", [serializedEvent1]);
int callCount = 0;
Mock<DurableTaskClient> mockClient = new("test");
mockClient.Setup(c => c.GetInstanceAsync(InstanceId, true, It.IsAny<CancellationToken>()))
.ReturnsAsync(() =>
{
callCount++;
return callCount switch
{
// First 3 polls return the same custom status (no new events after first)
<= 3 => CreateMetadata(OrchestrationRuntimeStatus.Running, serializedCustomStatus: customStatus),
_ => CreateMetadata(OrchestrationRuntimeStatus.Completed, serializedOutput: serializedOutput),
};
});
DurableStreamingWorkflowRun run = new(mockClient.Object, InstanceId, CreateTestWorkflow());
// Act
List<WorkflowEvent> events = [];
await foreach (WorkflowEvent evt in run.WatchStreamAsync())
{
events.Add(evt);
}
// Assert — event1 appears exactly once despite 3 polls with the same status
Assert.Equal(2, events.Count);
DurableHaltRequestedEvent haltResult = Assert.IsType<DurableHaltRequestedEvent>(events[0]);
Assert.Equal("executor-1", haltResult.ExecutorId);
DurableWorkflowCompletedEvent completedResult = Assert.IsType<DurableWorkflowCompletedEvent>(events[1]);
Assert.Equal("result", completedResult.Result);
}
[Fact]
public async Task WatchStreamAsync_Cancellation_EndsGracefullyAsync()
{
// Arrange
using CancellationTokenSource cts = new();
int pollCount = 0;
Mock<DurableTaskClient> mockClient = new("test");
mockClient.Setup(c => c.GetInstanceAsync(InstanceId, true, It.IsAny<CancellationToken>()))
.ReturnsAsync(() =>
{
if (++pollCount >= 2)
{
cts.Cancel();
}
return CreateMetadata(OrchestrationRuntimeStatus.Running);
});
DurableStreamingWorkflowRun run = new(mockClient.Object, InstanceId, CreateTestWorkflow());
// Act
List<WorkflowEvent> events = [];
await foreach (WorkflowEvent evt in run.WatchStreamAsync(cts.Token))
{
events.Add(evt);
}
// Assert — no exception thrown, stream ends cleanly
Assert.Empty(events);
}
[Fact]
public async Task WatchStreamAsync_PendingRequestPort_YieldsWaitingForInputEventAsync()
{
// Arrange
string customStatus = SerializeCustomStatusWithPendingEvents(
[],
[new PendingRequestPortStatus("ApprovalPort", """{"amount":100}""")]);
string serializedOutput = SerializeWorkflowResult("approved", []);
int callCount = 0;
Mock<DurableTaskClient> mockClient = new("test");
mockClient.Setup(c => c.GetInstanceAsync(InstanceId, true, It.IsAny<CancellationToken>()))
.ReturnsAsync(() =>
{
callCount++;
return callCount == 1
? CreateMetadata(OrchestrationRuntimeStatus.Running, serializedCustomStatus: customStatus)
: CreateMetadata(OrchestrationRuntimeStatus.Completed, serializedOutput: serializedOutput);
});
Workflow workflow = CreateTestWorkflowWithRequestPort("ApprovalPort");
DurableStreamingWorkflowRun run = new(mockClient.Object, InstanceId, workflow);
// Act
List<WorkflowEvent> events = [];
await foreach (WorkflowEvent evt in run.WatchStreamAsync())
{
events.Add(evt);
}
// Assert
Assert.Equal(2, events.Count);
DurableWorkflowWaitingForInputEvent waitingEvent = Assert.IsType<DurableWorkflowWaitingForInputEvent>(events[0]);
Assert.Equal("ApprovalPort", waitingEvent.RequestPort.Id);
Assert.Contains("amount", waitingEvent.Input);
DurableWorkflowCompletedEvent completedEvent = Assert.IsType<DurableWorkflowCompletedEvent>(events[1]);
Assert.Equal("approved", completedEvent.Result);
}
[Fact]
public async Task WatchStreamAsync_PendingRequestPort_DoesNotDuplicateOnSubsequentPollsAsync()
{
// Arrange — same pending event across 2 polls, then completion
string customStatus = SerializeCustomStatusWithPendingEvents(
[],
[new PendingRequestPortStatus("ApprovalPort", """{"amount":100}""")]);
string serializedOutput = SerializeWorkflowResult("done", []);
int callCount = 0;
Mock<DurableTaskClient> mockClient = new("test");
mockClient.Setup(c => c.GetInstanceAsync(InstanceId, true, It.IsAny<CancellationToken>()))
.ReturnsAsync(() =>
{
callCount++;
return callCount switch
{
<= 2 => CreateMetadata(OrchestrationRuntimeStatus.Running, serializedCustomStatus: customStatus),
_ => CreateMetadata(OrchestrationRuntimeStatus.Completed, serializedOutput: serializedOutput),
};
});
Workflow workflow = CreateTestWorkflowWithRequestPort("ApprovalPort");
DurableStreamingWorkflowRun run = new(mockClient.Object, InstanceId, workflow);
// Act
List<WorkflowEvent> events = [];
await foreach (WorkflowEvent evt in run.WatchStreamAsync())
{
events.Add(evt);
}
// Assert — WaitingForInputEvent yielded only once despite 2 polls
Assert.Equal(2, events.Count);
Assert.IsType<DurableWorkflowWaitingForInputEvent>(events[0]);
Assert.IsType<DurableWorkflowCompletedEvent>(events[1]);
}
#endregion
#region SendResponseAsync
[Fact]
public async Task SendResponseAsync_SerializesAndRaisesEventAsync()
{
// Arrange
Mock<DurableTaskClient> mockClient = new("test");
mockClient.Setup(c => c.RaiseEventAsync(
InstanceId,
"ApprovalPort",
It.IsAny<string>(),
It.IsAny<CancellationToken>()))
.Returns(Task.CompletedTask);
RequestPort approvalPort = RequestPort.Create<string, string>("ApprovalPort");
DurableWorkflowWaitingForInputEvent requestEvent = new("""{"amount":100}""", approvalPort);
Workflow workflow = CreateTestWorkflowWithRequestPort("ApprovalPort");
DurableStreamingWorkflowRun run = new(mockClient.Object, InstanceId, workflow);
// Act
await run.SendResponseAsync(requestEvent, new { approved = true, comments = "Looks good" });
// Assert
mockClient.Verify(c => c.RaiseEventAsync(
InstanceId,
"ApprovalPort",
It.Is<string>(s => s.Contains("approved") && s.Contains("true")),
It.IsAny<CancellationToken>()), Times.Once);
}
[Fact]
public async Task SendResponseAsync_NullRequestEvent_ThrowsAsync()
{
// Arrange
Mock<DurableTaskClient> mockClient = new("test");
DurableStreamingWorkflowRun run = new(mockClient.Object, InstanceId, CreateTestWorkflow());
// Act & Assert
await Assert.ThrowsAsync<ArgumentNullException>(() =>
run.SendResponseAsync(null!, "response").AsTask());
}
#endregion
#region WaitForCompletionAsync
[Fact]
public async Task WaitForCompletionAsync_Completed_ReturnsResultAsync()
{
// Arrange
string serializedOutput = SerializeWorkflowResult("hello world", []);
Mock<DurableTaskClient> mockClient = new("test");
mockClient.Setup(c => c.WaitForInstanceCompletionAsync(InstanceId, true, It.IsAny<CancellationToken>()))
.ReturnsAsync(CreateMetadata(OrchestrationRuntimeStatus.Completed, serializedOutput: serializedOutput));
DurableStreamingWorkflowRun run = new(mockClient.Object, InstanceId, CreateTestWorkflow());
// Act
string? result = await run.WaitForCompletionAsync<string>();
// Assert
Assert.Equal("hello world", result);
}
[Fact]
public async Task WaitForCompletionAsync_Failed_ThrowsTaskFailedExceptionAsync()
{
// Arrange
Mock<DurableTaskClient> mockClient = new("test");
mockClient.Setup(c => c.WaitForInstanceCompletionAsync(InstanceId, true, It.IsAny<CancellationToken>()))
.ReturnsAsync(CreateMetadata(
OrchestrationRuntimeStatus.Failed,
failureDetails: new TaskFailureDetails("Error", "kaboom", null, null, null)));
DurableStreamingWorkflowRun run = new(mockClient.Object, InstanceId, CreateTestWorkflow());
// Act & Assert
TaskFailedException ex = await Assert.ThrowsAsync<TaskFailedException>(
() => run.WaitForCompletionAsync<string>().AsTask());
Assert.Equal("kaboom", ex.FailureDetails.ErrorMessage);
}
[Fact]
public async Task WaitForCompletionAsync_UnexpectedStatus_ThrowsAsync()
{
// Arrange
Mock<DurableTaskClient> mockClient = new("test");
mockClient.Setup(c => c.WaitForInstanceCompletionAsync(InstanceId, true, It.IsAny<CancellationToken>()))
.ReturnsAsync(CreateMetadata(OrchestrationRuntimeStatus.Terminated));
DurableStreamingWorkflowRun run = new(mockClient.Object, InstanceId, CreateTestWorkflow());
// Act & Assert
await Assert.ThrowsAsync<InvalidOperationException>(
() => run.WaitForCompletionAsync<string>().AsTask());
}
#endregion
#region ExtractResult
[Fact]
public void ExtractResult_NullOutput_ReturnsDefault()
{
// Act
string? result = DurableStreamingWorkflowRun.ExtractResult<string>(null);
// Assert
Assert.Null(result);
}
[Fact]
public void ExtractResult_WrappedStringResult_ReturnsUnwrappedString()
{
// Arrange
string serializedOutput = SerializeWorkflowResult("hello", []);
// Act
string? result = DurableStreamingWorkflowRun.ExtractResult<string>(serializedOutput);
// Assert
Assert.Equal("hello", result);
}
[Fact]
public void ExtractResult_UnwrappedOutput_ThrowsInvalidOperationException()
{
// Arrange — raw output not wrapped in DurableWorkflowResult
string serializedOutput = JsonSerializer.Serialize("raw value");
// Act & Assert
Assert.Throws<InvalidOperationException>(
() => DurableStreamingWorkflowRun.ExtractResult<string>(serializedOutput));
}
[Fact]
public void ExtractResult_WrappedObjectResult_DeserializesCorrectly()
{
// Arrange
TestPayload original = new() { Name = "test", Value = 42 };
string resultJson = JsonSerializer.Serialize(original);
string serializedOutput = SerializeWorkflowResult(resultJson, []);
// Act
TestPayload? result = DurableStreamingWorkflowRun.ExtractResult<TestPayload>(serializedOutput);
// Assert
Assert.NotNull(result);
Assert.Equal("test", result.Name);
Assert.Equal(42, result.Value);
}
[Fact]
public void ExtractResult_CamelCaseSerializedObject_DeserializesToPascalCaseMembers()
{
// Arrange — executor outputs are serialized with DurableSerialization.Options (camelCase)
TestPayload original = new() { Name = "camel", Value = 99 };
string resultJson = JsonSerializer.Serialize(original, DurableSerialization.Options);
string serializedOutput = SerializeWorkflowResult(resultJson, []);
// Act
TestPayload? result = DurableStreamingWorkflowRun.ExtractResult<TestPayload>(serializedOutput);
// Assert
Assert.NotNull(result);
Assert.Equal("camel", result.Name);
Assert.Equal(99, result.Value);
}
#endregion
private sealed class TestPayload
{
public string? Name { get; set; }
public int Value { get; set; }
}
}
dotnet/tests/Microsoft.Agents.AI.DurableTask.UnitTests/Workflows/DurableWorkflowContextTests.cs
+504
View File
@@ -0,0 +1,504 @@
// Copyright (c) Microsoft. All rights reserved.
using Microsoft.Agents.AI.DurableTask.Workflows;
using Microsoft.Agents.AI.Workflows;
namespace Microsoft.Agents.AI.DurableTask.UnitTests.Workflows;
public sealed class DurableWorkflowContextTests
{
private static FunctionExecutor<string> CreateTestExecutor(string id = "test-executor")
=> new(id, (_, _, _) => default, outputTypes: [typeof(string)]);
#region ReadStateAsync
[Fact]
public async Task ReadStateAsync_KeyExistsInInitialState_ReturnsValueAsync()
{
// Arrange
Dictionary<string, string> state = new() { ["__default__:counter"] = "42" };
DurableWorkflowContext context = new(state, CreateTestExecutor());
// Act
int? result = await context.ReadStateAsync<int>("counter");
// Assert
Assert.Equal(42, result);
}
[Fact]
public async Task ReadStateAsync_KeyDoesNotExist_ReturnsNullAsync()
{
// Arrange
DurableWorkflowContext context = new(null, CreateTestExecutor());
// Act
string? result = await context.ReadStateAsync<string>("missing");
// Assert
Assert.Null(result);
}
[Fact]
public async Task ReadStateAsync_LocalUpdateTakesPriorityOverInitialStateAsync()
{
// Arrange
Dictionary<string, string> state = new() { ["__default__:key"] = "\"old\"" };
DurableWorkflowContext context = new(state, CreateTestExecutor());
await context.QueueStateUpdateAsync("key", "new");
// Act
string? result = await context.ReadStateAsync<string>("key");
// Assert
Assert.Equal("new", result);
}
[Fact]
public async Task ReadStateAsync_ScopeCleared_IgnoresInitialStateAsync()
{
// Arrange
Dictionary<string, string> state = new() { ["__default__:key"] = "\"value\"" };
DurableWorkflowContext context = new(state, CreateTestExecutor());
await context.QueueClearScopeAsync();
// Act
string? result = await context.ReadStateAsync<string>("key");
// Assert
Assert.Null(result);
}
[Fact]
public async Task ReadStateAsync_WithNamedScope_ReadsFromCorrectScopeAsync()
{
// Arrange
Dictionary<string, string> state = new()
{
["scopeA:key"] = "\"fromA\"",
["scopeB:key"] = "\"fromB\""
};
DurableWorkflowContext context = new(state, CreateTestExecutor());
// Act
string? resultA = await context.ReadStateAsync<string>("key", "scopeA");
string? resultB = await context.ReadStateAsync<string>("key", "scopeB");
// Assert
Assert.Equal("fromA", resultA);
Assert.Equal("fromB", resultB);
}
[Theory]
[InlineData(null)]
[InlineData("")]
public async Task ReadStateAsync_NullOrEmptyKey_ThrowsArgumentExceptionAsync(string? key)
{
// Arrange
DurableWorkflowContext context = new(null, CreateTestExecutor());
// Act & Assert
await Assert.ThrowsAnyAsync<ArgumentException>(() => context.ReadStateAsync<string>(key!).AsTask());
}
#endregion
#region ReadOrInitStateAsync
[Fact]
public async Task ReadOrInitStateAsync_KeyDoesNotExist_CallsFactoryAndQueuesUpdateAsync()
{
// Arrange
DurableWorkflowContext context = new(null, CreateTestExecutor());
// Act
string result = await context.ReadOrInitStateAsync("key", () => "initialized");
// Assert
Assert.Equal("initialized", result);
Assert.True(context.StateUpdates.ContainsKey("__default__:key"));
}
[Fact]
public async Task ReadOrInitStateAsync_KeyExists_ReturnsExistingValueAsync()
{
// Arrange
Dictionary<string, string> state = new() { ["__default__:key"] = "\"existing\"" };
DurableWorkflowContext context = new(state, CreateTestExecutor());
bool factoryCalled = false;
// Act
string result = await context.ReadOrInitStateAsync("key", () =>
{
factoryCalled = true;
return "should-not-be-used";
});
// Assert
Assert.Equal("existing", result);
Assert.False(factoryCalled);
}
[Theory]
[InlineData(null)]
[InlineData("")]
public async Task ReadOrInitStateAsync_NullOrEmptyKey_ThrowsArgumentExceptionAsync(string? key)
{
// Arrange
DurableWorkflowContext context = new(null, CreateTestExecutor());
// Act & Assert
await Assert.ThrowsAnyAsync<ArgumentException>(
() => context.ReadOrInitStateAsync(key!, () => "value").AsTask());
}
[Fact]
public async Task ReadOrInitStateAsync_ValueType_MissingKey_CallsFactoryAsync()
{
// Arrange
// Validates that ReadStateAsync<int> returns null (not 0) for missing keys,
// because the return type is int? (Nullable<int>). This ensures the factory
// is correctly invoked for value types when the key does not exist.
DurableWorkflowContext context = new(null, CreateTestExecutor());
// Act
int result = await context.ReadOrInitStateAsync("counter", () => 42);
// Assert
Assert.Equal(42, result);
Assert.True(context.StateUpdates.ContainsKey("__default__:counter"));
}
[Fact]
public async Task ReadOrInitStateAsync_NullFactory_ThrowsArgumentNullExceptionAsync()
{
// Arrange
DurableWorkflowContext context = new(null, CreateTestExecutor());
// Act & Assert
await Assert.ThrowsAsync<ArgumentNullException>(
() => context.ReadOrInitStateAsync<string>("key", null!).AsTask());
}
#endregion
#region QueueStateUpdateAsync
[Fact]
public async Task QueueStateUpdateAsync_SetsValue_VisibleToSubsequentReadAsync()
{
// Arrange
DurableWorkflowContext context = new(null, CreateTestExecutor());
// Act
await context.QueueStateUpdateAsync("key", "hello");
string? result = await context.ReadStateAsync<string>("key");
// Assert
Assert.Equal("hello", result);
}
[Fact]
public async Task QueueStateUpdateAsync_NullValue_RecordsDeletionAsync()
{
// Arrange
Dictionary<string, string> state = new() { ["__default__:key"] = "\"value\"" };
DurableWorkflowContext context = new(state, CreateTestExecutor());
// Act
await context.QueueStateUpdateAsync<string>("key", null);
// Assert
Assert.True(context.StateUpdates.ContainsKey("__default__:key"));
Assert.Null(context.StateUpdates["__default__:key"]);
}
[Theory]
[InlineData(null)]
[InlineData("")]
public async Task QueueStateUpdateAsync_NullOrEmptyKey_ThrowsArgumentExceptionAsync(string? key)
{
// Arrange
DurableWorkflowContext context = new(null, CreateTestExecutor());
// Act & Assert
await Assert.ThrowsAnyAsync<ArgumentException>(
() => context.QueueStateUpdateAsync(key!, "value").AsTask());
}
#endregion
#region QueueClearScopeAsync
[Fact]
public async Task QueueClearScopeAsync_DefaultScope_ClearsStateAndPendingUpdatesAsync()
{
// Arrange
Dictionary<string, string> state = new() { ["__default__:key"] = "\"value\"" };
DurableWorkflowContext context = new(state, CreateTestExecutor());
await context.QueueStateUpdateAsync("pending", "data");
// Act
await context.QueueClearScopeAsync();
// Assert
Assert.Contains("__default__", context.ClearedScopes);
Assert.Empty(context.StateUpdates);
}
[Fact]
public async Task QueueClearScopeAsync_NamedScope_OnlyClearsThatScopeAsync()
{
// Arrange
DurableWorkflowContext context = new(null, CreateTestExecutor());
await context.QueueStateUpdateAsync("keyA", "valueA", scopeName: "scopeA");
await context.QueueStateUpdateAsync("keyB", "valueB", scopeName: "scopeB");
// Act
await context.QueueClearScopeAsync("scopeA");
// Assert
Assert.DoesNotContain("scopeA:keyA", context.StateUpdates.Keys);
Assert.Contains("scopeB:keyB", context.StateUpdates.Keys);
}
#endregion
#region ReadStateKeysAsync
[Fact]
public async Task ReadStateKeysAsync_ReturnsKeysFromInitialStateAsync()
{
// Arrange
Dictionary<string, string> state = new()
{
["__default__:alpha"] = "\"a\"",
["__default__:beta"] = "\"b\""
};
DurableWorkflowContext context = new(state, CreateTestExecutor());
// Act
HashSet<string> keys = await context.ReadStateKeysAsync();
// Assert
Assert.Equal(2, keys.Count);
Assert.Contains("alpha", keys);
Assert.Contains("beta", keys);
}
[Fact]
public async Task ReadStateKeysAsync_MergesLocalUpdatesAndDeletionsAsync()
{
// Arrange
Dictionary<string, string> state = new()
{
["__default__:existing"] = "\"val\"",
["__default__:toDelete"] = "\"val\""
};
DurableWorkflowContext context = new(state, CreateTestExecutor());
await context.QueueStateUpdateAsync("newKey", "value");
await context.QueueStateUpdateAsync<string>("toDelete", null);
// Act
HashSet<string> keys = await context.ReadStateKeysAsync();
// Assert
Assert.Contains("existing", keys);
Assert.Contains("newKey", keys);
Assert.DoesNotContain("toDelete", keys);
}
[Fact]
public async Task ReadStateKeysAsync_AfterClearScope_ExcludesInitialStateAsync()
{
// Arrange
Dictionary<string, string> state = new() { ["__default__:old"] = "\"val\"" };
DurableWorkflowContext context = new(state, CreateTestExecutor());
await context.QueueClearScopeAsync();
await context.QueueStateUpdateAsync("new", "value");
// Act
HashSet<string> keys = await context.ReadStateKeysAsync();
// Assert
Assert.DoesNotContain("old", keys);
Assert.Contains("new", keys);
}
[Fact]
public async Task ReadStateKeysAsync_WithNamedScope_OnlyReturnsKeysFromThatScopeAsync()
{
// Arrange
Dictionary<string, string> state = new()
{
["scopeA:key1"] = "\"val\"",
["scopeB:key2"] = "\"val\""
};
DurableWorkflowContext context = new(state, CreateTestExecutor());
// Act
HashSet<string> keysA = await context.ReadStateKeysAsync("scopeA");
// Assert
Assert.Single(keysA);
Assert.Contains("key1", keysA);
}
#endregion
#region AddEventAsync
[Fact]
public async Task AddEventAsync_AddsEventToCollectionAsync()
{
// Arrange
DurableWorkflowContext context = new(null, CreateTestExecutor());
WorkflowEvent evt = new ExecutorInvokedEvent("test", "test-data");
// Act
await context.AddEventAsync(evt);
// Assert
Assert.Single(context.OutboundEvents);
Assert.Same(evt, context.OutboundEvents[0]);
}
[Fact]
public async Task AddEventAsync_NullEvent_DoesNotAddAsync()
{
// Arrange
DurableWorkflowContext context = new(null, CreateTestExecutor());
// Act
#pragma warning disable CS8625 // Cannot convert null literal to non-nullable reference type.
await context.AddEventAsync(null);
#pragma warning restore CS8625
// Assert
Assert.Empty(context.OutboundEvents);
}
#endregion
#region SendMessageAsync
[Fact]
public async Task SendMessageAsync_SerializesMessageWithTypeNameAsync()
{
// Arrange
DurableWorkflowContext context = new(null, CreateTestExecutor());
// Act
await context.SendMessageAsync("hello");
// Assert
Assert.Single(context.SentMessages);
Assert.Equal(typeof(string).AssemblyQualifiedName, context.SentMessages[0].TypeName);
Assert.NotNull(context.SentMessages[0].Data);
}
[Fact]
public async Task SendMessageAsync_NullMessage_DoesNotAddAsync()
{
// Arrange
DurableWorkflowContext context = new(null, CreateTestExecutor());
// Act
#pragma warning disable CS8625 // Cannot convert null literal to non-nullable reference type.
await context.SendMessageAsync(null);
#pragma warning restore CS8625
// Assert
Assert.Empty(context.SentMessages);
}
#endregion
#region YieldOutputAsync
[Fact]
public async Task YieldOutputAsync_AddsWorkflowOutputEventAsync()
{
// Arrange
DurableWorkflowContext context = new(null, CreateTestExecutor());
// Act
await context.YieldOutputAsync("result");
// Assert
Assert.Single(context.OutboundEvents);
WorkflowOutputEvent outputEvent = Assert.IsType<WorkflowOutputEvent>(context.OutboundEvents[0]);
Assert.Equal("result", outputEvent.Data);
}
[Fact]
public async Task YieldOutputAsync_NullOutput_DoesNotAddAsync()
{
// Arrange
DurableWorkflowContext context = new(null, CreateTestExecutor());
// Act
#pragma warning disable CS8625 // Cannot convert null literal to non-nullable reference type.
await context.YieldOutputAsync(null);
#pragma warning restore CS8625
// Assert
Assert.Empty(context.OutboundEvents);
}
#endregion
#region RequestHaltAsync
[Fact]
public async Task RequestHaltAsync_SetsHaltRequestedAndAddsEventAsync()
{
// Arrange
DurableWorkflowContext context = new(null, CreateTestExecutor());
// Act
await context.RequestHaltAsync();
// Assert
Assert.True(context.HaltRequested);
Assert.Single(context.OutboundEvents);
Assert.IsType<DurableHaltRequestedEvent>(context.OutboundEvents[0]);
}
#endregion
#region Properties
[Fact]
public void TraceContext_ReturnsNull()
{
// Arrange
DurableWorkflowContext context = new(null, CreateTestExecutor());
// Assert
Assert.Null(context.TraceContext);
}
[Fact]
public void ConcurrentRunsEnabled_ReturnsFalse()
{
// Arrange
DurableWorkflowContext context = new(null, CreateTestExecutor());
// Assert
Assert.False(context.ConcurrentRunsEnabled);
}
[Fact]
public async Task Constructor_NullInitialState_CreatesEmptyStateAsync()
{
// Arrange & Act
DurableWorkflowContext context = new(null, CreateTestExecutor());
// Assert
string? result = await context.ReadStateAsync<string>("anything");
Assert.Null(result);
}
#endregion
}
dotnet/tests/Microsoft.Agents.AI.DurableTask.UnitTests/Workflows/WorkflowNamingHelperTests.cs
+90
View File
@@ -0,0 +1,90 @@
// Copyright (c) Microsoft. All rights reserved.
using Microsoft.Agents.AI.DurableTask.Workflows;
namespace Microsoft.Agents.AI.DurableTask.UnitTests.Workflows;
public sealed class WorkflowNamingHelperTests
{
[Fact]
public void ToOrchestrationFunctionName_ValidWorkflowName_ReturnsPrefixedName()
{
string result = WorkflowNamingHelper.ToOrchestrationFunctionName("MyWorkflow");
Assert.Equal("dafx-MyWorkflow", result);
}
[Theory]
[InlineData(null)]
[InlineData("")]
public void ToOrchestrationFunctionName_NullOrEmpty_ThrowsArgumentException(string? workflowName)
{
Assert.ThrowsAny<ArgumentException>(() => WorkflowNamingHelper.ToOrchestrationFunctionName(workflowName!));
}
[Fact]
public void ToWorkflowName_ValidOrchestrationFunctionName_ReturnsWorkflowName()
{
string result = WorkflowNamingHelper.ToWorkflowName("dafx-MyWorkflow");
Assert.Equal("MyWorkflow", result);
}
[Theory]
[InlineData(null)]
[InlineData("")]
public void ToWorkflowName_NullOrEmpty_ThrowsArgumentException(string? orchestrationFunctionName)
{
Assert.ThrowsAny<ArgumentException>(() => WorkflowNamingHelper.ToWorkflowName(orchestrationFunctionName!));
}
[Theory]
[InlineData("MyWorkflow")]
[InlineData("invalid-prefix-MyWorkflow")]
[InlineData("dafx")]
[InlineData("dafx-")]
public void ToWorkflowName_InvalidOrMissingPrefix_ThrowsArgumentException(string orchestrationFunctionName)
{
Assert.Throws<ArgumentException>(() => WorkflowNamingHelper.ToWorkflowName(orchestrationFunctionName));
}
[Fact]
public void GetExecutorName_SimpleExecutorId_ReturnsSameName()
{
string result = WorkflowNamingHelper.GetExecutorName("OrderParser");
Assert.Equal("OrderParser", result);
}
[Fact]
public void GetExecutorName_ExecutorIdWithGuidSuffix_ReturnsNameWithoutSuffix()
{
string result = WorkflowNamingHelper.GetExecutorName("Physicist_8884e71021334ce49517fa2b17b1695b");
Assert.Equal("Physicist", result);
}
[Fact]
public void GetExecutorName_NameWithUnderscoresAndGuidSuffix_ReturnsFullName()
{
string result = WorkflowNamingHelper.GetExecutorName("my_agent_8884e71021334ce49517fa2b17b1695b");
Assert.Equal("my_agent", result);
}
[Fact]
public void GetExecutorName_NameWithUnderscoreButNoGuidSuffix_ReturnsSameName()
{
string result = WorkflowNamingHelper.GetExecutorName("my_custom_executor");
Assert.Equal("my_custom_executor", result);
}
[Theory]
[InlineData(null)]
[InlineData("")]
public void GetExecutorName_NullOrEmpty_ThrowsArgumentException(string? executorId)
{
Assert.ThrowsAny<ArgumentException>(() => WorkflowNamingHelper.GetExecutorName(executorId!));
}
}
dotnet/tests/Microsoft.Agents.AI.Hosting.AzureFunctions.IntegrationTests/SamplesValidation.cs
+13 -3
View File
@@ -21,6 +21,12 @@ public sealed class SamplesValidation(ITestOutputHelper outputHelper) : IAsyncLi
private const string RedisPort = "6379";
private static readonly string s_dotnetTargetFramework = GetTargetFramework();
#if DEBUG
private const string BuildConfiguration = "Debug";
#else
private const string BuildConfiguration = "Release";
#endif
private static readonly HttpClient s_sharedHttpClient = new();
private static readonly IConfiguration s_configuration =
new ConfigurationBuilder()
@@ -30,6 +36,10 @@ public sealed class SamplesValidation(ITestOutputHelper outputHelper) : IAsyncLi
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"));
@@ -821,7 +831,7 @@ public sealed class SamplesValidation(ITestOutputHelper outputHelper) : IAsyncLi
ProcessStartInfo buildInfo = new()
{
FileName = "dotnet",
Arguments = $"build -f {s_dotnetTargetFramework}",
Arguments = $"build -f {s_dotnetTargetFramework} -c {BuildConfiguration}",
WorkingDirectory = samplePath,
UseShellExecute = false,
RedirectStandardOutput = true,
@@ -851,7 +861,7 @@ public sealed class SamplesValidation(ITestOutputHelper outputHelper) : IAsyncLi
ProcessStartInfo startInfo = new()
{
FileName = "dotnet",
Arguments = $"run --no-build -f {s_dotnetTargetFramework} --port {AzureFunctionsPort}",
Arguments = $"run --no-build -f {s_dotnetTargetFramework} -c {BuildConfiguration} --port {AzureFunctionsPort}",
WorkingDirectory = samplePath,
UseShellExecute = false,
RedirectStandardOutput = true,
@@ -930,7 +940,7 @@ public sealed class SamplesValidation(ITestOutputHelper outputHelper) : IAsyncLi
}
},
message: "Azure Functions Core Tools is ready",
timeout: TimeSpan.FromSeconds(60));
timeout: s_functionsReadyTimeout);
}
private async Task WaitForOrchestrationCompletionAsync(Uri statusUri)
dotnet/tests/Microsoft.Agents.AI.Hosting.AzureFunctions.IntegrationTests/WorkflowSamplesValidation.cs
+593
View File
@@ -0,0 +1,593 @@
// Copyright (c) Microsoft. All rights reserved.
using System.Diagnostics;
using System.Reflection;
using System.Text;
using Microsoft.Extensions.Configuration;
using Microsoft.Extensions.Logging;
namespace Microsoft.Agents.AI.Hosting.AzureFunctions.IntegrationTests;
/// <summary>
/// Integration tests for validating the durable workflow Azure Functions samples
/// located in samples/04-hosting/DurableWorkflows/AzureFunctions.
/// </summary>
[Collection("Samples")]
[Trait("Category", "SampleValidation")]
public sealed class WorkflowSamplesValidation(ITestOutputHelper outputHelper) : IAsyncLifetime
{
private const string AzureFunctionsPort = "7071";
private const string AzuritePort = "10000";
private const string DtsPort = "8080";
private static readonly string s_dotnetTargetFramework = GetTargetFramework();
#if DEBUG
private const string BuildConfiguration = "Debug";
#else
private const string BuildConfiguration = "Release";
#endif
private static readonly HttpClient s_sharedHttpClient = new();
private static readonly IConfiguration s_configuration =
new ConfigurationBuilder()
.AddUserSecrets(Assembly.GetExecutingAssembly())
.AddEnvironmentVariables()
.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", "DurableWorkflows", "AzureFunctions"));
private readonly ITestOutputHelper _outputHelper = outputHelper;
public async ValueTask InitializeAsync()
{
if (!s_infrastructureStarted)
{
await this.StartSharedInfrastructureAsync();
s_infrastructureStarted = true;
}
}
public ValueTask DisposeAsync()
{
GC.SuppressFinalize(this);
return default;
}
[Fact]
public async Task SequentialWorkflowSampleValidationAsync()
{
string samplePath = Path.Combine(s_samplesPath, "01_SequentialWorkflow");
await this.RunSampleTestAsync(samplePath, requiresOpenAI: false, async (logs) =>
{
// Test the CancelOrder workflow
Uri cancelOrderUri = new($"http://localhost:{AzureFunctionsPort}/api/workflows/CancelOrder/run");
this._outputHelper.WriteLine($"Starting CancelOrder workflow via POST request to {cancelOrderUri}...");
using HttpContent cancelContent = new StringContent("12345", Encoding.UTF8, "text/plain");
using HttpResponseMessage cancelResponse = await s_sharedHttpClient.PostAsync(cancelOrderUri, cancelContent);
Assert.True(cancelResponse.IsSuccessStatusCode, $"CancelOrder request failed with status: {cancelResponse.StatusCode}");
string cancelResponseText = await cancelResponse.Content.ReadAsStringAsync();
Assert.Contains("CancelOrder", cancelResponseText);
this._outputHelper.WriteLine($"CancelOrder response: {cancelResponseText}");
// Wait for the CancelOrder workflow to complete by checking logs
await this.WaitForConditionAsync(
condition: () =>
{
lock (logs)
{
bool exists = logs.Any(log => log.Message.Contains("Workflow completed"));
return Task.FromResult(exists);
}
},
message: "CancelOrder workflow completed",
timeout: s_orchestrationTimeout);
// Verify the executor activities ran in sequence
lock (logs)
{
Assert.True(logs.Any(log => log.Message.Contains("[Activity] OrderLookup:")), "OrderLookup activity not found in logs.");
Assert.True(logs.Any(log => log.Message.Contains("[Activity] OrderCancel:")), "OrderCancel activity not found in logs.");
Assert.True(logs.Any(log => log.Message.Contains("[Activity] SendEmail:")), "SendEmail activity not found in logs.");
}
// Test the OrderStatus workflow (shares OrderLookup executor with CancelOrder)
Uri orderStatusUri = new($"http://localhost:{AzureFunctionsPort}/api/workflows/OrderStatus/run");
this._outputHelper.WriteLine($"Starting OrderStatus workflow via POST request to {orderStatusUri}...");
using HttpContent statusContent = new StringContent("67890", Encoding.UTF8, "text/plain");
using HttpResponseMessage statusResponse = await s_sharedHttpClient.PostAsync(orderStatusUri, statusContent);
Assert.True(statusResponse.IsSuccessStatusCode, $"OrderStatus request failed with status: {statusResponse.StatusCode}");
string statusResponseText = await statusResponse.Content.ReadAsStringAsync();
Assert.Contains("OrderStatus", statusResponseText);
this._outputHelper.WriteLine($"OrderStatus response: {statusResponseText}");
// Wait for the OrderStatus workflow to complete
await this.WaitForConditionAsync(
condition: () =>
{
lock (logs)
{
// Look for StatusReport activity which is unique to OrderStatus workflow
bool exists = logs.Any(log => log.Message.Contains("[Activity] StatusReport:"));
return Task.FromResult(exists);
}
},
message: "OrderStatus workflow completed",
timeout: s_orchestrationTimeout);
});
}
[Fact]
public async Task HITLWorkflowSampleValidationAsync()
{
string samplePath = Path.Combine(s_samplesPath, "03_WorkflowHITL");
await this.RunSampleTestAsync(samplePath, requiresOpenAI: false, async (logs) =>
{
// Use a unique run ID to avoid conflicts with previous test runs
string runId = $"hitl-test-{Guid.NewGuid():N}";
// Step 1: Start the expense reimbursement workflow
Uri runUri = new($"http://localhost:{AzureFunctionsPort}/api/workflows/ExpenseReimbursement/run?runId={runId}");
this._outputHelper.WriteLine($"Starting ExpenseReimbursement workflow via POST request to {runUri}...");
using HttpContent runContent = new StringContent("EXP-2025-001", Encoding.UTF8, "text/plain");
using HttpResponseMessage runResponse = await s_sharedHttpClient.PostAsync(runUri, runContent);
Assert.True(runResponse.IsSuccessStatusCode, $"Run request failed with status: {runResponse.StatusCode}");
string runResponseText = await runResponse.Content.ReadAsStringAsync();
Assert.Contains("ExpenseReimbursement", runResponseText);
this._outputHelper.WriteLine($"Run response: {runResponseText}");
// Step 2: Wait for the workflow to pause at the ManagerApproval RequestPort
await this.WaitForConditionAsync(
condition: () =>
{
lock (logs)
{
bool exists = logs.Any(log => log.Message.Contains("Workflow waiting for external input at RequestPort 'ManagerApproval'"));
return Task.FromResult(exists);
}
},
message: "Workflow paused at ManagerApproval RequestPort",
timeout: s_orchestrationTimeout);
// Step 3: Send approval response to resume the workflow
Uri respondUri = new($"http://localhost:{AzureFunctionsPort}/api/workflows/ExpenseReimbursement/respond/{runId}");
this._outputHelper.WriteLine($"Sending approval response via POST request to {respondUri}...");
using HttpContent respondContent = new StringContent(
"""{"eventName": "ManagerApproval", "response": {"Approved": true, "Comments": "Approved by test."}}""",
Encoding.UTF8, "application/json");
using HttpResponseMessage respondResponse = await s_sharedHttpClient.PostAsync(respondUri, respondContent);
Assert.True(respondResponse.IsSuccessStatusCode, $"Respond request failed with status: {respondResponse.StatusCode}");
string respondResponseText = await respondResponse.Content.ReadAsStringAsync();
Assert.Contains("Response sent to workflow", respondResponseText);
this._outputHelper.WriteLine($"Respond response: {respondResponseText}");
// Step 4: Wait for the workflow to pause at the parallel BudgetApproval and ComplianceApproval RequestPorts
await this.WaitForConditionAsync(
condition: () =>
{
lock (logs)
{
bool exists = logs.Any(log => log.Message.Contains("Workflow waiting for external input at RequestPort 'BudgetApproval'"));
return Task.FromResult(exists);
}
},
message: "Workflow paused at BudgetApproval RequestPort",
timeout: s_orchestrationTimeout);
// Step 5a: Send budget approval response
this._outputHelper.WriteLine("Sending BudgetApproval response...");
using HttpContent budgetContent = new StringContent(
"""{"eventName": "BudgetApproval", "response": {"Approved": true, "Comments": "Budget approved by test."}}""",
Encoding.UTF8, "application/json");
using HttpResponseMessage budgetResponse = await s_sharedHttpClient.PostAsync(respondUri, budgetContent);
Assert.True(budgetResponse.IsSuccessStatusCode, $"BudgetApproval request failed with status: {budgetResponse.StatusCode}");
this._outputHelper.WriteLine($"BudgetApproval response: {await budgetResponse.Content.ReadAsStringAsync()}");
// Step 5b: Send compliance approval response
this._outputHelper.WriteLine("Sending ComplianceApproval response...");
using HttpContent complianceContent = new StringContent(
"""{"eventName": "ComplianceApproval", "response": {"Approved": true, "Comments": "Compliance approved by test."}}""",
Encoding.UTF8, "application/json");
using HttpResponseMessage complianceResponse = await s_sharedHttpClient.PostAsync(respondUri, complianceContent);
Assert.True(complianceResponse.IsSuccessStatusCode, $"ComplianceApproval request failed with status: {complianceResponse.StatusCode}");
this._outputHelper.WriteLine($"ComplianceApproval response: {await complianceResponse.Content.ReadAsStringAsync()}");
// Step 6: Wait for the workflow to complete
await this.WaitForConditionAsync(
condition: () =>
{
lock (logs)
{
bool exists = logs.Any(log => log.Message.Contains("Workflow completed"));
return Task.FromResult(exists);
}
},
message: "HITL workflow completed",
timeout: s_orchestrationTimeout);
// Verify executor activities ran
lock (logs)
{
Assert.True(logs.Any(log => log.Message.Contains("Received external event for RequestPort 'ManagerApproval'")),
"ManagerApproval external event receipt not found in logs.");
Assert.True(logs.Any(log => log.Message.Contains("Received external event for RequestPort 'BudgetApproval'")),
"BudgetApproval external event receipt not found in logs.");
Assert.True(logs.Any(log => log.Message.Contains("Received external event for RequestPort 'ComplianceApproval'")),
"ComplianceApproval external event receipt not found in logs.");
}
});
}
[Fact]
public async Task ConcurrentWorkflowSampleValidationAsync()
{
string samplePath = Path.Combine(s_samplesPath, "02_ConcurrentWorkflow");
await this.RunSampleTestAsync(samplePath, requiresOpenAI: true, async (logs) =>
{
// Start the ExpertReview workflow with a science question
const string RequestBody = "What is temperature?";
using HttpContent content = new StringContent(RequestBody, Encoding.UTF8, "text/plain");
Uri startUri = new($"http://localhost:{AzureFunctionsPort}/api/workflows/ExpertReview/run");
this._outputHelper.WriteLine($"Starting ExpertReview workflow via POST request to {startUri}...");
using HttpResponseMessage startResponse = await s_sharedHttpClient.PostAsync(startUri, content);
Assert.True(startResponse.IsSuccessStatusCode, $"ExpertReview request failed with status: {startResponse.StatusCode}");
string startResponseText = await startResponse.Content.ReadAsStringAsync();
Assert.Contains("ExpertReview", startResponseText);
this._outputHelper.WriteLine($"ExpertReview response: {startResponseText}");
// Wait for the ParseQuestion executor to run
await this.WaitForConditionAsync(
condition: () =>
{
lock (logs)
{
bool exists = logs.Any(log => log.Message.Contains("[ParseQuestion]"));
return Task.FromResult(exists);
}
},
message: "ParseQuestion executor ran",
timeout: s_orchestrationTimeout);
// Wait for the Aggregator to complete (indicates fan-in from parallel agents)
await this.WaitForConditionAsync(
condition: () =>
{
lock (logs)
{
bool exists = logs.Any(log => log.Message.Contains("Aggregation complete"));
return Task.FromResult(exists);
}
},
message: "Aggregator completed with parallel agent responses",
timeout: s_orchestrationTimeout);
// Verify the aggregator received responses from both AI agents
lock (logs)
{
Assert.True(
logs.Any(log => log.Message.Contains("AI agent responses")),
"Aggregator did not log receiving AI agent responses.");
}
});
}
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"]);
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"]);
await this.WaitForConditionAsync(
condition: this.IsDtsEmulatorRunningAsync,
message: "DTS emulator 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));
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...");
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 StartDockerContainerAsync(string containerName, string image, string[] ports)
{
await this.RunCommandAsync("docker", ["stop", containerName]);
await this.RunCommandAsync("docker", ["rm", containerName]);
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 sealed record OutputLog(DateTime Timestamp, LogLevel Level, string Message);
private async Task RunSampleTestAsync(string samplePath, bool requiresOpenAI, Func<IReadOnlyList<OutputLog>, Task> testAction)
{
List<OutputLog> logsContainer = [];
using Process funcProcess = this.StartFunctionApp(samplePath, logsContainer, requiresOpenAI);
try
{
await this.WaitForAzureFunctionsAsync();
await testAction(logsContainer);
}
finally
{
await this.StopProcessAsync(funcProcess);
}
}
private Process StartFunctionApp(string samplePath, List<OutputLog> logs, bool requiresOpenAI)
{
ProcessStartInfo startInfo = new()
{
FileName = "dotnet",
Arguments = $"run -f {s_dotnetTargetFramework} -c {BuildConfiguration} --port {AzureFunctionsPort}",
WorkingDirectory = samplePath,
UseShellExecute = false,
RedirectStandardOutput = true,
RedirectStandardError = true,
};
if (requiresOpenAI)
{
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_CHAT_DEPLOYMENT_NAME"] ??
throw new InvalidOperationException("The required AZURE_OPENAI_CHAT_DEPLOYMENT_NAME env variable is not set.");
this._outputHelper.WriteLine($"Using Azure OpenAI endpoint: {openAiEndpoint}, deployment: {openAiDeployment}");
startInfo.EnvironmentVariables["AZURE_OPENAI_ENDPOINT"] = openAiEndpoint;
startInfo.EnvironmentVariables["AZURE_OPENAI_DEPLOYMENT"] = openAiDeployment;
}
startInfo.EnvironmentVariables["DURABLE_TASK_SCHEDULER_CONNECTION_STRING"] =
$"Endpoint=http://localhost:{DtsPort};TaskHub=default;Authentication=None";
startInfo.EnvironmentVariables["AzureWebJobsStorage"] = "UseDevelopmentStorage=true";
Process process = new() { StartInfo = startInfo };
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)
{
return false;
}
},
message: "Azure Functions Core Tools is ready",
timeout: s_functionsReadyTimeout);
}
private async Task RunCommandAsync(string command, string[] args)
{
ProcessStartInfo startInfo = new()
{
FileName = command,
Arguments = string.Join(" ", args),
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()
{
string filePath = new Uri(typeof(WorkflowSamplesValidation).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}");
}
}