From 459f5de156d930bf0e64396c10f3210b0d682811 Mon Sep 17 00:00:00 2001 From: Stijn Moreels <9039753+stijnmoreels@users.noreply.github.com> Date: Tue, 3 Mar 2026 09:30:51 +0100 Subject: [PATCH] chore(az.df): add cancellation token support for azure data factory test fixtures --- .../06-Integration/01-data-factory.mdx | 4 +- ...cus.Testing.Integration.DataFactory.csproj | 6 +- .../TemporaryDataFlowDebugSession.cs | 272 +++++++++++++++--- .../Fixture/TemporaryDataFactoryDataFlow.cs | 47 +-- .../DataFactory/RunDataFlowTests.cs | 35 ++- .../TemporaryDataFlowDebugSessionTests.cs | 2 +- 6 files changed, 279 insertions(+), 87 deletions(-) diff --git a/docs/preview/03-Features/04-Azure/06-Integration/01-data-factory.mdx b/docs/preview/03-Features/04-Azure/06-Integration/01-data-factory.mdx index 03d271ab..d87cd4c0 100644 --- a/docs/preview/03-Features/04-Azure/06-Integration/01-data-factory.mdx +++ b/docs/preview/03-Features/04-Azure/06-Integration/01-data-factory.mdx @@ -31,7 +31,7 @@ ResourceIdentifier dataFactoryResourceId = DataFactoryResource.CreateResourceIdentifier("", "", ""); await using var session = - await TemporaryDataFlowDebugSession.StartDebugSessionAsync(dataFactoryResourceId, logger); + await TemporaryDataFlowDebugSession.StartDebugSessionAsync(dataFactoryResourceId, logger, cancellationToken); ``` > ⚡ Uses by default the `DefaultAzureCredential` but other types of authentication mechanism are supported with overloads that take in the `DataFactoryResource` directly: @@ -193,7 +193,7 @@ using Arcus.Testing; await using TemporaryDataFlowDebugSession session = ... DataFlowRunResult result = - await session.RunDataFlowAsync("", ""); + await session.RunDataFlowAsync("", "", cancellationToken); // The run status of data preview, statistics or expression preview. string status = result.Status; diff --git a/src/Arcus.Testing.Integration.DataFactory/Arcus.Testing.Integration.DataFactory.csproj b/src/Arcus.Testing.Integration.DataFactory/Arcus.Testing.Integration.DataFactory.csproj index 5d83921c..8b9e42a1 100644 --- a/src/Arcus.Testing.Integration.DataFactory/Arcus.Testing.Integration.DataFactory.csproj +++ b/src/Arcus.Testing.Integration.DataFactory/Arcus.Testing.Integration.DataFactory.csproj @@ -22,6 +22,10 @@ All + + + + @@ -31,7 +35,7 @@ - + diff --git a/src/Arcus.Testing.Integration.DataFactory/TemporaryDataFlowDebugSession.cs b/src/Arcus.Testing.Integration.DataFactory/TemporaryDataFlowDebugSession.cs index f49eabe9..3b4667d4 100644 --- a/src/Arcus.Testing.Integration.DataFactory/TemporaryDataFlowDebugSession.cs +++ b/src/Arcus.Testing.Integration.DataFactory/TemporaryDataFlowDebugSession.cs @@ -2,6 +2,7 @@ using System.Collections.Generic; using System.Collections.ObjectModel; using System.Linq; +using System.Threading; using System.Threading.Tasks; using Azure; using Azure.Core; @@ -65,6 +66,7 @@ public class TemporaryDataFlowDebugSession : IAsyncDisposable private readonly ILogger _logger; private bool _isDisposed; + private readonly DataFactoryResource _dataFactory; private TemporaryDataFlowDebugSession(bool startedByUs, Guid sessionId, DataFactoryResource resource, ILogger logger) { @@ -72,15 +74,22 @@ private TemporaryDataFlowDebugSession(bool startedByUs, Guid sessionId, DataFact _startedByUs = startedByUs; _sessionId = sessionId; + _dataFactory = resource; _logger = logger ?? NullLogger.Instance; - - DataFactory = resource; } /// /// Gets the Azure Data Factory resource where the active data flow debug session is started. /// - private DataFactoryResource DataFactory { get; } + /// Thrown when the test fixture was already teared down. + private DataFactoryResource DataFactory + { + get + { + ObjectDisposedException.ThrowIf(_isDisposed, this); + return _dataFactory; + } + } /// /// Gets the session ID of the active data flow debug session. @@ -115,6 +124,7 @@ public Guid SessionId /// The logger to write diagnostic messages during the debug session. /// Thrown when the is null. /// Thrown when the starting of the data flow debug session did not result in a session ID. + [Obsolete("Will be removed in v3.0, please use the " + nameof(StartDebugSessionAsync) + " overload with cancellation token support", DiagnosticId = ObsoleteDefaults.DiagnosticId)] public static Task StartDebugSessionAsync(ResourceIdentifier dataFactoryResourceId, ILogger logger) { return StartDebugSessionAsync(dataFactoryResourceId, logger, configureOptions: null); @@ -125,7 +135,33 @@ public static Task StartDebugSessionAsync(Resourc /// /// /// Uses for authentication; - /// use the overload to provide a custom authentication mechanism. + /// use the overload to provide a custom authentication mechanism. + /// + /// + /// The resource ID to the Azure Data Factory instance where to start the active data flow debug session. + /// The resource ID can be constructed via : + /// + /// + /// ResourceIdentifier dataFactoryResourceId = + /// DataFactoryResource.CreateResourceIdentifier("<subscription-id>", "<resource-group>", "<factory-name>"); + /// + /// + /// + /// The logger to write diagnostic messages during the debug session. + /// The optional token to propagate notifications that the operation should be cancelled. + /// Thrown when the is null. + /// Thrown when the starting of the data flow debug session did not result in a session ID. + public static Task StartDebugSessionAsync(ResourceIdentifier dataFactoryResourceId, ILogger logger, CancellationToken cancellationToken) + { + return StartDebugSessionAsync(dataFactoryResourceId, logger, configureOptions: null, cancellationToken); + } + + /// + /// Starts a new active Azure Data Factory data flow debug session for the given . + /// + /// + /// Uses for authentication; + /// use the overload to provide a custom authentication mechanism. /// /// /// The resource ID to the Azure Data Factory instance where to start the active data flow debug session. @@ -140,17 +176,49 @@ public static Task StartDebugSessionAsync(Resourc /// The function to configure the options of the debug session. /// Thrown when the is null. /// Thrown when the starting of the data flow debug session did not result in a session ID. + [Obsolete("Will be removed in v3.0, please use the " + nameof(StartDebugSessionAsync) + " overload with cancellation token support", DiagnosticId = ObsoleteDefaults.DiagnosticId)] public static Task StartDebugSessionAsync( ResourceIdentifier dataFactoryResourceId, ILogger logger, Action configureOptions) { + return StartDebugSessionAsync(dataFactoryResourceId, logger, configureOptions, CancellationToken.None); + } + + /// + /// Starts a new active Azure Data Factory data flow debug session for the given . + /// + /// + /// Uses for authentication; + /// use the overload to provide a custom authentication mechanism. + /// + /// + /// The resource ID to the Azure Data Factory instance where to start the active data flow debug session. + /// The resource ID can be constructed via : + /// + /// + /// ResourceIdentifier dataFactoryResourceId = + /// DataFactoryResource.CreateResourceIdentifier("<subscription-id>", "<resource-group>", "<factory-name>"); + /// + /// + /// /// The logger to write diagnostic messages during the debug session. + /// The function to configure the options of the debug session. + /// The optional token to propagate notifications that the operation should be cancelled. + /// Thrown when the is null. + /// Thrown when the starting of the data flow debug session did not result in a session ID. + public static Task StartDebugSessionAsync( + ResourceIdentifier dataFactoryResourceId, + ILogger logger, + Action configureOptions, + CancellationToken cancellationToken) + { + cancellationToken.ThrowIfCancellationRequested(); ArgumentNullException.ThrowIfNull(dataFactoryResourceId); var armClient = new ArmClient(new DefaultAzureCredential()); DataFactoryResource resource = armClient.GetDataFactoryResource(dataFactoryResourceId); - return StartDebugSessionAsync(resource, logger, configureOptions); + return StartDebugSessionAsync(resource, logger, configureOptions, cancellationToken); } /// @@ -170,11 +238,35 @@ public static Task StartDebugSessionAsync( /// The logger to write diagnostic messages during the debug session. /// Thrown when the is null. /// Thrown when the starting of the data flow debug session did not result in a session ID. + [Obsolete("Will be removed in v3.0, please use the " + nameof(StartDebugSessionAsync) + " overload with cancellation token support", DiagnosticId = ObsoleteDefaults.DiagnosticId)] public static Task StartDebugSessionAsync(DataFactoryResource resource, ILogger logger) { return StartDebugSessionAsync(resource, logger, configureOptions: null); } + /// + /// Starts a new active Azure Data Factory data flow debug session for the given . + /// + /// + /// The resource to start the active data flow debug session for. + /// The resource should be retrieved via the : + /// + /// + /// var credential = new DefaultAzureCredential(); + /// var arm = new ArmClient(credential); + /// DataFactoryResource resource = arm.GetDataFactoryResource(dataFactoryResourceId); + /// + /// + /// + /// The logger to write diagnostic messages during the debug session. + /// The optional token to propagate notifications that the operation should be cancelled. + /// Thrown when the is null. + /// Thrown when the starting of the data flow debug session did not result in a session ID. + public static Task StartDebugSessionAsync(DataFactoryResource resource, ILogger logger, CancellationToken cancellationToken) + { + return StartDebugSessionAsync(resource, logger, configureOptions: null, cancellationToken); + } + /// /// Starts a new active Azure Data Factory data flow debug session for the given . /// @@ -193,18 +285,48 @@ public static Task StartDebugSessionAsync(DataFac /// The function to configure the options of the debug session. /// Thrown when the is null. /// Thrown when the starting of the data flow debug session did not result in a session ID. - public static async Task StartDebugSessionAsync( + [Obsolete("Will be removed in v3.0, please use the " + nameof(StartDebugSessionAsync) + " overload with cancellation token support", DiagnosticId = ObsoleteDefaults.DiagnosticId)] + public static Task StartDebugSessionAsync( DataFactoryResource resource, ILogger logger, Action configureOptions) { + return StartDebugSessionAsync(resource, logger, configureOptions, CancellationToken.None); + } + + /// + /// Starts a new active Azure Data Factory data flow debug session for the given . + /// + /// + /// The resource to start the active data flow debug session for. + /// The resource should be retrieved via the : + /// + /// + /// var credential = new DefaultAzureCredential(); + /// var arm = new ArmClient(credential); + /// DataFactoryResource resource = arm.GetDataFactoryResource(dataFactoryResourceId); + /// + /// + /// + /// The logger to write diagnostic messages during the debug session. + /// The function to configure the options of the debug session. + /// The optional token to propagate notifications that the operation should be cancelled. + /// Thrown when the is null. + /// Thrown when the starting of the data flow debug session did not result in a session ID. + public static async Task StartDebugSessionAsync( + DataFactoryResource resource, + ILogger logger, + Action configureOptions, + CancellationToken cancellationToken) + { + cancellationToken.ThrowIfCancellationRequested(); ArgumentNullException.ThrowIfNull(resource); logger ??= NullLogger.Instance; var options = new TemporaryDataFlowDebugSessionOptions(); configureOptions?.Invoke(options); - DataFlowDebugSessionInfo activeSession = await GetActiveDebugSessionOrDefaultAsync(resource, options.ActiveSessionId).ConfigureAwait(false); + DataFlowDebugSessionInfo activeSession = await GetActiveDebugSessionOrDefaultAsync(resource, options.ActiveSessionId, cancellationToken).ConfigureAwait(false); if (activeSession is not null) { logger.LogSetupReusingSession(resource.Id.Name, activeSession.SessionId); @@ -213,7 +335,7 @@ public static async Task StartDebugSessionAsync( logger.LogSetupStartingSession(resource.Id.Name); ArmOperation result = - await resource.CreateDataFlowDebugSessionAsync(WaitUntil.Completed, new DataFactoryDataFlowDebugSessionContent { TimeToLiveInMinutes = options.TimeToLiveInMinutes }) + await resource.CreateDataFlowDebugSessionAsync(WaitUntil.Completed, new DataFactoryDataFlowDebugSessionContent { TimeToLiveInMinutes = options.TimeToLiveInMinutes }, cancellationToken) .ConfigureAwait(false); Guid sessionId = result.Value.SessionId ?? throw new InvalidOperationException($"[Test:Setup] Starting Data Factory '{resource.Id.Name}' data flow debug session did not result in a session ID"); @@ -222,14 +344,15 @@ public static async Task StartDebugSessionAsync( return new TemporaryDataFlowDebugSession(startedByUs: true, sessionId, resource, logger); } - private static async Task GetActiveDebugSessionOrDefaultAsync(DataFactoryResource resource, Guid existingSessionId) + private static async Task GetActiveDebugSessionOrDefaultAsync(DataFactoryResource resource, Guid existingSessionId, CancellationToken cancellationToken) { + cancellationToken.ThrowIfCancellationRequested(); if (existingSessionId == Guid.Empty) { return null; } - await foreach (DataFlowDebugSessionInfo session in resource.GetDataFlowDebugSessionsAsync().ConfigureAwait(false)) + await foreach (DataFlowDebugSessionInfo session in resource.GetDataFlowDebugSessionsAsync(cancellationToken).ConfigureAwait(false)) { if (existingSessionId == session.SessionId) { @@ -251,11 +374,29 @@ private static async Task GetActiveDebugSessionOrDefau /// Thrown when the or is blank. /// Thrown when the data flow execution did not result in a successful status. /// Thrown when one or more interactions with the Azure DataFactory resource failed. + [Obsolete("Will be removed in v3.0, please use the " + nameof(RunDataFlowAsync) + " overload with cancellation token support", DiagnosticId = ObsoleteDefaults.DiagnosticId)] public Task RunDataFlowAsync(string dataFlowName, string targetSinkName) { return RunDataFlowAsync(dataFlowName, targetSinkName, configureOptions: null); } + /// + /// Starts a given data flow within the debug session, + /// which should give a result in the . + /// + /// The name of the data flow to start. + /// The name of the target sink to get the result from. + /// The optional token to propagate notifications that the operation should be cancelled. + /// The final result of the data flow run. + /// Thrown when the test fixture was already teared down. + /// Thrown when the or is blank. + /// Thrown when the data flow execution did not result in a successful status. + /// Thrown when one or more interactions with the Azure DataFactory resource failed. + public Task RunDataFlowAsync(string dataFlowName, string targetSinkName, CancellationToken cancellationToken) + { + return RunDataFlowAsync(dataFlowName, targetSinkName, configureOptions: null, cancellationToken); + } + /// /// Starts a given data flow within the debug session, /// which should give a result in the . @@ -268,11 +409,35 @@ public Task RunDataFlowAsync(string dataFlowName, string targ /// Thrown when the or is blank. /// Thrown when the data flow execution did not result in a successful status. /// Thrown when one or more interactions with the Azure DataFactory resource failed. - public async Task RunDataFlowAsync( + [Obsolete("Will be removed in v3.0, please use the " + nameof(RunDataFlowAsync) + " overload with cancellation token support", DiagnosticId = ObsoleteDefaults.DiagnosticId)] + public Task RunDataFlowAsync( string dataFlowName, string targetSinkName, Action configureOptions) { + return RunDataFlowAsync(dataFlowName, targetSinkName, configureOptions, CancellationToken.None); + } + + /// + /// Starts a given data flow within the debug session, + /// which should give a result in the . + /// + /// The name of the data flow to start. + /// The name of the target sink to get the result from. + /// The function to configure the options of the data flow run. + /// The optional token to propagate notifications that the operation should be cancelled. + /// The final result of the data flow run. + /// Thrown when the test fixture was already teared down. + /// Thrown when the or is blank. + /// Thrown when the data flow execution did not result in a successful status. + /// Thrown when one or more interactions with the Azure DataFactory resource failed. + public async Task RunDataFlowAsync( + string dataFlowName, + string targetSinkName, + Action configureOptions, + CancellationToken cancellationToken) + { + cancellationToken.ThrowIfCancellationRequested(); ObjectDisposedException.ThrowIf(_isDisposed, this); ArgumentException.ThrowIfNullOrWhiteSpace(dataFlowName); ArgumentException.ThrowIfNullOrWhiteSpace(targetSinkName); @@ -280,33 +445,34 @@ public async Task RunDataFlowAsync( var options = new RunDataFlowOptions(); configureOptions?.Invoke(options); - await StartDataFlowAsync(dataFlowName, options).ConfigureAwait(false); + await StartDataFlowAsync(dataFlowName, options, cancellationToken).ConfigureAwait(false); - return await GetDataFlowResultAsync(dataFlowName, targetSinkName, options).ConfigureAwait(false); + return await GetDataFlowResultAsync(dataFlowName, targetSinkName, options, cancellationToken).ConfigureAwait(false); } - private async Task StartDataFlowAsync(string dataFlowName, RunDataFlowOptions options) + private async Task StartDataFlowAsync(string dataFlowName, RunDataFlowOptions options, CancellationToken cancellationToken) { + cancellationToken.ThrowIfCancellationRequested(); _logger.LogSetupAddDataFlowToSession(dataFlowName, DataFactory.Id.Name); - DataFactoryDataFlowResource dataFlow = await DataFactory.GetDataFactoryDataFlowAsync(dataFlowName).ConfigureAwait(false); + DataFactoryDataFlowResource dataFlow = await DataFactory.GetDataFactoryDataFlowAsync(dataFlowName, cancellationToken: cancellationToken).ConfigureAwait(false); var debug = new DataFactoryDataFlowDebugPackageContent { - DataFlow = new DataFactoryDataFlowDebugInfo(dataFlow.Data.Properties) { Name = dataFlowName }, + DataFlow = new(dataFlow.Data.Properties) { Name = dataFlowName }, DebugSettings = CreateDebugSettings(options), SessionId = SessionId }; foreach (string serviceName in options.LinkedServiceNames) { - await AddLinkedServiceAsync(debug, DataFactory, serviceName).ConfigureAwait(false); + await AddLinkedServiceAsync(debug, DataFactory, serviceName, cancellationToken).ConfigureAwait(false); } - await AddDebugVariantsOfDataFlowSourcesAsync(debug, DataFactory, dataFlow).ConfigureAwait(false); - await AddDebugVariantsOfDataFlowSinksAsync(debug, DataFactory, dataFlow).ConfigureAwait(false); - await AddDebugVariantsOfFlowletsAsync(debug, DataFactory, options).ConfigureAwait(false); + await AddDebugVariantsOfDataFlowSourcesAsync(debug, DataFactory, dataFlow, cancellationToken).ConfigureAwait(false); + await AddDebugVariantsOfDataFlowSinksAsync(debug, DataFactory, dataFlow, cancellationToken).ConfigureAwait(false); + await AddDebugVariantsOfFlowletsAsync(debug, DataFactory, options, cancellationToken).ConfigureAwait(false); - await DataFactory.AddDataFlowToDebugSessionAsync(debug).ConfigureAwait(false); + await DataFactory.AddDataFlowToDebugSessionAsync(debug, cancellationToken: cancellationToken).ConfigureAwait(false); } private DataFlowDebugPackageDebugSettings CreateDebugSettings(RunDataFlowOptions options) @@ -336,17 +502,19 @@ private DataFlowDebugPackageDebugSettings CreateDebugSettings(RunDataFlowOptions private async Task AddDebugVariantsOfDataFlowSourcesAsync( DataFactoryDataFlowDebugPackageContent debug, DataFactoryResource dataFactory, - DataFactoryDataFlowResource dataFlow) + DataFactoryDataFlowResource dataFlow, + CancellationToken cancellationToken) { + cancellationToken.ThrowIfCancellationRequested(); if (dataFlow.Data.Properties is DataFactoryMappingDataFlowProperties properties) { foreach (DataFlowSource source in properties.Sources) { - debug.DebugSettings.SourceSettings.Add(new DataFlowSourceSetting { SourceName = source.Name, RowLimit = 100 }); + debug.DebugSettings.SourceSettings.Add(new() { SourceName = source.Name, RowLimit = 100 }); if (source.Dataset != null) { - DataFactoryDatasetResource dataset = await AddDataSetAsync(debug, dataFactory, source.Dataset.ReferenceName).ConfigureAwait(false); - await AddLinkedServiceAsync(debug, dataFactory, dataset.Data.Properties.LinkedServiceName.ReferenceName).ConfigureAwait(false); + DataFactoryDatasetResource dataset = await AddDataSetAsync(debug, dataFactory, source.Dataset.ReferenceName, cancellationToken).ConfigureAwait(false); + await AddLinkedServiceAsync(debug, dataFactory, dataset.Data.Properties.LinkedServiceName.ReferenceName, cancellationToken).ConfigureAwait(false); } } } @@ -355,15 +523,17 @@ private async Task AddDebugVariantsOfDataFlowSourcesAsync( private async Task AddDebugVariantsOfDataFlowSinksAsync( DataFactoryDataFlowDebugPackageContent debug, DataFactoryResource dataFactory, - DataFactoryDataFlowResource dataFlow) + DataFactoryDataFlowResource dataFlow, + CancellationToken cancellationToken) { + cancellationToken.ThrowIfCancellationRequested(); if (dataFlow.Data.Properties is DataFactoryMappingDataFlowProperties properties) { DataFlowSink[] sinks = properties.Sinks.Where(s => s != null).ToArray(); foreach (DataFlowSink sink in sinks) { - DataFactoryDatasetResource dataset = await AddDataSetAsync(debug, dataFactory, sink.Dataset.ReferenceName).ConfigureAwait(false); - await AddLinkedServiceAsync(debug, dataFactory, dataset.Data.Properties.LinkedServiceName.ReferenceName).ConfigureAwait(false); + DataFactoryDatasetResource dataset = await AddDataSetAsync(debug, dataFactory, sink.Dataset.ReferenceName, cancellationToken).ConfigureAwait(false); + await AddLinkedServiceAsync(debug, dataFactory, dataset.Data.Properties.LinkedServiceName.ReferenceName, cancellationToken).ConfigureAwait(false); } } } @@ -371,8 +541,10 @@ private async Task AddDebugVariantsOfDataFlowSinksAsync( private async Task AddDebugVariantsOfFlowletsAsync( DataFactoryDataFlowDebugPackageContent debug, DataFactoryResource dataFactory, - RunDataFlowOptions options) + RunDataFlowOptions options, + CancellationToken cancellationToken) { + cancellationToken.ThrowIfCancellationRequested(); if (options.FlowletNames.Count == 0) { return; @@ -382,36 +554,46 @@ private async Task AddDebugVariantsOfFlowletsAsync( { _logger.LogSetupAddFlowletToSession(flowletName, dataFactory.Id.Name); - DataFactoryDataFlowResource flowlet = (await dataFactory.GetDataFactoryDataFlowAsync(flowletName).ConfigureAwait(false)).Value; - - var dataFactoryFlowletDebugInfo = new DataFactoryDataFlowDebugInfo(flowlet.Data.Properties) - { - Name = flowletName - }; - debug.DataFlows.Add(dataFactoryFlowletDebugInfo); + DataFactoryDataFlowResource flowlet = await dataFactory.GetDataFactoryDataFlowAsync(flowletName, cancellationToken: cancellationToken).ConfigureAwait(false); + debug.DataFlows.Add(new(flowlet.Data.Properties) { Name = flowletName }); } } - private async Task AddDataSetAsync(DataFactoryDataFlowDebugPackageContent debug, DataFactoryResource dataFactory, string datasetName) + private async Task AddDataSetAsync( + DataFactoryDataFlowDebugPackageContent debug, + DataFactoryResource dataFactory, + string datasetName, + CancellationToken cancellationToken) { + cancellationToken.ThrowIfCancellationRequested(); _logger.LogSetupAddDatasetToSession(datasetName, dataFactory.Id.Name); - DataFactoryDatasetResource dataset = await dataFactory.GetDataFactoryDatasetAsync(datasetName).ConfigureAwait(false); - debug.Datasets.Add(new DataFactoryDatasetDebugInfo(dataset.Data.Properties) { Name = dataset.Data.Name }); + DataFactoryDatasetResource dataset = await dataFactory.GetDataFactoryDatasetAsync(datasetName, cancellationToken: cancellationToken).ConfigureAwait(false); + debug.Datasets.Add(new(dataset.Data.Properties) { Name = dataset.Data.Name }); return dataset; } - private async Task AddLinkedServiceAsync(DataFactoryDataFlowDebugPackageContent debug, DataFactoryResource dataFactory, string serviceName) + private async Task AddLinkedServiceAsync( + DataFactoryDataFlowDebugPackageContent debug, + DataFactoryResource dataFactory, + string serviceName, + CancellationToken cancellationToken) { + cancellationToken.ThrowIfCancellationRequested(); _logger.LogSetupAddLinkedServiceToSession(serviceName, dataFactory.Id.Name); - DataFactoryLinkedServiceResource linkedService = await dataFactory.GetDataFactoryLinkedServiceAsync(serviceName).ConfigureAwait(false); - debug.LinkedServices.Add(new DataFactoryLinkedServiceDebugInfo(linkedService.Data.Properties) { Name = linkedService.Data.Name }); + DataFactoryLinkedServiceResource linkedService = await dataFactory.GetDataFactoryLinkedServiceAsync(serviceName, cancellationToken: cancellationToken).ConfigureAwait(false); + debug.LinkedServices.Add(new(linkedService.Data.Properties) { Name = linkedService.Data.Name }); } - private async Task GetDataFlowResultAsync(string dataFlowName, string targetSinkName, RunDataFlowOptions options) + private async Task GetDataFlowResultAsync( + string dataFlowName, + string targetSinkName, + RunDataFlowOptions options, + CancellationToken cancellationToken) { + cancellationToken.ThrowIfCancellationRequested(); _logger.LogRunDataFlow(dataFlowName, targetSinkName); ArmOperation result = @@ -424,7 +606,7 @@ private async Task GetDataFlowResultAsync(string dataFlowName }, SessionId = SessionId - }).ConfigureAwait(false); + }, cancellationToken).ConfigureAwait(false); if (result.Value.Status != "Succeeded") { @@ -450,7 +632,7 @@ public async ValueTask DisposeAsync() if (_startedByUs) { _logger.LogTeardownStopSession(DataFactory.Id.Name, _sessionId); - await DataFactory.DeleteDataFlowDebugSessionAsync(new DeleteDataFlowDebugSessionContent { SessionId = _sessionId }).ConfigureAwait(false); + await DataFactory.DeleteDataFlowDebugSessionAsync(new DeleteDataFlowDebugSessionContent { SessionId = _sessionId }, CancellationToken.None).ConfigureAwait(false); } _isDisposed = true; diff --git a/src/Arcus.Testing.Tests.Integration/Integration/DataFactory/Fixture/TemporaryDataFactoryDataFlow.cs b/src/Arcus.Testing.Tests.Integration/Integration/DataFactory/Fixture/TemporaryDataFactoryDataFlow.cs index c730af16..1683ba04 100644 --- a/src/Arcus.Testing.Tests.Integration/Integration/DataFactory/Fixture/TemporaryDataFactoryDataFlow.cs +++ b/src/Arcus.Testing.Tests.Integration/Integration/DataFactory/Fixture/TemporaryDataFactoryDataFlow.cs @@ -2,6 +2,7 @@ using System.Collections.Generic; using System.IO; using System.Linq; +using System.Threading; using System.Threading.Tasks; using Arcus.Testing.Tests.Integration.Configuration; using Arcus.Testing.Tests.Integration.Storage.Configuration; @@ -13,6 +14,7 @@ using Azure.ResourceManager.DataFactory; using Azure.ResourceManager.DataFactory.Models; using Microsoft.Extensions.Logging; +using Xunit; namespace Arcus.Testing.Tests.Integration.Integration.DataFactory.Fixture { @@ -35,6 +37,7 @@ public class TemporaryDataFactoryDataFlow : IAsyncDisposable private readonly TestConfig _config; private readonly ArmClient _arm; private readonly DataFlowDataType _dataType; + private readonly CancellationToken _cancellationToken; private readonly ILogger _logger; private TemporaryBlobContainer _sourceContainer; @@ -43,7 +46,7 @@ public class TemporaryDataFactoryDataFlow : IAsyncDisposable private List _flowletNames = new(); private DataFactoryDataFlowResource _dataFlow; - private TemporaryDataFactoryDataFlow(DataFlowDataType dataType, TestConfig config, ILogger logger) + private TemporaryDataFactoryDataFlow(DataFlowDataType dataType, TestConfig config, ILogger logger, CancellationToken cancellationToken) { _dataType = dataType; _linkedServiceName = RandomizeWith("storage"); @@ -51,6 +54,7 @@ private TemporaryDataFactoryDataFlow(DataFlowDataType dataType, TestConfig confi _arm = new ArmClient(new DefaultAzureCredential()); _config = config; _logger = logger; + _cancellationToken = cancellationToken; var env = config.GetAzureEnvironment(); SubscriptionId = env.SubscriptionId; @@ -106,7 +110,7 @@ public static async Task CreateWithCsvSinkSourceAs var dfOptions = new TempDataFlowOptions(); tempDataFlowOptions?.Invoke(dfOptions); - var temp = new TemporaryDataFactoryDataFlow(DataFlowDataType.Csv, config, logger); + var temp = new TemporaryDataFactoryDataFlow(DataFlowDataType.Csv, config, logger, TestContext.Current.CancellationToken); try { await temp.AddSourceBlobContainerAsync(); @@ -130,7 +134,7 @@ public static async Task CreateWithCsvSinkSourceAs /// public static async Task CreateWithJsonSinkSourceAsync(JsonDocForm docForm, TestConfig config, ILogger logger) { - var temp = new TemporaryDataFactoryDataFlow(DataFlowDataType.Json, config, logger); + var temp = new TemporaryDataFactoryDataFlow(DataFlowDataType.Json, config, logger, TestContext.Current.CancellationToken); try { await temp.AddSourceBlobContainerAsync(); @@ -160,6 +164,7 @@ private static string RandomizeWith(string label) private async Task AddLinkedServiceAsync() { + _cancellationToken.ThrowIfCancellationRequested(); _logger.LogTrace("Adding Azure Blob storage linked service '{LinkedServiceName}' to Azure DataFactory '{DataFactoryName}'", _linkedServiceName, DataFactory.Name); ResourceIdentifier resourceId = DataFactoryLinkedServiceResource.CreateResourceIdentifier(SubscriptionId, ResourceGroupName, DataFactory.Name, _linkedServiceName); @@ -169,7 +174,8 @@ private async Task AddLinkedServiceAsync() { AuthenticationType = AzureStorageAuthenticationType.AccountKey, ConnectionString = StorageAccount.ConnectionString, - })); + + }), cancellationToken: _cancellationToken); } private async Task AddCsvSourceAsync(AssertCsvOptions options, TempDataFlowOptions dataFlowOptions) @@ -195,7 +201,7 @@ private async Task AddCsvSourceAsync(AssertCsvOptions options, TempDataFlowOptio dataFlowOptions?.Source.ApplyOptions(sourceProperties, SourceDataSetName); - await _sourceDataset.UpdateAsync(WaitUntil.Completed, new DataFactoryDatasetData(sourceProperties)); + await _sourceDataset.UpdateAsync(WaitUntil.Completed, new DataFactoryDatasetData(sourceProperties), cancellationToken: _cancellationToken); } private async Task AddJsonSourceAsync() @@ -214,7 +220,7 @@ private async Task AddJsonSourceAsync() FolderPath = SourceDataSetName } }; - await _sourceDataset.UpdateAsync(WaitUntil.Completed, new DataFactoryDatasetData(sourceProperties)); + await _sourceDataset.UpdateAsync(WaitUntil.Completed, new DataFactoryDatasetData(sourceProperties), cancellationToken: _cancellationToken); } private async Task AddCsvSinkAsync(AssertCsvOptions options, TempDataFlowOptions dataFlowOptions) @@ -240,7 +246,7 @@ private async Task AddCsvSinkAsync(AssertCsvOptions options, TempDataFlowOptions dataFlowOptions?.Sink.ApplyOptions(sinkProperties, SinkDataSetName); - await _sinkDataset.UpdateAsync(WaitUntil.Completed, new DataFactoryDatasetData(sinkProperties)); + await _sinkDataset.UpdateAsync(WaitUntil.Completed, new DataFactoryDatasetData(sinkProperties), cancellationToken: _cancellationToken); } private async Task AddJsonSinkAsync() @@ -259,17 +265,12 @@ private async Task AddJsonSinkAsync() FolderPath = SinkDataSetName } }; - await _sinkDataset.UpdateAsync(WaitUntil.Completed, new DataFactoryDatasetData(sinkProperties)); + await _sinkDataset.UpdateAsync(WaitUntil.Completed, new DataFactoryDatasetData(sinkProperties), cancellationToken: _cancellationToken); } private async Task AddFlowletsAsync(TempDataFlowOptions dataFlowOptions) { - if (dataFlowOptions?.FlowletNames.Count == 0) - { - return; - } - - foreach (var flowletName in dataFlowOptions?.FlowletNames) + foreach (var flowletName in dataFlowOptions.FlowletNames) { _logger.LogTrace("Adding Flowlet '{FlowletName}' to DataFlow '{DataFlowName}' within Azure DataFactory '{DataFactoryName}'", flowletName, Name, DataFactory.Name); _flowletNames.Add(flowletName); @@ -296,7 +297,7 @@ private async Task AddFlowletsAsync(TempDataFlowOptions dataFlowOptions) properties.ScriptLines.Add(item); } - await flowlet.UpdateAsync(WaitUntil.Completed, new DataFactoryDataFlowData(properties)); + await flowlet.UpdateAsync(WaitUntil.Completed, new DataFactoryDataFlowData(properties), cancellationToken: _cancellationToken); } } @@ -355,7 +356,7 @@ private async Task AddDataFlowAsync(JsonDocForm docForm = JsonDocForm.SingleDoc, properties.ScriptLines.Add(line); } - await _dataFlow.UpdateAsync(WaitUntil.Completed, new DataFactoryDataFlowData(properties)); + await _dataFlow.UpdateAsync(WaitUntil.Completed, new DataFactoryDataFlowData(properties), cancellationToken: _cancellationToken); } private static IEnumerable DataFlowParametersScriptLines(IDictionary dataFlowParameters) @@ -509,7 +510,7 @@ public async ValueTask DisposeAsync() disposables.Add(AsyncDisposable.Create(async () => { _logger.LogTrace("Deleting DataFlow '{DataFlowName}' from Azure DataFactory '{DataFactoryName}'", Name, DataFactory.Name); - await _dataFlow.DeleteAsync(WaitUntil.Completed); + await _dataFlow.DeleteAsync(WaitUntil.Completed, cancellationToken: CancellationToken.None); })); } @@ -518,7 +519,7 @@ public async ValueTask DisposeAsync() disposables.Add(AsyncDisposable.Create(async () => { _logger.LogTrace("Deleting CSV source '{SourceName}' from Azure DataFactory '{DataFactoryName}'", SourceDataSetName, DataFactory.Name); - await _sourceDataset.DeleteAsync(WaitUntil.Completed); + await _sourceDataset.DeleteAsync(WaitUntil.Completed, cancellationToken: CancellationToken.None); })); } @@ -527,7 +528,7 @@ public async ValueTask DisposeAsync() disposables.Add(AsyncDisposable.Create(async () => { _logger.LogTrace("Deleting CSV sink '{SinkName}' from Azure DataFactory '{DataFactoryName}'", SinkDataSetName, DataFactory.Name); - await _sinkDataset.DeleteAsync(WaitUntil.Completed); + await _sinkDataset.DeleteAsync(WaitUntil.Completed, cancellationToken: CancellationToken.None); })); } @@ -538,9 +539,9 @@ public async ValueTask DisposeAsync() foreach (string flowletName in _flowletNames) { DataFactoryDataFlowResource flowlet = GetFlowlet(SubscriptionId, ResourceGroupName, DataFactory.Name, _arm, flowletName); - var flowletResource = await flowlet.GetAsync(); + var flowletResource = await flowlet.GetAsync(cancellationToken: CancellationToken.None); _logger.LogTrace("Deleting flowlet '{FlowletName}' from Azure DataFactory '{DataFactoryName}'", flowletResource.Value.Data.Name, DataFactory.Name); - await flowlet.DeleteAsync(WaitUntil.Completed); + await flowlet.DeleteAsync(WaitUntil.Completed, cancellationToken: CancellationToken.None); } })); } @@ -550,7 +551,7 @@ public async ValueTask DisposeAsync() disposables.Add(AsyncDisposable.Create(async () => { _logger.LogTrace("Deleting Azure Blob storage linked service '{LinkedServiceName}' from Azure DataFactory '{DataFactoryName}'", _linkedServiceName, DataFactory.Name); - await _linkedService.DeleteAsync(WaitUntil.Completed); + await _linkedService.DeleteAsync(WaitUntil.Completed, cancellationToken: CancellationToken.None); })); } } @@ -570,7 +571,7 @@ public class TempDataFlowOptions /// Gets the unique key and values of the parameters of the temporary Data Flow in Azure DataFactory. /// public IDictionary DataFlowParameters { get; } = new Dictionary(); - public List FlowletNames { get; } = new(); + public List FlowletNames { get; } = []; } public class TempDataFlowSourceOptions diff --git a/src/Arcus.Testing.Tests.Integration/Integration/DataFactory/RunDataFlowTests.cs b/src/Arcus.Testing.Tests.Integration/Integration/DataFactory/RunDataFlowTests.cs index 73a788dd..8abdb23f 100644 --- a/src/Arcus.Testing.Tests.Integration/Integration/DataFactory/RunDataFlowTests.cs +++ b/src/Arcus.Testing.Tests.Integration/Integration/DataFactory/RunDataFlowTests.cs @@ -3,6 +3,7 @@ using System.Data; using System.Linq; using System.Text.Json.Nodes; +using System.Threading; using System.Threading.Tasks; using Arcus.Testing.Tests.Core.Assert_.Fixture; using Arcus.Testing.Tests.Core.Integration.DataFactory; @@ -46,7 +47,7 @@ public async Task RunDataFlow_WithJsonFileOnSource_SucceedsByGettingJsonFileOnSi await dataFlow.UploadToSourceAsync(expected!.ToString()); // Act - DataFlowRunResult result = await _session.Value.RunDataFlowAsync(dataFlow.Name, dataFlow.SinkName); + DataFlowRunResult result = await _session.Value.RunDataFlowAsync(dataFlow.Name, dataFlow.SinkName, TestContext.Current.CancellationToken); // Assert AssertJson.Equal(expected, result.GetDataAsJson()); @@ -62,7 +63,7 @@ public async Task RunDataFlow_WithCsvFileOnSource_SucceedsByGettingCsvFileOnSink await dataFlow.UploadToSourceAsync(expectedCsv); // Act - DataFlowRunResult result = await _session.Value.RunDataFlowAsync(dataFlow.Name, dataFlow.SinkName); + DataFlowRunResult result = await _session.Value.RunDataFlowAsync(dataFlow.Name, dataFlow.SinkName, TestContext.Current.CancellationToken); // Assert CsvTable expected = AssertCsv.Load(expectedCsv, ConfigureCsv); @@ -96,12 +97,12 @@ public async Task RunDataFlow_WithCsvFileOnSourceAndDataSetParameter_SucceedsByG DataFlowRunResult result = await _session.Value.RunDataFlowAsync( dataFlow.Name, dataFlow.SinkName, - options => + cancellationToken: TestContext.Current.CancellationToken, + configureOptions: options => { options.AddDataSetParameters(dataFlow.SourceName, sourceDataSetParameterKeyValues); options.AddDataSetParameters(dataFlow.SinkName, sinkDataSetParameterKeyValues); - } - ); + }); // Assert CsvTable expected = AssertCsv.Load(expectedCsv, ConfigureCsv); @@ -145,11 +146,12 @@ public async Task RunDataFlow_WithCsvFileOnSourceAndDataSetParameterAndDataFlowP await dataFlow.UploadToSourceAsync(expectedCsv, sourceDataSetParameterKeyValues.Select(d => d.Value).ToArray()); // Act - DataFlowRunResult result = await _session.Value.RunDataFlowAsync(dataFlow.Name, dataFlow.SinkName, options => - { - options.AddDataFlowParameters(dataFlowParametersWithValues); - options.AddDataSetParameters(dataFlow.SourceName, sourceDataSetParameterKeyValues); - }); + DataFlowRunResult result = await _session.Value.RunDataFlowAsync(dataFlow.Name, dataFlow.SinkName, cancellationToken: TestContext.Current.CancellationToken, + configureOptions: options => + { + options.AddDataFlowParameters(dataFlowParametersWithValues); + options.AddDataSetParameters(dataFlow.SourceName, sourceDataSetParameterKeyValues); + }); // Assert CsvTable expected = AssertCsv.Load(expectedCsv, ConfigureCsv); @@ -197,7 +199,8 @@ public async Task RunDataFlow_WithCsvFileOnSourceAndFlowlet_SucceedsByGettingCsv DataFlowRunResult result = await _session.Value.RunDataFlowAsync(dataFlow.Name, dataFlow.SinkName, options => { options.AddFlowlet(flowletName); - }); + + }, TestContext.Current.CancellationToken); // Assert CsvTable expected = AssertCsv.Load(expectedCsv, ConfigureCsv); @@ -300,12 +303,13 @@ public async ValueTask InitializeAsync() Guid unknownSessionId = Guid.NewGuid(); Value = Bogus.Random.Bool() - ? await TemporaryDataFlowDebugSession.StartDebugSessionAsync(dataFactory.ResourceId, NullLogger.Instance) + ? await TemporaryDataFlowDebugSession.StartDebugSessionAsync(dataFactory.ResourceId, NullLogger.Instance, TestContext.Current.CancellationToken) : await TemporaryDataFlowDebugSession.StartDebugSessionAsync(dataFactory.ResourceId, NullLogger.Instance, opt => { opt.TimeToLiveInMinutes = Bogus.Random.Int(10, 15); opt.ActiveSessionId = unknownSessionId; - }); + + }, TestContext.Current.CancellationToken); Assert.NotEqual(unknownSessionId, Value.SessionId); SessionId = Value.SessionId; @@ -329,7 +333,7 @@ public static async Task IsDebugSessionActiveAsync(ResourceIdentifier reso DataFactoryResource resource = armClient.GetDataFactoryResource(resourceId); var isActive = false; - await foreach (DataFlowDebugSessionInfo session in resource.GetDataFlowDebugSessionsAsync()) + await foreach (DataFlowDebugSessionInfo session in resource.GetDataFlowDebugSessionsAsync(TestContext.Current.CancellationToken)) { if (session.SessionId == sessionId) { @@ -359,7 +363,8 @@ public async ValueTask DisposeAsync() } Assert.Throws(() => Value.SessionId); - await Assert.ThrowsAsync(() => Value.RunDataFlowAsync(Bogus.Lorem.Word(), Bogus.Lorem.Word())); + await Assert.ThrowsAsync( + () => Value.RunDataFlowAsync(Bogus.Lorem.Word(), Bogus.Lorem.Word(), CancellationToken.None)); } } } diff --git a/src/Arcus.Testing.Tests.Integration/Integration/DataFactory/TemporaryDataFlowDebugSessionTests.cs b/src/Arcus.Testing.Tests.Integration/Integration/DataFactory/TemporaryDataFlowDebugSessionTests.cs index 890c7963..2e6d2539 100644 --- a/src/Arcus.Testing.Tests.Integration/Integration/DataFactory/TemporaryDataFlowDebugSessionTests.cs +++ b/src/Arcus.Testing.Tests.Integration/Integration/DataFactory/TemporaryDataFlowDebugSessionTests.cs @@ -22,7 +22,7 @@ public TemporaryDataFlowDebugSessionTests(DataFactoryDebugSession fixture, ITest public async Task StartDebugSession_WithActiveSession_SucceedsByReusingSession() { await using (var otherSession = await TemporaryDataFlowDebugSession.StartDebugSessionAsync(DataFactory.ResourceId, Logger, - options => options.ActiveSessionId = _fixture.Value.SessionId)) + options => options.ActiveSessionId = _fixture.Value.SessionId, TestContext.Current.CancellationToken)) { // Assert Assert.Equal(_fixture.Value.SessionId, otherSession.SessionId);