diff --git a/sdk/storage/Azure.Storage.DataMovement.Blobs/src/BlobStorageResourceContainer.cs b/sdk/storage/Azure.Storage.DataMovement.Blobs/src/BlobStorageResourceContainer.cs
index 89bdc27d0216b..b06694e9f84cb 100644
--- a/sdk/storage/Azure.Storage.DataMovement.Blobs/src/BlobStorageResourceContainer.cs
+++ b/sdk/storage/Azure.Storage.DataMovement.Blobs/src/BlobStorageResourceContainer.cs
@@ -150,5 +150,19 @@ private string ApplyOptionalPrefix(string path)
             => IsDirectory
                 ? string.Join("/", DirectoryPrefix, path)
                 : path;
+
+        // We will require containers to be created before the transfer starts
+        // Since blobs is a flat namespace, we do not need to create directories (as they are virtual).
+        protected override Task CreateIfNotExistsAsync(CancellationToken cancellationToken)
+            => Task.CompletedTask;
+
+        protected override StorageResourceContainer GetChildStorageResourceContainer(string path)
+        {
+            BlobStorageResourceContainerOptions options = _options.DeepCopy();
+            options.BlobDirectoryPrefix = string.Join("/", DirectoryPrefix, path);
+            return new BlobStorageResourceContainer(
+                BlobContainerClient,
+                options);
+        }
     }
 }
diff --git a/sdk/storage/Azure.Storage.DataMovement.Blobs/src/DataMovementBlobsExtensions.cs b/sdk/storage/Azure.Storage.DataMovement.Blobs/src/DataMovementBlobsExtensions.cs
index 52243239fb1d3..0debaf844ff99 100644
--- a/sdk/storage/Azure.Storage.DataMovement.Blobs/src/DataMovementBlobsExtensions.cs
+++ b/sdk/storage/Azure.Storage.DataMovement.Blobs/src/DataMovementBlobsExtensions.cs
@@ -471,5 +471,23 @@ internal static BlobStorageResourceContainerOptions GetBlobContainerOptions(
                 BlobOptions = baseOptions,
             };
         }
+
+        internal static BlobStorageResourceContainerOptions DeepCopy(this BlobStorageResourceContainerOptions options)
+            => new BlobStorageResourceContainerOptions()
+            {
+                BlobType = options?.BlobType ?? BlobType.Block,
+                BlobDirectoryPrefix = options?.BlobDirectoryPrefix,
+                BlobOptions = new BlobStorageResourceOptions()
+                {
+                    Metadata = options?.BlobOptions?.Metadata,
+                    Tags = options?.BlobOptions?.Tags,
+                    HttpHeaders = options?.BlobOptions?.HttpHeaders,
+                    AccessTier = options?.BlobOptions?.AccessTier,
+                    DestinationImmutabilityPolicy = options?.BlobOptions?.DestinationImmutabilityPolicy,
+                    LegalHold = options?.BlobOptions?.LegalHold,
+                    UploadTransferValidationOptions = options?.BlobOptions?.UploadTransferValidationOptions,
+                    DownloadTransferValidationOptions = options?.BlobOptions?.DownloadTransferValidationOptions,
+                }
+            };
     }
 }
diff --git a/sdk/storage/Azure.Storage.DataMovement.Files.Shares/src/DataMovementSharesExtensions.cs b/sdk/storage/Azure.Storage.DataMovement.Files.Shares/src/DataMovementSharesExtensions.cs
index eca0c2ed3d99b..58cf1f38ef18a 100644
--- a/sdk/storage/Azure.Storage.DataMovement.Files.Shares/src/DataMovementSharesExtensions.cs
+++ b/sdk/storage/Azure.Storage.DataMovement.Files.Shares/src/DataMovementSharesExtensions.cs
@@ -25,10 +25,12 @@ internal static ShareFileUploadRangeOptions ToShareFileUploadRangeOptions(
             };
 
         internal static ShareFileUploadRangeFromUriOptions ToShareFileUploadRangeFromUriOptions(
-            this ShareFileStorageResourceOptions options)
+            this ShareFileStorageResourceOptions options,
+            HttpAuthorization sourceAuthorization)
             => new()
             {
                 Conditions = options?.DestinationConditions,
+                SourceAuthentication = sourceAuthorization
             };
 
         internal static StorageResourceProperties ToStorageResourceProperties(
diff --git a/sdk/storage/Azure.Storage.DataMovement.Files.Shares/src/ShareDirectoryStorageResourceContainer.cs b/sdk/storage/Azure.Storage.DataMovement.Files.Shares/src/ShareDirectoryStorageResourceContainer.cs
index 7999acfcd0078..c1449fc230ef3 100644
--- a/sdk/storage/Azure.Storage.DataMovement.Files.Shares/src/ShareDirectoryStorageResourceContainer.cs
+++ b/sdk/storage/Azure.Storage.DataMovement.Files.Shares/src/ShareDirectoryStorageResourceContainer.cs
@@ -43,10 +43,17 @@ protected override StorageResourceItem GetStorageResourceReference(string path)
         protected override async IAsyncEnumerable<StorageResource> GetStorageResourcesAsync(
             [EnumeratorCancellation] CancellationToken cancellationToken = default)
         {
-            await foreach (ShareFileClient client in PathScanner.ScanFilesAsync(
+            await foreach ((ShareDirectoryClient dir, ShareFileClient file) in PathScanner.ScanAsync(
                 ShareDirectoryClient, cancellationToken).ConfigureAwait(false))
             {
-                yield return new ShareFileStorageResource(client, ResourceOptions);
+                if (file != default)
+                {
+                    yield return new ShareFileStorageResource(file, ResourceOptions);
+                }
+                else
+                {
+                    yield return new ShareDirectoryStorageResourceContainer(dir, ResourceOptions);
+                }
             }
         }
 
@@ -59,5 +66,17 @@ protected override StorageResourceCheckpointData GetDestinationCheckpointData()
         {
             return new ShareFileDestinationCheckpointData(null, null);
         }
+
+        protected override async Task CreateIfNotExistsAsync(CancellationToken cancellationToken = default)
+        {
+            await ShareDirectoryClient.CreateIfNotExistsAsync(
+                metadata: default,
+                smbProperties: default,
+                filePermission: default,
+                cancellationToken: cancellationToken).ConfigureAwait(false);
+        }
+
+        protected override StorageResourceContainer GetChildStorageResourceContainer(string path)
+            => new ShareDirectoryStorageResourceContainer(ShareDirectoryClient.GetSubdirectoryClient(path), ResourceOptions);
     }
 }
diff --git a/sdk/storage/Azure.Storage.DataMovement.Files.Shares/src/ShareFileStorageResource.cs b/sdk/storage/Azure.Storage.DataMovement.Files.Shares/src/ShareFileStorageResource.cs
index 8ff96c57122e0..06bd423821db2 100644
--- a/sdk/storage/Azure.Storage.DataMovement.Files.Shares/src/ShareFileStorageResource.cs
+++ b/sdk/storage/Azure.Storage.DataMovement.Files.Shares/src/ShareFileStorageResource.cs
@@ -116,7 +116,7 @@ await ShareFileClient.UploadRangeFromUriAsync(
                 sourceUri: sourceResource.Uri,
                 range: range,
                 sourceRange: range,
-                options: _options.ToShareFileUploadRangeFromUriOptions(),
+                options: _options.ToShareFileUploadRangeFromUriOptions(options?.SourceAuthentication),
                 cancellationToken: cancellationToken).ConfigureAwait(false);
         }
 
@@ -165,7 +165,7 @@ await ShareFileClient.UploadRangeFromUriAsync(
                     sourceUri: sourceResource.Uri,
                     range: new HttpRange(0, completeLength),
                     sourceRange: new HttpRange(0, completeLength),
-                    options: _options.ToShareFileUploadRangeFromUriOptions(),
+                    options: _options.ToShareFileUploadRangeFromUriOptions(options?.SourceAuthentication),
                     cancellationToken: cancellationToken).ConfigureAwait(false);
             }
         }
diff --git a/sdk/storage/Azure.Storage.DataMovement.Files.Shares/tests/Azure.Storage.DataMovement.Files.Shares.Tests.csproj b/sdk/storage/Azure.Storage.DataMovement.Files.Shares/tests/Azure.Storage.DataMovement.Files.Shares.Tests.csproj
index 8173b54d74520..c61c0794bb7be 100644
--- a/sdk/storage/Azure.Storage.DataMovement.Files.Shares/tests/Azure.Storage.DataMovement.Files.Shares.Tests.csproj
+++ b/sdk/storage/Azure.Storage.DataMovement.Files.Shares/tests/Azure.Storage.DataMovement.Files.Shares.Tests.csproj
@@ -41,6 +41,7 @@
     <Compile Include="$(AzureStorageDataMovementTestSharedSources)TestEventsRaised.cs" LinkBase="Shared\DataMovement" />
     <Compile Include="$(AzureStorageDataMovementTestSharedSources)DisposingLocalDirectory.cs" LinkBase="Shared\DataMovement" />
     <Compile Include="$(AzureStorageDataMovementTestSharedSources)StartTransferUploadTestBase.cs" LinkBase="Shared\DataMovement" />
+    <Compile Include="$(AzureStorageDataMovementTestSharedSources)StartTransferDirectoryCopyTestBase.cs" LinkBase="Shared\DataMovement" />
     <Compile Include="$(AzureStorageDataMovementTestSharedSources)StartTransferCopyTestBase.cs" LinkBase="Shared\DataMovement" />
     <Compile Include="$(AzureStorageDataMovementTestSharedSources)StartTransferDownloadTestBase.cs" LinkBase="Shared\DataMovement" />
   </ItemGroup>
diff --git a/sdk/storage/Azure.Storage.DataMovement.Files.Shares/tests/ClientBuilderExtensions.cs b/sdk/storage/Azure.Storage.DataMovement.Files.Shares/tests/ClientBuilderExtensions.cs
index 973d321c3e8ee..0bd6ecb2ee86d 100644
--- a/sdk/storage/Azure.Storage.DataMovement.Files.Shares/tests/ClientBuilderExtensions.cs
+++ b/sdk/storage/Azure.Storage.DataMovement.Files.Shares/tests/ClientBuilderExtensions.cs
@@ -10,7 +10,8 @@
 using SharesClientBuilder = Azure.Storage.Test.Shared.ClientBuilder<
     Azure.Storage.Files.Shares.ShareServiceClient,
     Azure.Storage.Files.Shares.ShareClientOptions>;
-using Azure.Storage.Files.Shares.Models;
+using System.Threading;
+using Azure.Core;
 
 namespace Azure.Storage.DataMovement.Files.Shares.Tests
 {
@@ -43,24 +44,15 @@ public static SharesClientBuilder GetNewShareClientBuilder(TenantConfigurationBu
                 (uri, azureSasCredential, clientOptions) => new ShareServiceClient(uri, azureSasCredential, clientOptions),
                 () => new ShareClientOptions(serviceVersion));
 
-        public static ShareServiceClient GetServiceClient_OAuthAccount_SharedKey(this SharesClientBuilder clientBuilder) =>
-            clientBuilder.GetServiceClientFromSharedKeyConfig(clientBuilder.Tenants.TestConfigOAuth);
-
-        public static ShareServiceClient GetServiceClient_OAuth(
-            this SharesClientBuilder clientBuilder, ShareClientOptions options = default)
-        {
-            options ??= clientBuilder.GetOptions();
-            options.ShareTokenIntent = ShareTokenIntent.Backup;
-            return clientBuilder.GetServiceClientFromOauthConfig(clientBuilder.Tenants.TestConfigOAuth, options);
-        }
-
         public static async Task<DisposingShare> GetTestShareAsync(
             this SharesClientBuilder clientBuilder,
             ShareServiceClient service = default,
             string shareName = default,
             IDictionary<string, string> metadata = default,
-            ShareClientOptions options = default)
+            ShareClientOptions options = default,
+            CancellationToken cancellationToken = default)
         {
+            CancellationHelper.ThrowIfCancellationRequested(cancellationToken);
             service ??= clientBuilder.GetServiceClientFromSharedKeyConfig(clientBuilder.Tenants.TestConfigDefault, options);
             metadata ??= new Dictionary<string, string>(StringComparer.OrdinalIgnoreCase);
             shareName ??= clientBuilder.GetNewShareName();
diff --git a/sdk/storage/Azure.Storage.DataMovement.Files.Shares/tests/ShareDirectoryStartTransferCopyTests.cs b/sdk/storage/Azure.Storage.DataMovement.Files.Shares/tests/ShareDirectoryStartTransferCopyTests.cs
new file mode 100644
index 0000000000000..794681a64f2c3
--- /dev/null
+++ b/sdk/storage/Azure.Storage.DataMovement.Files.Shares/tests/ShareDirectoryStartTransferCopyTests.cs
@@ -0,0 +1,204 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
+
+using System;
+using System.Collections.Generic;
+using System.Drawing;
+using System.IO;
+using System.Linq;
+using System.Text;
+using System.Threading.Tasks;
+using Azure.Core.TestFramework;
+using Azure.Storage.DataMovement.Tests;
+using Azure.Storage.Files.Shares;
+using Azure.Storage.Files.Shares.Models;
+using Azure.Storage.Test.Shared;
+using Azure.Storage.Files.Shares.Tests;
+using NUnit.Framework;
+using System.Security.AccessControl;
+using Microsoft.Extensions.Options;
+using System.Threading;
+using Azure.Core;
+
+namespace Azure.Storage.DataMovement.Files.Shares.Tests
+{
+    [ShareClientTestFixture]
+    public class ShareDirectoryStartTransferCopyTests : StartTransferDirectoryCopyTestBase<
+        ShareServiceClient,
+        ShareClient,
+        ShareClientOptions,
+        ShareServiceClient,
+        ShareClient,
+        ShareClientOptions,
+        StorageTestEnvironment>
+    {
+        private const string _fileResourcePrefix = "test-file-";
+        private const string _expectedOverwriteExceptionMessage = "Cannot overwrite file.";
+
+        public ShareDirectoryStartTransferCopyTests(bool async, ShareClientOptions.ServiceVersion serviceVersion)
+            : base(async, _expectedOverwriteExceptionMessage, _fileResourcePrefix, null /* RecordedTestMode.Record /* to re-record */)
+        {
+            SourceClientBuilder = ClientBuilderExtensions.GetNewShareClientBuilder(Tenants, serviceVersion);
+            DestinationClientBuilder = ClientBuilderExtensions.GetNewShareClientBuilder(Tenants, serviceVersion);
+        }
+
+        protected override async Task CreateObjectInSourceAsync(
+            ShareClient container,
+            long? objectLength = null,
+            string objectName = null,
+            Stream contents = default,
+            CancellationToken cancellationToken = default)
+            => await CreateShareFileAsync(container, objectLength, objectName, contents, cancellationToken);
+
+        protected override async Task CreateObjectInDestinationAsync(
+            ShareClient container,
+            long? objectLength = null,
+            string objectName = null,
+            Stream contents = null,
+            CancellationToken cancellationToken = default)
+            => await CreateShareFileAsync(container, objectLength, objectName, contents, cancellationToken);
+
+        protected override async Task<IDisposingContainer<ShareClient>> GetDestinationDisposingContainerAsync(
+            ShareServiceClient service = null,
+            string containerName = null,
+            CancellationToken cancellationToken = default)
+            => await DestinationClientBuilder.GetTestShareAsync(service, containerName, cancellationToken: cancellationToken);
+
+        protected override StorageResourceContainer GetDestinationStorageResourceContainer(ShareClient containerClient, string prefix)
+            => new ShareDirectoryStorageResourceContainer(containerClient.GetDirectoryClient(prefix), default);
+
+        protected override ShareClient GetOAuthSourceContainerClient(string containerName)
+        {
+            ShareClientOptions options = SourceClientBuilder.GetOptions();
+            options.ShareTokenIntent = ShareTokenIntent.Backup;
+            ShareServiceClient oauthService = SourceClientBuilder.GetServiceClientFromOauthConfig(Tenants.TestConfigOAuth, options);
+            return oauthService.GetShareClient(containerName);
+        }
+
+        protected override ShareClient GetOAuthDestinationContainerClient(string containerName)
+        {
+            ShareClientOptions options = DestinationClientBuilder.GetOptions();
+            options.ShareTokenIntent = ShareTokenIntent.Backup;
+            ShareServiceClient oauthService = DestinationClientBuilder.GetServiceClientFromOauthConfig(Tenants.TestConfigOAuth, options);
+            return oauthService.GetShareClient(containerName);
+        }
+
+        protected override async Task<IDisposingContainer<ShareClient>> GetSourceDisposingContainerAsync(ShareServiceClient service = null, string containerName = null, CancellationToken cancellationToken = default)
+        {
+            service ??= SourceClientBuilder.GetServiceClientFromSharedKeyConfig(SourceClientBuilder.Tenants.TestConfigDefault, SourceClientBuilder.GetOptions());
+            ShareServiceClient sasService = new ShareServiceClient(service.GenerateAccountSasUri(
+                Sas.AccountSasPermissions.All,
+                SourceClientBuilder.Recording.UtcNow.AddDays(1),
+                Sas.AccountSasResourceTypes.All),
+                SourceClientBuilder.GetOptions());
+            return await SourceClientBuilder.GetTestShareAsync(sasService, containerName, cancellationToken: cancellationToken);
+        }
+
+        protected override StorageResourceContainer GetSourceStorageResourceContainer(ShareClient containerClient, string prefix = null)
+            => new ShareDirectoryStorageResourceContainer(containerClient.GetDirectoryClient(prefix), default);
+
+        protected override async Task CreateDirectoryInSourceAsync(ShareClient sourceContainer, string directoryPath, CancellationToken cancellationToken = default)
+            => await CreateDirectoryTreeAsync(sourceContainer, directoryPath, cancellationToken);
+
+        protected override async Task CreateDirectoryInDestinationAsync(ShareClient destinationContainer, string directoryPath, CancellationToken cancellationToken = default)
+            => await CreateDirectoryTreeAsync(destinationContainer, directoryPath, cancellationToken);
+
+        protected override async Task VerifyEmptyDestinationContainerAsync(
+            ShareClient destinationContainer,
+            string destinationPrefix,
+            CancellationToken cancellationToken = default)
+        {
+            CancellationHelper.ThrowIfCancellationRequested(cancellationToken);
+            ShareDirectoryClient destinationDirectory = string.IsNullOrEmpty(destinationPrefix) ?
+                destinationContainer.GetRootDirectoryClient() :
+                destinationContainer.GetDirectoryClient(destinationPrefix);
+            IList<ShareFileItem> items = await destinationDirectory.GetFilesAndDirectoriesAsync(cancellationToken: cancellationToken).ToListAsync();
+            Assert.IsEmpty(items);
+        }
+
+        protected override async Task VerifyResultsAsync(
+            ShareClient sourceContainer,
+            string sourcePrefix,
+            ShareClient destinationContainer,
+            string destinationPrefix,
+            CancellationToken cancellationToken = default)
+        {
+            CancellationHelper.ThrowIfCancellationRequested(cancellationToken);
+
+            // List all files in source blob folder path
+            List<string> sourceFileNames = new List<string>();
+            List<string> sourceDirectoryNames = new List<string>();
+
+            // Get source directory client and list the paths
+            ShareDirectoryClient sourceDirectory = string.IsNullOrEmpty(sourcePrefix) ?
+                sourceContainer.GetRootDirectoryClient() :
+                sourceContainer.GetDirectoryClient(sourcePrefix);
+            await foreach (Page<ShareFileItem> page in sourceDirectory.GetFilesAndDirectoriesAsync().AsPages())
+            {
+                sourceFileNames.AddRange(page.Values.Where((ShareFileItem item) => !item.IsDirectory).Select((ShareFileItem item) => item.Name));
+                sourceDirectoryNames.AddRange(page.Values.Where((ShareFileItem item) => item.IsDirectory).Select((ShareFileItem item) => item.Name));
+            }
+
+            // List all files in the destination blob folder path
+            List<string> destinationFileNames = new List<string>();
+            List<string> destinationDirectoryNames = new List<string>();
+
+            ShareDirectoryClient destinationDirectory = string.IsNullOrEmpty(destinationPrefix) ?
+                destinationContainer.GetRootDirectoryClient() :
+                destinationContainer.GetDirectoryClient(destinationPrefix);
+            await foreach (Page<ShareFileItem> page in destinationDirectory.GetFilesAndDirectoriesAsync().AsPages())
+            {
+                destinationFileNames.AddRange(page.Values.Where((ShareFileItem item) => !item.IsDirectory).Select((ShareFileItem item) => item.Name));
+                destinationDirectoryNames.AddRange(page.Values.Where((ShareFileItem item) => item.IsDirectory).Select((ShareFileItem item) => item.Name));
+            }
+
+            // Assert subdirectories
+            Assert.AreEqual(sourceDirectoryNames.Count, destinationDirectoryNames.Count);
+            Assert.AreEqual(sourceDirectoryNames, destinationDirectoryNames);
+
+            // Assert file and file contents
+            Assert.AreEqual(sourceFileNames.Count, destinationFileNames.Count);
+            for (int i = 0; i < sourceFileNames.Count; i++)
+            {
+                Assert.AreEqual(
+                    sourceFileNames[i],
+                    destinationFileNames[i]);
+
+                // Verify Download
+                string sourceFileName = Path.Combine(sourcePrefix, sourceFileNames[i]);
+                using Stream sourceStream = await sourceDirectory.GetFileClient(sourceFileNames[i]).OpenReadAsync(cancellationToken: cancellationToken);
+                using Stream destinationStream = await destinationDirectory.GetFileClient(destinationFileNames[i]).OpenReadAsync(cancellationToken: cancellationToken);
+                Assert.AreEqual(sourceStream, destinationStream);
+            }
+        }
+
+        private async Task CreateShareFileAsync(
+            ShareClient container,
+            long? objectLength = null,
+            string objectName = null,
+            Stream contents = default,
+            CancellationToken cancellationToken = default)
+        {
+            CancellationHelper.ThrowIfCancellationRequested(cancellationToken);
+            objectName ??= GetNewObjectName();
+            if (!objectLength.HasValue)
+            {
+                throw new InvalidOperationException($"Cannot create share file without size specified. Specify {nameof(objectLength)}.");
+            }
+            ShareFileClient fileClient = container.GetRootDirectoryClient().GetFileClient(objectName);
+            await fileClient.CreateAsync(objectLength.Value);
+
+            if (contents != default)
+            {
+                await fileClient.UploadAsync(contents, cancellationToken: cancellationToken);
+            }
+        }
+
+        private async Task CreateDirectoryTreeAsync(ShareClient container, string directoryPath, CancellationToken cancellationToken = default)
+        {
+            CancellationHelper.ThrowIfCancellationRequested(cancellationToken);
+            ShareDirectoryClient directory = container.GetRootDirectoryClient().GetSubdirectoryClient(directoryPath);
+            await directory.CreateIfNotExistsAsync(cancellationToken: cancellationToken);
+        }
+    }
+}
diff --git a/sdk/storage/Azure.Storage.DataMovement.Files.Shares/tests/ShareDirectoryStorageResourceContainerTests.cs b/sdk/storage/Azure.Storage.DataMovement.Files.Shares/tests/ShareDirectoryStorageResourceContainerTests.cs
index 327b072cd6376..1b06dcd0a35f6 100644
--- a/sdk/storage/Azure.Storage.DataMovement.Files.Shares/tests/ShareDirectoryStorageResourceContainerTests.cs
+++ b/sdk/storage/Azure.Storage.DataMovement.Files.Shares/tests/ShareDirectoryStorageResourceContainerTests.cs
@@ -7,8 +7,12 @@
 using System.Text;
 using System.Threading;
 using System.Threading.Tasks;
+using Azure.Core.TestFramework;
 using Azure.Storage.Files.Shares;
+using Azure.Storage.Files.Shares.Models;
+using Azure.Storage.Test;
 using Azure.Storage.Tests;
+using BenchmarkDotNet.Toolchains.Roslyn;
 using Moq;
 using NUnit.Framework;
 
@@ -16,13 +20,19 @@ namespace Azure.Storage.DataMovement.Files.Shares.Tests
 {
     internal class ShareDirectoryStorageResourceContainerTests
     {
-        private async IAsyncEnumerable<T> ToAsyncEnumerable<T>(IEnumerable<T> items)
+        private async IAsyncEnumerable<(TDirectory Directory, TFile File)> ToAsyncEnumerable<TDirectory, TFile>(
+            IEnumerable<TDirectory> directories,
+            IEnumerable<TFile> files)
         {
-            foreach (var item in items)
+            if (files.Count() != directories.Count())
+            {
+                throw new ArgumentException("Items and Directories should be the same amount");
+            }
+            for (int i = 0; i < files.Count(); i++)
             {
                 // returning async enumerable must be an async method
                 // so we need something to await
-                yield return await Task.FromResult(item);
+                yield return await Task.FromResult((directories.ElementAt(i), files.ElementAt(i)));
             }
         }
 
@@ -31,15 +41,21 @@ public async Task GetStorageResourcesCallsPathScannerCorrectly()
         {
             // Given clients
             Mock<ShareDirectoryClient> mainClient = new();
-            List<Mock<ShareFileClient>> expectedFiles = Enumerable.Range(0, 10)
+            int pathCount = 10;
+            List<Mock<ShareFileClient>> expectedFiles = Enumerable.Range(0, pathCount)
                 .Select(i => new Mock<ShareFileClient>())
                 .ToList();
+            List<Mock<ShareDirectoryClient>> expectedDirectories = Enumerable.Range(0, pathCount)
+                .Select(i => new Mock<ShareDirectoryClient>())
+                .ToList();
 
             // And a mock path scanner
             Mock<PathScanner> pathScanner = new();
-            pathScanner.Setup(p => p.ScanFilesAsync(mainClient.Object, It.IsAny<CancellationToken>()))
+            pathScanner.Setup(p => p.ScanAsync(mainClient.Object, It.IsAny<CancellationToken>()))
                 .Returns<ShareDirectoryClient, CancellationToken>(
-                    (dir, cancellationToken) => ToAsyncEnumerable(expectedFiles.Select(m => m.Object)));
+                (dir, cancellationToken) => ToAsyncEnumerable(
+                    expectedDirectories.Select(m => m.Object),
+                    expectedFiles.Select(m => m.Object)));
 
             // Setup StorageResourceContainer
             ShareDirectoryStorageResourceContainer resource = new(mainClient.Object, default)
@@ -61,7 +77,7 @@ public async Task GetStorageResourcesCallsPathScannerCorrectly()
         public void GetCorrectStorageResourceItem()
         {
             // Given a resource container
-            ShareDirectoryClient startingDir = new(new Uri("https://myaccount.file.core.windows.net/myshare/mydir"));
+            ShareDirectoryClient startingDir = new(new Uri("https://myaccount.file.core.windows.net/myshare/mydir"), new ShareClientOptions());
             ShareDirectoryStorageResourceContainer resourceContainer = new(startingDir, default);
 
             // and a subpath to get
@@ -79,5 +95,86 @@ public void GetCorrectStorageResourceItem()
                 Is.EqualTo(startingDir.Path + "/" + string.Join("/", pathSegments)));
             Assert.That(fileResourceItem.ShareFileClient.Name, Is.EqualTo(pathSegments.Last()));
         }
+
+        [Test]
+        public async Task CreateIfNotExists()
+        {
+            // Arrange
+            Mock<ShareDirectoryClient> mock = new(new Uri("https://myaccount.file.core.windows.net/myshare/mydir"), new ShareClientOptions());
+            mock.Setup(b => b.CreateIfNotExistsAsync(It.IsAny<IDictionary<string, string>>(), It.IsAny<FileSmbProperties>(), It.IsAny<string>(), It.IsAny<CancellationToken>()))
+                .Returns(Task.FromResult(Response.FromValue(
+                    SharesModelFactory.StorageDirectoryInfo(
+                        eTag: new ETag("etag"),
+                        lastModified: DateTimeOffset.UtcNow,
+                        filePermissionKey: default,
+                        fileAttributes: default,
+                        fileCreationTime: DateTimeOffset.MinValue,
+                        fileLastWriteTime: DateTimeOffset.MinValue,
+                        fileChangeTime: DateTimeOffset.MinValue,
+                        fileId: default,
+                        fileParentId: default),
+                    new MockResponse(200))));
+
+            ShareDirectoryStorageResourceContainer resourceContainer = new(mock.Object, default);
+
+            // Act
+            await resourceContainer.CreateIfNotExistsInternalAsync();
+
+            // Assert
+            mock.Verify(b => b.CreateIfNotExistsAsync(It.IsAny<IDictionary<string, string>>(), It.IsAny<FileSmbProperties>(), It.IsAny<string>(), It.IsAny<CancellationToken>()),
+                Times.Once());
+            mock.VerifyNoOtherCalls();
+        }
+
+        [Test]
+        public async Task CreateIfNotExists_Error()
+        {
+            // Arrange
+            Mock<ShareDirectoryClient> mock = new(new Uri("https://myaccount.file.core.windows.net/myshare/mydir"), new ShareClientOptions());
+            mock.Setup(b => b.CreateIfNotExistsAsync(It.IsAny<IDictionary<string, string>>(), It.IsAny<FileSmbProperties>(), It.IsAny<string>(), It.IsAny<CancellationToken>()))
+                .Throws(new RequestFailedException(status: 404, message: "The parent path does not exist.", errorCode: "ResourceNotFound", default));
+
+            ShareDirectoryStorageResourceContainer resourceContainer = new(mock.Object, default);
+
+            // Act
+            await TestHelper.AssertExpectedExceptionAsync<RequestFailedException>(
+                resourceContainer.CreateIfNotExistsInternalAsync(),
+                e =>
+                {
+                    Assert.AreEqual("ResourceNotFound", e.ErrorCode);
+                });
+
+            mock.Verify(b => b.CreateIfNotExistsAsync(It.IsAny<IDictionary<string, string>>(), It.IsAny<FileSmbProperties>(), It.IsAny<string>(), It.IsAny<CancellationToken>()),
+                Times.Once());
+            mock.VerifyNoOtherCalls();
+        }
+
+        [Test]
+        public void GetChildStorageResourceContainer()
+        {
+            // Arrange
+            Uri uri = new Uri("https://storageaccount.file.core.windows.net/container/directory");
+            Mock<ShareDirectoryClient> mock = new Mock<ShareDirectoryClient>(uri, new ShareClientOptions());
+            mock.Setup(b => b.Uri).Returns(uri);
+            mock.Setup(b => b.GetSubdirectoryClient(It.IsAny<string>()))
+                .Returns<string>((path) =>
+                {
+                    UriBuilder builder = new UriBuilder(uri);
+                    builder.Path = string.Join("/", builder.Path, path);
+                    return new ShareDirectoryClient(builder.Uri);
+                });
+
+            ShareDirectoryStorageResourceContainer containerResource =
+                new(mock.Object, new ShareFileStorageResourceOptions());
+
+            // Act
+            string childPath = "foo";
+            StorageResourceContainer childContainer = containerResource.GetChildStorageResourceContainerInternal(childPath);
+
+            // Assert
+            UriBuilder builder = new UriBuilder(containerResource.Uri);
+            builder.Path = string.Join("/", builder.Path, childPath);
+            Assert.AreEqual(builder.Uri, childContainer.Uri);
+        }
     }
 }
diff --git a/sdk/storage/Azure.Storage.DataMovement/api/Azure.Storage.DataMovement.net6.0.cs b/sdk/storage/Azure.Storage.DataMovement/api/Azure.Storage.DataMovement.net6.0.cs
index 4140218d0bdb5..58bdd57ce296a 100644
--- a/sdk/storage/Azure.Storage.DataMovement/api/Azure.Storage.DataMovement.net6.0.cs
+++ b/sdk/storage/Azure.Storage.DataMovement/api/Azure.Storage.DataMovement.net6.0.cs
@@ -129,6 +129,8 @@ public abstract partial class StorageResourceContainer : Azure.Storage.DataMovem
     {
         protected StorageResourceContainer() { }
         protected internal override bool IsContainer { get { throw null; } }
+        protected internal abstract System.Threading.Tasks.Task CreateIfNotExistsAsync(System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken));
+        protected internal abstract Azure.Storage.DataMovement.StorageResourceContainer GetChildStorageResourceContainer(string path);
         protected internal abstract Azure.Storage.DataMovement.StorageResourceItem GetStorageResourceReference(string path);
         protected internal abstract System.Collections.Generic.IAsyncEnumerable<Azure.Storage.DataMovement.StorageResource> GetStorageResourcesAsync(System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken));
     }
diff --git a/sdk/storage/Azure.Storage.DataMovement/api/Azure.Storage.DataMovement.netstandard2.0.cs b/sdk/storage/Azure.Storage.DataMovement/api/Azure.Storage.DataMovement.netstandard2.0.cs
index 4140218d0bdb5..58bdd57ce296a 100644
--- a/sdk/storage/Azure.Storage.DataMovement/api/Azure.Storage.DataMovement.netstandard2.0.cs
+++ b/sdk/storage/Azure.Storage.DataMovement/api/Azure.Storage.DataMovement.netstandard2.0.cs
@@ -129,6 +129,8 @@ public abstract partial class StorageResourceContainer : Azure.Storage.DataMovem
     {
         protected StorageResourceContainer() { }
         protected internal override bool IsContainer { get { throw null; } }
+        protected internal abstract System.Threading.Tasks.Task CreateIfNotExistsAsync(System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken));
+        protected internal abstract Azure.Storage.DataMovement.StorageResourceContainer GetChildStorageResourceContainer(string path);
         protected internal abstract Azure.Storage.DataMovement.StorageResourceItem GetStorageResourceReference(string path);
         protected internal abstract System.Collections.Generic.IAsyncEnumerable<Azure.Storage.DataMovement.StorageResource> GetStorageResourcesAsync(System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken));
     }
diff --git a/sdk/storage/Azure.Storage.DataMovement/src/LocalDirectoryStorageResourceContainer.cs b/sdk/storage/Azure.Storage.DataMovement/src/LocalDirectoryStorageResourceContainer.cs
index 8b0383e1f820d..28c67ffee4a74 100644
--- a/sdk/storage/Azure.Storage.DataMovement/src/LocalDirectoryStorageResourceContainer.cs
+++ b/sdk/storage/Azure.Storage.DataMovement/src/LocalDirectoryStorageResourceContainer.cs
@@ -6,6 +6,7 @@
 using System.IO;
 using System.Runtime.CompilerServices;
 using System.Threading;
+using System.Threading.Tasks;
 using Azure.Core;
 
 namespace Azure.Storage.DataMovement
@@ -72,9 +73,17 @@ protected internal override async IAsyncEnumerable<StorageResource> GetStorageRe
             PathScanner scanner = new PathScanner(_uri.LocalPath);
             foreach (FileSystemInfo fileSystemInfo in scanner.Scan(false))
             {
-                // Skip over directories for now since directory creation is unnecessary.
-                if (!fileSystemInfo.Attributes.HasFlag(FileAttributes.Directory))
+                if (fileSystemInfo.Attributes.HasFlag(FileAttributes.Directory))
                 {
+                    // Directory - but check for the case where it returns the directory you're currently listing
+                    if (fileSystemInfo.FullName != _uri.LocalPath)
+                    {
+                        yield return new LocalDirectoryStorageResourceContainer(fileSystemInfo.FullName);
+                    }
+                }
+                else
+                {
+                    // File
                     yield return new LocalFileStorageResource(fileSystemInfo.FullName);
                 }
             }
@@ -89,5 +98,15 @@ protected internal override StorageResourceCheckpointData GetDestinationCheckpoi
         {
             return new LocalDestinationCheckpointData();
         }
+
+        protected internal override Task CreateIfNotExistsAsync(CancellationToken cancellationToken = default)
+            => Task.CompletedTask;
+
+        protected internal override StorageResourceContainer GetChildStorageResourceContainer(string path)
+        {
+            UriBuilder uri = new UriBuilder(_uri);
+            uri.Path = Path.Combine(uri.Path, path);
+            return new LocalDirectoryStorageResourceContainer(uri.Uri);
+        }
     }
 }
diff --git a/sdk/storage/Azure.Storage.DataMovement/src/ServiceToServiceTransferJob.cs b/sdk/storage/Azure.Storage.DataMovement/src/ServiceToServiceTransferJob.cs
index 5e8926f3fa63b..733a1179880e4 100644
--- a/sdk/storage/Azure.Storage.DataMovement/src/ServiceToServiceTransferJob.cs
+++ b/sdk/storage/Azure.Storage.DataMovement/src/ServiceToServiceTransferJob.cs
@@ -171,31 +171,46 @@ private async IAsyncEnumerable<JobPartInternal> GetStorageResourcesAsync()
                 }
 
                 StorageResource current = enumerator.Current;
-                if (!existingSources.Contains(current.Uri))
+
+                if (current.IsContainer)
                 {
+                    // Create sub-container
                     string containerUriPath = _sourceResourceContainer.Uri.GetPath();
-                    string sourceName = string.IsNullOrEmpty(containerUriPath)
+                    string subContainerPath = string.IsNullOrEmpty(containerUriPath)
                         ? current.Uri.GetPath()
                         : current.Uri.GetPath().Substring(containerUriPath.Length + 1);
-
-                    ServiceToServiceJobPart part;
-                    try
-                    {
-                        part = await ServiceToServiceJobPart.CreateJobPartAsync(
-                            job: this,
-                            partNumber: partNumber,
-                            sourceResource: (StorageResourceItem)current,
-                            destinationResource: _destinationResourceContainer.GetStorageResourceReference(sourceName))
-                            .ConfigureAwait(false);
-                        AppendJobPart(part);
-                    }
-                    catch (Exception ex)
+                    StorageResourceContainer subContainer =
+                        _destinationResourceContainer.GetChildStorageResourceContainer(subContainerPath);
+                    await subContainer.CreateIfNotExistsAsync().ConfigureAwait(false);
+                }
+                else
+                {
+                    if (!existingSources.Contains(current.Uri))
                     {
-                        await InvokeFailedArgAsync(ex).ConfigureAwait(false);
-                        yield break;
+                        string containerUriPath = _sourceResourceContainer.Uri.GetPath();
+                        string sourceName = string.IsNullOrEmpty(containerUriPath)
+                            ? current.Uri.GetPath()
+                            : current.Uri.GetPath().Substring(containerUriPath.Length + 1);
+
+                        ServiceToServiceJobPart part;
+                        try
+                        {
+                            part = await ServiceToServiceJobPart.CreateJobPartAsync(
+                                job: this,
+                                partNumber: partNumber,
+                                sourceResource: (StorageResourceItem)current,
+                                destinationResource: _destinationResourceContainer.GetStorageResourceReference(sourceName))
+                                .ConfigureAwait(false);
+                            AppendJobPart(part);
+                        }
+                        catch (Exception ex)
+                        {
+                            await InvokeFailedArgAsync(ex).ConfigureAwait(false);
+                            yield break;
+                        }
+                        yield return part;
+                        partNumber++;
                     }
-                    yield return part;
-                    partNumber++;
                 }
             }
         }
diff --git a/sdk/storage/Azure.Storage.DataMovement/src/Shared/StorageResourceContainerInternal.cs b/sdk/storage/Azure.Storage.DataMovement/src/Shared/StorageResourceContainerInternal.cs
index 9aeb4e0d5b908..2edd22a507e5a 100644
--- a/sdk/storage/Azure.Storage.DataMovement/src/Shared/StorageResourceContainerInternal.cs
+++ b/sdk/storage/Azure.Storage.DataMovement/src/Shared/StorageResourceContainerInternal.cs
@@ -3,6 +3,7 @@
 
 using System.Collections.Generic;
 using System.Threading;
+using System.Threading.Tasks;
 
 namespace Azure.Storage.DataMovement
 {
@@ -20,5 +21,11 @@ internal IAsyncEnumerable<StorageResource> GetStorageResourcesInternalAsync(
 
         internal StorageResourceItem GetStorageResourceReferenceInternal(string path)
             => GetStorageResourceReference(path);
+
+        internal Task CreateIfNotExistsInternalAsync(CancellationToken cancellationToken = default)
+            => CreateIfNotExistsAsync(cancellationToken);
+
+        internal StorageResourceContainer GetChildStorageResourceContainerInternal(string path)
+            => GetChildStorageResourceContainer(path);
     }
 }
diff --git a/sdk/storage/Azure.Storage.DataMovement/src/StorageResourceContainer.cs b/sdk/storage/Azure.Storage.DataMovement/src/StorageResourceContainer.cs
index 1503623a4d605..886f2b9449f03 100644
--- a/sdk/storage/Azure.Storage.DataMovement/src/StorageResourceContainer.cs
+++ b/sdk/storage/Azure.Storage.DataMovement/src/StorageResourceContainer.cs
@@ -3,6 +3,7 @@
 
 using System.Collections.Generic;
 using System.Threading;
+using System.Threading.Tasks;
 
 namespace Azure.Storage.DataMovement
 {
@@ -28,6 +29,21 @@ protected internal abstract IAsyncEnumerable<StorageResource> GetStorageResource
         /// <param name="path"></param>
         protected internal abstract StorageResourceItem GetStorageResourceReference(string path);
 
+        /// <summary>
+        /// Creates storage resource container if it does not already exists.
+        /// </summary>
+        /// <returns></returns>
+        protected internal abstract Task CreateIfNotExistsAsync(CancellationToken cancellationToken = default);
+
+        /// <summary>
+        /// Gets the child StorageResourceContainer of the respective container.
+        /// </summary>
+        /// <param name="path">
+        /// The path of the child container.
+        /// </param>
+        /// <returns></returns>
+        protected internal abstract StorageResourceContainer GetChildStorageResourceContainer(string path);
+
         /// <summary>
         /// Storage Resource is a container.
         /// </summary>
diff --git a/sdk/storage/Azure.Storage.DataMovement/src/StreamToUriTransferJob.cs b/sdk/storage/Azure.Storage.DataMovement/src/StreamToUriTransferJob.cs
index 2691762d08974..2ca04b94c144b 100644
--- a/sdk/storage/Azure.Storage.DataMovement/src/StreamToUriTransferJob.cs
+++ b/sdk/storage/Azure.Storage.DataMovement/src/StreamToUriTransferJob.cs
@@ -169,31 +169,46 @@ private async IAsyncEnumerable<JobPartInternal> GetStorageResourcesAsync()
                 }
 
                 StorageResource current = enumerator.Current;
-                if (!existingSources.Contains(current.Uri))
+
+                if (current.IsContainer)
                 {
+                    // Create sub-container
                     string containerUriPath = _sourceResourceContainer.Uri.GetPath();
-                    string sourceName = string.IsNullOrEmpty(containerUriPath)
+                    string subContainerPath = string.IsNullOrEmpty(containerUriPath)
                         ? current.Uri.GetPath()
                         : current.Uri.GetPath().Substring(containerUriPath.Length + 1);
-
-                    StreamToUriJobPart part;
-                    try
-                    {
-                        part = await StreamToUriJobPart.CreateJobPartAsync(
-                            job: this,
-                            partNumber: partNumber,
-                            sourceResource: (StorageResourceItem)current,
-                            destinationResource: _destinationResourceContainer.GetStorageResourceReference(sourceName))
-                            .ConfigureAwait(false);
-                        AppendJobPart(part);
-                    }
-                    catch (Exception ex)
+                    StorageResourceContainer subContainer =
+                        _destinationResourceContainer.GetChildStorageResourceContainer(subContainerPath);
+                    await subContainer.CreateIfNotExistsAsync().ConfigureAwait(false);
+                }
+                else
+                {
+                    if (!existingSources.Contains(current.Uri))
                     {
-                        await InvokeFailedArgAsync(ex).ConfigureAwait(false);
-                        yield break;
+                        string containerUriPath = _sourceResourceContainer.Uri.GetPath();
+                        string sourceName = string.IsNullOrEmpty(containerUriPath)
+                            ? current.Uri.GetPath()
+                            : current.Uri.GetPath().Substring(containerUriPath.Length + 1);
+
+                        StreamToUriJobPart part;
+                        try
+                        {
+                            part = await StreamToUriJobPart.CreateJobPartAsync(
+                                job: this,
+                                partNumber: partNumber,
+                                sourceResource: (StorageResourceItem)current,
+                                destinationResource: _destinationResourceContainer.GetStorageResourceReference(sourceName))
+                                .ConfigureAwait(false);
+                            AppendJobPart(part);
+                        }
+                        catch (Exception ex)
+                        {
+                            await InvokeFailedArgAsync(ex).ConfigureAwait(false);
+                            yield break;
+                        }
+                        yield return part;
+                        partNumber++;
                     }
-                    yield return part;
-                    partNumber++;
                 }
             }
         }
diff --git a/sdk/storage/Azure.Storage.DataMovement/tests/BlobStorageResourceContainerTests.cs b/sdk/storage/Azure.Storage.DataMovement/tests/BlobStorageResourceContainerTests.cs
index 2f5206a3489fe..e3713589485de 100644
--- a/sdk/storage/Azure.Storage.DataMovement/tests/BlobStorageResourceContainerTests.cs
+++ b/sdk/storage/Azure.Storage.DataMovement/tests/BlobStorageResourceContainerTests.cs
@@ -12,6 +12,7 @@
 using Azure.Storage.Blobs.Tests;
 using Azure.Storage.DataMovement.Tests;
 using DMBlobs::Azure.Storage.DataMovement.Blobs;
+using Moq;
 using NUnit.Framework;
 
 namespace Azure.Storage.DataMovement.Blobs.Tests
@@ -114,5 +115,27 @@ public async Task GetChildStorageResourceAsync()
             Assert.IsNotNull(properties);
             Assert.IsNotNull(properties.ETag);
         }
+
+        [Test]
+        public void GetChildStorageResourceContainer()
+        {
+            // Arrange
+            Uri uri = new Uri("https://storageaccount.blob.core.windows.net/container");
+            Mock<BlobContainerClient> mock = new(uri, new BlobClientOptions());
+            mock.Setup(b => b.Uri).Returns(uri);
+
+            string prefix = "foo";
+            StorageResourceContainer containerResource =
+                new BlobStorageResourceContainer(mock.Object, new() { BlobDirectoryPrefix = prefix });
+
+            // Act
+            string childPath = "bar";
+            StorageResourceContainer childContainer = containerResource.GetChildStorageResourceContainer(childPath);
+
+            // Assert
+            UriBuilder builder = new UriBuilder(containerResource.Uri);
+            builder.Path = string.Join("/", builder.Path, childPath);
+            Assert.AreEqual(builder.Uri, childContainer.Uri);
+        }
     }
 }
diff --git a/sdk/storage/Azure.Storage.DataMovement/tests/LocalDirectoryStorageResourceTests.cs b/sdk/storage/Azure.Storage.DataMovement/tests/LocalDirectoryStorageResourceTests.cs
index e0801b9795729..b136b68175cb8 100644
--- a/sdk/storage/Azure.Storage.DataMovement/tests/LocalDirectoryStorageResourceTests.cs
+++ b/sdk/storage/Azure.Storage.DataMovement/tests/LocalDirectoryStorageResourceTests.cs
@@ -3,6 +3,7 @@
 
 using System;
 using System.Collections.Generic;
+using System.IO;
 using System.Linq;
 using System.Threading.Tasks;
 using NUnit.Framework;
@@ -134,5 +135,20 @@ public async Task GetChildStorageResourceAsync_SubDir()
                 await resource.GetPropertiesAsync().ConfigureAwait(false);
             }
         }
+
+        [Test]
+        public void GetChildStorageResourceContainer()
+        {
+            using DisposingLocalDirectory test = DisposingLocalDirectory.GetTestDirectory();
+            string folderPath = test.DirectoryPath;
+
+            StorageResourceContainer container = new LocalDirectoryStorageResourceContainer(folderPath);
+
+            string childPath = "childPath";
+            StorageResourceContainer childContainer = container.GetChildStorageResourceContainer(childPath);
+
+            string fullPath = Path.Combine(folderPath, childPath);
+            Assert.AreEqual(childContainer.Uri, new Uri(fullPath));
+        }
     }
 }
diff --git a/sdk/storage/Azure.Storage.DataMovement/tests/Shared/MemoryStorageResourceContainer.cs b/sdk/storage/Azure.Storage.DataMovement/tests/Shared/MemoryStorageResourceContainer.cs
index 757a7daea9d21..8164d12ef050d 100644
--- a/sdk/storage/Azure.Storage.DataMovement/tests/Shared/MemoryStorageResourceContainer.cs
+++ b/sdk/storage/Azure.Storage.DataMovement/tests/Shared/MemoryStorageResourceContainer.cs
@@ -93,5 +93,15 @@ private IEnumerable<StorageResource> GetStorageResources(bool includeContainers)
                 }
             }
         }
+
+        protected internal override Task CreateIfNotExistsAsync(CancellationToken cancellationToken = default)
+        {
+            throw new NotImplementedException();
+        }
+
+        protected internal override StorageResourceContainer GetChildStorageResourceContainer(string path)
+        {
+            throw new NotImplementedException();
+        }
     }
 }
diff --git a/sdk/storage/Azure.Storage.DataMovement/tests/Shared/StartTransferDirectoryCopyTestBase.cs b/sdk/storage/Azure.Storage.DataMovement/tests/Shared/StartTransferDirectoryCopyTestBase.cs
new file mode 100644
index 0000000000000..2f834ebafee8d
--- /dev/null
+++ b/sdk/storage/Azure.Storage.DataMovement/tests/Shared/StartTransferDirectoryCopyTestBase.cs
@@ -0,0 +1,907 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
+
+using System;
+using System.IO;
+using System.Linq;
+using System.Threading;
+using System.Threading.Tasks;
+using Azure.Core;
+using Azure.Core.TestFramework;
+using Azure.Storage.Test.Shared;
+using NUnit.Framework;
+
+namespace Azure.Storage.DataMovement.Tests
+{
+    public abstract class StartTransferDirectoryCopyTestBase
+            <TSourceServiceClient,
+            TSourceContainerClient,
+            TSourceClientOptions,
+            TDestinationServiceClient,
+            TDestinationContainerClient,
+            TDestinationClientOptions,
+            TEnvironment> : StorageTestBase<TEnvironment>
+        where TSourceServiceClient : class
+        where TSourceContainerClient : class
+        where TSourceClientOptions : ClientOptions
+        where TDestinationServiceClient : class
+        where TDestinationContainerClient : class
+        where TDestinationClientOptions : ClientOptions
+        where TEnvironment : StorageTestEnvironment, new()
+    {
+        private readonly string _generatedResourceNamePrefix;
+        private readonly string _expectedOverwriteExceptionMessage;
+        private readonly string _firstItemName;
+
+        public ClientBuilder<TSourceServiceClient, TSourceClientOptions> SourceClientBuilder { get; protected set; }
+        public ClientBuilder<TDestinationServiceClient, TDestinationClientOptions> DestinationClientBuilder { get; protected set; }
+
+        /// <summary>
+        /// Constructor for TransferManager.StartTransferAsync tests
+        ///
+        /// The async is defaulted to true, since we do not have sync StartTransfer methods.
+        /// </summary>
+        /// <param name="generatedResourcenamePrefix"></param>
+        /// <param name="mode"></param>
+        public StartTransferDirectoryCopyTestBase(
+            bool async,
+            string expectedOverwriteExceptionMessage,
+            string generatedResourceNamePrefix = default,
+            RecordedTestMode? mode = null) : base(async, mode)
+        {
+            Argument.CheckNotNullOrEmpty(expectedOverwriteExceptionMessage, nameof(expectedOverwriteExceptionMessage));
+            _generatedResourceNamePrefix = generatedResourceNamePrefix ?? "test-resource-";
+            _expectedOverwriteExceptionMessage = expectedOverwriteExceptionMessage;
+            _firstItemName = "item1";
+        }
+
+        #region Service-Specific Methods
+        /// <summary>
+        /// Gets the service client using OAuth to authenticate.
+        /// </summary>
+        protected abstract TSourceContainerClient GetOAuthSourceContainerClient(string containerName);
+
+        /// <summary>
+        /// Gets a service-specific disposing container for use with tests in this class.
+        /// </summary>
+        /// <param name="service">Optionally specified service client to get container from.</param>
+        /// <param name="containerName">Optional container name specification.</param>
+        protected abstract Task<IDisposingContainer<TSourceContainerClient>> GetSourceDisposingContainerAsync(
+            TSourceServiceClient service = default,
+            string containerName = default,
+            CancellationToken cancellationToken = default);
+
+        /// <summary>
+        /// Gets the specific storage resource from the given TDestinationObjectClient
+        /// e.g. ShareFileClient to a ShareFileStorageResource, BlockBlobClient to a BlockBlobStorageResource.
+        /// </summary>
+        /// <param name="containerClient">The object client to create the storage resource object.</param>
+        /// <param name="directoryPath">The path of the directory.</param>
+        /// <returns></returns>
+        protected abstract StorageResourceContainer GetSourceStorageResourceContainer(TSourceContainerClient containerClient, string directoryPath);
+
+        /// <summary>
+        /// Creates the directory within the source container.
+        /// </summary>
+        /// <param name="sourceContainer">
+        /// The respective source container to create the directory in.
+        /// </param>
+        /// <param name="directoryPath">
+        /// The directory path.
+        /// </param>
+        protected abstract Task CreateDirectoryInSourceAsync(
+            TSourceContainerClient sourceContainer,
+            string directoryPath,
+            CancellationToken cancellationToken = default);
+
+        /// <summary>
+        /// Creates the object in the source storage resource container.
+        /// </summary>
+        /// <param name="objectLength">The length to create the object of.</param>
+        /// <param name="objectName">The name of the object to create.</param>
+        /// <param name="contents">The contents to set in the object.</param>
+        /// <returns></returns>
+        protected abstract Task CreateObjectInSourceAsync(
+            TSourceContainerClient container,
+            long? objectLength = null,
+            string objectName = null,
+            Stream contents = default,
+            CancellationToken cancellationToken = default);
+
+        /// <summary>
+        /// Gets a service-specific disposing container for use with tests in this class.
+        /// </summary>
+        /// <param name="service">Optionally specified service client to get container from.</param>
+        /// <param name="containerName">Optional container name specification.</param>
+        protected abstract Task<IDisposingContainer<TDestinationContainerClient>> GetDestinationDisposingContainerAsync(
+            TDestinationServiceClient service = default,
+            string containerName = default,
+            CancellationToken cancellationToken = default);
+
+        /// <summary>
+        /// Gets the service client using OAuth to authenticate.
+        /// </summary>
+        protected abstract TDestinationContainerClient GetOAuthDestinationContainerClient(string containerName);
+
+        /// <summary>
+        /// Gets the specific storage resource from the given TDestinationObjectClient
+        /// e.g. ShareFileClient to a ShareFileStorageResource, BlockBlobClient to a BlockBlobStorageResource.
+        /// </summary>
+        /// <param name="directoryClient">The object client to create the storage resource object.</param>
+        /// <returns></returns>
+        protected abstract StorageResourceContainer GetDestinationStorageResourceContainer(TDestinationContainerClient sourceContainerClient, string directoryPath);
+
+        /// <summary>
+        /// Creates the directory within the source container. Will also create any parent directories if required and is a hierarchical structure.
+        /// </summary>
+        /// <param name="sourceContainer">
+        /// The respective source container to create the directory in.
+        /// </param>
+        /// <param name="directoryPath">
+        /// The directory path. If parent paths are required, will also create any parent directories if required and is a hierarchical structure.
+        /// </param>
+        protected abstract Task CreateDirectoryInDestinationAsync(
+            TDestinationContainerClient destinationContainer,
+            string directoryPath,
+            CancellationToken cancellationToken = default);
+
+        /// <summary>
+        /// Creates the object in the source storage resource container.
+        /// </summary>
+        /// <param name="objectLength">The length to create the object of.</param>
+        /// <param name="objectName">The name of the object to create.</param>
+        /// <param name="contents">The contents to set in the object.</param>
+        /// <returns></returns>
+        protected abstract Task CreateObjectInDestinationAsync(
+            TDestinationContainerClient container,
+            long? objectLength = null,
+            string objectName = null,
+            Stream contents = default,
+            CancellationToken cancellationToken = default);
+
+        /// <summary>
+        /// Verifies that the destination container is empty when we expect it to be.
+        /// </summary>
+        /// <param name="destinationContainer">
+        /// The respective destination container to verify empty contents.
+        /// </param>
+        /// <returns></returns>
+        protected abstract Task VerifyEmptyDestinationContainerAsync(
+            TDestinationContainerClient destinationContainer,
+            string destinationPrefix,
+            CancellationToken cancellationToken = default);
+
+        /// <summary>
+        /// Verifies the results between the source and the destination container.
+        /// </summary>
+        /// <param name="sourceContainer">The source client to check the contents and compare against the destination.</param>
+        /// <param name="destinationContainer">The destinatiojn client to check the contents and compare against the source.</param>
+        /// <param name="sourcePrefix">Optional. The prefix to start listing at the source container.</param>
+        /// <param name="destinationPrefix">Optional. The prefix to start listing at the destination container.</param>
+        /// <returns></returns>
+        protected abstract Task VerifyResultsAsync(
+            TSourceContainerClient sourceContainer,
+            string sourcePrefix,
+            TDestinationContainerClient destinationContainer,
+            string destinationPrefix,
+            CancellationToken cancellationToken = default);
+        #endregion
+
+        protected string GetNewObjectName()
+            => _generatedResourceNamePrefix + SourceClientBuilder.Recording.Random.NewGuid();
+
+        /// <summary>
+        /// Upload and verify the contents of the items
+        ///
+        /// By default in this function an event argument will be added to the options event handler
+        /// to detect when the upload has finished.
+        /// </summary>
+        /// <param name="sourceContainer">The source container which will contains the source items</param>
+        /// <param name="sourcePrefix">The source prefix/folder</param>
+        /// <param name="destinationPrefix">The destination local path to download the items to</param>
+        /// <param name="waitTimeInSec">
+        /// How long we should wait until we cancel the operation. If this timeout is reached the test will fail.
+        /// </param>
+        /// <param name="transferManagerOptions">Options for the transfer manager</param>
+        /// <param name="options">Options for the transfer Options</param>
+        /// <returns></returns>
+        private async Task CopyDirectoryAndVerifyAsync(
+            TSourceContainerClient sourceContainer,
+            TDestinationContainerClient destinationContainer,
+            string sourcePrefix,
+            string destinationPrefix,
+            int itemTransferCount,
+            int waitTimeInSec = 30,
+            TransferManagerOptions transferManagerOptions = default,
+            DataTransferOptions options = default)
+        {
+            // Set transfer options
+            options ??= new DataTransferOptions();
+            TestEventsRaised testEventFailed = new TestEventsRaised(options);
+
+            transferManagerOptions ??= new TransferManagerOptions()
+            {
+                ErrorHandling = DataTransferErrorMode.ContinueOnFailure
+            };
+
+            // Initialize transferManager
+            TransferManager transferManager = new TransferManager(transferManagerOptions);
+
+            StorageResourceContainer sourceResource =
+                GetSourceStorageResourceContainer(sourceContainer, sourcePrefix);
+            StorageResourceContainer destinationResource =
+                GetDestinationStorageResourceContainer(destinationContainer, destinationPrefix);
+
+            DataTransfer transfer = await transferManager.StartTransferAsync(sourceResource, destinationResource, options);
+
+            // Assert
+            CancellationTokenSource tokenSource = new CancellationTokenSource(TimeSpan.FromSeconds(waitTimeInSec));
+            await transfer.WaitForCompletionAsync(tokenSource.Token);
+
+            await testEventFailed.AssertContainerCompletedCheck(itemTransferCount);
+            Assert.IsTrue(transfer.HasCompleted);
+            Assert.AreEqual(DataTransferState.Completed, transfer.TransferStatus.State);
+
+            // List all files in source folder path
+            await VerifyResultsAsync(
+                sourceContainer: sourceContainer,
+                sourcePrefix: sourcePrefix,
+                destinationContainer: destinationContainer,
+                destinationPrefix: destinationPrefix);
+        }
+
+        [Test]
+        [LiveOnly] // https://github.com/Azure/azure-sdk-for-net/issues/33082
+        [TestCase(0, 10)]
+        [TestCase(100, 10)]
+        [TestCase(Constants.KB, 10)]
+        public async Task DirectoryToDirectory_SmallSize(long size, int waitTimeInSec)
+        {
+            // Arrange
+            await using IDisposingContainer<TSourceContainerClient> source = await GetSourceDisposingContainerAsync();
+            await using IDisposingContainer<TDestinationContainerClient> destination = await GetDestinationDisposingContainerAsync();
+            string sourcePrefix = "sourceFolder";
+            string destinationPrefix = "destinationFolder";
+
+            await CreateDirectoryInSourceAsync(source.Container, sourcePrefix);
+            string itemName1 = string.Join("/", sourcePrefix, GetNewObjectName());
+            string itemName2 = string.Join("/", sourcePrefix, GetNewObjectName());
+            await CreateObjectInSourceAsync(source.Container, size, itemName1);
+            await CreateObjectInSourceAsync(source.Container, size, itemName2);
+
+            string subDirName = string.Join("/", sourcePrefix, "bar");
+            await CreateDirectoryInSourceAsync(source.Container, subDirName);
+            string itemName3 = string.Join("/", subDirName, GetNewObjectName());
+            await CreateObjectInSourceAsync(source.Container, size, itemName3);
+
+            string subDirName2 = string.Join("/", sourcePrefix, "pik");
+            await CreateDirectoryInSourceAsync(source.Container, subDirName2);
+            string itemName4 = string.Join("/", subDirName2, GetNewObjectName());
+            await CreateObjectInSourceAsync(source.Container, size, itemName4);
+
+            await CreateDirectoryInDestinationAsync(destination.Container, destinationPrefix);
+
+            await CopyDirectoryAndVerifyAsync(
+                source.Container,
+                destination.Container,
+                sourcePrefix,
+                destinationPrefix,
+                4,
+                waitTimeInSec).ConfigureAwait(false);
+        }
+
+        [Ignore("These tests currently take 40+ mins for little additional coverage")]
+        [Test]
+        [LiveOnly]
+        [TestCase(4 * Constants.MB, 20)]
+        [TestCase(4 * Constants.MB, 200)]
+        [TestCase(257 * Constants.MB, 500)]
+        [TestCase(Constants.GB, 500)]
+        public async Task DirectoryToDirectory_LargeSize(long size, int waitTimeInSec)
+        {
+            // Arrange
+            await using IDisposingContainer<TSourceContainerClient> source = await GetSourceDisposingContainerAsync();
+            await using IDisposingContainer<TDestinationContainerClient> destination = await GetDestinationDisposingContainerAsync();
+            string sourcePrefix = "sourceFolder";
+
+            await CreateDirectoryInSourceAsync(source.Container, sourcePrefix);
+            string itemName1 = string.Join("/", sourcePrefix, GetNewObjectName());
+            string itemName2 = string.Join("/", sourcePrefix, GetNewObjectName());
+            await CreateObjectInSourceAsync(source.Container, size, itemName1);
+            await CreateObjectInSourceAsync(source.Container, size, itemName2);
+
+            string subDirName = string.Join("/", sourcePrefix, "bar");
+            await CreateDirectoryInSourceAsync(source.Container, subDirName);
+            string itemName3 = string.Join("/", subDirName, GetNewObjectName());
+            await CreateObjectInSourceAsync(source.Container, size, itemName3);
+
+            string subDirName2 = string.Join("/", sourcePrefix, "pik");
+            await CreateDirectoryInSourceAsync(source.Container, subDirName2);
+            string itemName4 = string.Join("/", subDirName2, GetNewObjectName());
+            await CreateObjectInSourceAsync(source.Container, size, itemName4);
+
+            string destinationPrefix = "destFolder";
+            await CreateDirectoryInDestinationAsync(destination.Container, destinationPrefix);
+
+            await CopyDirectoryAndVerifyAsync(
+                source.Container,
+                destination.Container,
+                sourcePrefix,
+                destinationPrefix,
+                waitTimeInSec).ConfigureAwait(false);
+        }
+
+        [Test]
+        [LiveOnly] // https://github.com/Azure/azure-sdk-for-net/issues/33082
+        public async Task DirectoryToDirectory_EmptyFolder()
+        {
+            // Arrange
+            await using IDisposingContainer<TSourceContainerClient> source = await GetSourceDisposingContainerAsync();
+            await using IDisposingContainer<TDestinationContainerClient> destination = await GetDestinationDisposingContainerAsync();
+            var sourcePath = GetNewObjectName();
+            var destinationPath = GetNewObjectName();
+
+            // Set up resources
+            await CreateDirectoryInSourceAsync(source.Container, sourcePath);
+            StorageResourceContainer sourceResource = GetSourceStorageResourceContainer(source.Container, sourcePath);
+            await CreateDirectoryInDestinationAsync(destination.Container, destinationPath);
+            StorageResourceContainer destinationResource = GetDestinationStorageResourceContainer(destination.Container, destinationPath);
+
+            TransferManagerOptions managerOptions = new TransferManagerOptions()
+            {
+                ErrorHandling = DataTransferErrorMode.ContinueOnFailure,
+                MaximumConcurrency = 1,
+            };
+            TransferManager transferManager = new TransferManager(managerOptions);
+            DataTransferOptions options = new DataTransferOptions();
+            TestEventsRaised testEventsRaised = new TestEventsRaised(options);
+
+            // Act
+            DataTransfer transfer = await transferManager.StartTransferAsync(sourceResource, destinationResource, options);
+
+            CancellationTokenSource tokenSource = new CancellationTokenSource(TimeSpan.FromSeconds(10));
+            await transfer.WaitForCompletionAsync(tokenSource.Token);
+            Assert.IsTrue(transfer.HasCompleted);
+            Assert.AreEqual(DataTransferState.Completed, transfer.TransferStatus.State);
+
+            // Assert
+            await VerifyEmptyDestinationContainerAsync(destination.Container, destinationPath);
+            testEventsRaised.AssertUnexpectedFailureCheck();
+        }
+
+        [Test]
+        [LiveOnly] // https://github.com/Azure/azure-sdk-for-net/issues/33082
+        public async Task DirectoryToDirectory_SingleFile()
+        {
+            // Arrange
+            await using IDisposingContainer<TSourceContainerClient> source = await GetSourceDisposingContainerAsync();
+            await using IDisposingContainer<TDestinationContainerClient> destination = await GetDestinationDisposingContainerAsync();
+
+            string sourcePrefix = "sourceFolder";
+
+            string itemName1 = string.Join("/", sourcePrefix, GetNewObjectName());
+            await CreateDirectoryInSourceAsync(source.Container, sourcePrefix);
+            await CreateObjectInSourceAsync(source.Container, Constants.KB, itemName1);
+
+            string destinationPrefix = "destFolder";
+
+            await CreateDirectoryInDestinationAsync(destination.Container, destinationPrefix);
+            await CopyDirectoryAndVerifyAsync(
+                sourceContainer: source.Container,
+                destinationContainer: destination.Container,
+                sourcePrefix: sourcePrefix,
+                destinationPrefix: destinationPrefix,
+                itemTransferCount: 1).ConfigureAwait(false);
+        }
+
+        [Test]
+        [LiveOnly] // https://github.com/Azure/azure-sdk-for-net/issues/33082
+        public async Task DirectoryToDirectory_ManySubDirectories()
+        {
+            // Arrange
+            await using IDisposingContainer<TSourceContainerClient> source = await GetSourceDisposingContainerAsync();
+            await using IDisposingContainer<TDestinationContainerClient> destination = await GetDestinationDisposingContainerAsync();
+
+            string sourcePrefix = "sourceFolder";
+
+            await CreateDirectoryInSourceAsync(source.Container, sourcePrefix);
+            string subDir1 = string.Join("/", sourcePrefix, "foo");
+            await CreateDirectoryInSourceAsync(source.Container, subDir1);
+            string itemName1 = string.Join("/", subDir1, GetNewObjectName());
+            await CreateObjectInSourceAsync(source.Container, Constants.KB, itemName1);
+            string subDir2 = string.Join("/", sourcePrefix, "rul");
+            await CreateDirectoryInSourceAsync(source.Container, subDir2);
+            string itemName2 = string.Join("/", subDir2, GetNewObjectName());
+            await CreateObjectInSourceAsync(source.Container, Constants.KB, itemName2);
+            string subDir3 = string.Join("/", sourcePrefix, "pik");
+            await CreateDirectoryInSourceAsync(source.Container, subDir3);
+            string itemName3 = string.Join("/", subDir3, GetNewObjectName());
+            await CreateObjectInSourceAsync(source.Container, Constants.KB, itemName3);
+
+            string destinationPrefix = "destFolder";
+
+            await CreateDirectoryInDestinationAsync(destination.Container, destinationPrefix);
+            await CopyDirectoryAndVerifyAsync(
+                sourceContainer: source.Container,
+                destinationContainer: destination.Container,
+                sourcePrefix: sourcePrefix,
+                destinationPrefix: destinationPrefix,
+                itemTransferCount: 3).ConfigureAwait(false);
+        }
+
+        [Test]
+        [LiveOnly] // https://github.com/Azure/azure-sdk-for-net/issues/33082
+        [TestCase(1)]
+        [TestCase(2)]
+        [TestCase(3)]
+        public async Task DirectoryToDirectory_SubDirectoriesLevels(int level)
+        {
+            // Arrange
+            await using IDisposingContainer<TSourceContainerClient> source = await GetSourceDisposingContainerAsync();
+            await using IDisposingContainer<TDestinationContainerClient> destination = await GetDestinationDisposingContainerAsync();
+
+            string sourcePrefix = "sourceFolder";
+
+            await CreateDirectoryInSourceAsync(source.Container, sourcePrefix);
+
+            string subDirPrefix = sourcePrefix;
+            for (int i = 0; i < level; i++)
+            {
+                subDirPrefix = string.Join("/", subDirPrefix, $"folder{i}");
+                await CreateDirectoryInSourceAsync(source.Container, subDirPrefix);
+                string fullFilePath = string.Join("/", subDirPrefix, GetNewObjectName());
+                await CreateObjectInSourceAsync(source.Container, Constants.KB, fullFilePath);
+            }
+
+            string destinationPrefix = "destFolder";
+            await CreateDirectoryInDestinationAsync(destination.Container, destinationPrefix);
+
+            await CopyDirectoryAndVerifyAsync(
+                source.Container,
+                destination.Container,
+                sourcePrefix,
+                destinationPrefix,
+                itemTransferCount: level).ConfigureAwait(false);
+        }
+
+        [Test]
+        [LiveOnly] // https://github.com/Azure/azure-sdk-for-net/issues/33082
+        public async Task DirectoryToDirectory_OverwriteExists()
+        {
+            // Arrange
+            await using IDisposingContainer<TSourceContainerClient> source = await GetSourceDisposingContainerAsync();
+            await using IDisposingContainer<TDestinationContainerClient> destination = await GetDestinationDisposingContainerAsync();
+
+            long size = Constants.KB;
+            string sourcePrefix = "sourceFolder";
+            string destinationPrefix = "destPrefix";
+
+            await CreateDirectoryInSourceAsync(source.Container, sourcePrefix);
+            await CreateDirectoryInDestinationAsync(destination.Container, destinationPrefix);
+
+            string itemName1 = string.Join("/", sourcePrefix, _firstItemName);
+            await CreateObjectInSourceAsync(source.Container, size, itemName1);
+
+            // Create same object in the destination, so when both files are seen overwrite will trigger.
+            string destItemName1 = string.Join("/", destinationPrefix, _firstItemName);
+            await CreateObjectInDestinationAsync(destination.Container, size, destItemName1);
+
+            string itemName2 = string.Join("/", sourcePrefix, GetNewObjectName());
+            await CreateObjectInSourceAsync(source.Container, size, itemName2);
+
+            string subDirName = string.Join("/", sourcePrefix, "bar");
+            await CreateDirectoryInSourceAsync(source.Container, subDirName);
+            string itemName3 = string.Join("/", subDirName, GetNewObjectName());
+            await CreateObjectInSourceAsync(source.Container, size, itemName3);
+
+            string subDirName2 = string.Join("/", sourcePrefix, "pik");
+            await CreateDirectoryInSourceAsync(source.Container, subDirName2);
+            string itemName4 = string.Join("/", subDirName2, GetNewObjectName());
+            await CreateObjectInSourceAsync(source.Container, size, itemName4);
+
+            DataTransferOptions options = new DataTransferOptions()
+            {
+                CreationPreference = StorageResourceCreationPreference.OverwriteIfExists
+            };
+
+            // Act
+            await CopyDirectoryAndVerifyAsync(
+                source.Container,
+                destination.Container,
+                sourcePrefix,
+                destinationPrefix,
+                itemTransferCount: 4,
+                options: options).ConfigureAwait(false);
+        }
+
+        [Test]
+        [LiveOnly] // https://github.com/Azure/azure-sdk-for-net/issues/33082
+        public async Task DirectoryToDirectory_OverwriteNotExists()
+        {
+            // Arrange
+            await using IDisposingContainer<TSourceContainerClient> source = await GetSourceDisposingContainerAsync();
+            await using IDisposingContainer<TDestinationContainerClient> destination = await GetDestinationDisposingContainerAsync();
+
+            long size = Constants.KB;
+            string sourcePrefix = "sourceFolder";
+            string destinationPrefix = "destPrefix";
+
+            await CreateDirectoryInSourceAsync(source.Container, sourcePrefix);
+            await CreateDirectoryInDestinationAsync(destination.Container, destinationPrefix);
+
+            string itemName1 = string.Join("/", sourcePrefix, GetNewObjectName());
+            await CreateObjectInSourceAsync(source.Container, size, itemName1);
+
+            string itemName2 = string.Join("/", sourcePrefix, GetNewObjectName());
+            await CreateObjectInSourceAsync(source.Container, size, itemName2);
+
+            string subDirName = string.Join("/", sourcePrefix, "bar");
+            await CreateDirectoryInSourceAsync(source.Container, subDirName);
+            string itemName3 = string.Join("/", subDirName, GetNewObjectName());
+            await CreateObjectInSourceAsync(source.Container, size, itemName3);
+
+            string subDirName2 = string.Join("/", sourcePrefix, "pik");
+            await CreateDirectoryInSourceAsync(source.Container, subDirName2);
+            string itemName4 = string.Join("/", subDirName2, GetNewObjectName());
+            await CreateObjectInSourceAsync(source.Container, size, itemName4);
+
+            DataTransferOptions options = new DataTransferOptions()
+            {
+                CreationPreference = StorageResourceCreationPreference.OverwriteIfExists
+            };
+
+            // Act
+            await CopyDirectoryAndVerifyAsync(
+                source.Container,
+                destination.Container,
+                sourcePrefix,
+                destinationPrefix,
+                itemTransferCount: 4,
+                options: options).ConfigureAwait(false);
+        }
+
+        [Test]
+        [LiveOnly] // https://github.com/Azure/azure-sdk-for-net/issues/33082
+        public async Task DirectoryToDirectory_OAuth()
+        {
+            // Arrange
+            long size = Constants.KB;
+            int waitTimeInSec = 20;
+            string sourceContainerName = GetNewObjectName();
+            await using IDisposingContainer<TSourceContainerClient> source = await GetSourceDisposingContainerAsync(containerName: sourceContainerName);
+            TSourceContainerClient oauthSourceContainer = GetOAuthSourceContainerClient(containerName: sourceContainerName);
+
+            await using IDisposingContainer<TDestinationContainerClient> destination = await GetDestinationDisposingContainerAsync();
+            TDestinationContainerClient oauthDestinationContainer = GetOAuthDestinationContainerClient(containerName: sourceContainerName);
+
+            string sourcePrefix = "sourceFolder";
+            string destinationPrefix = "destFolder";
+
+            await CreateDirectoryInSourceAsync(oauthSourceContainer, sourcePrefix);
+            await CreateDirectoryInDestinationAsync(oauthDestinationContainer, destinationPrefix);
+
+            string itemName1 = string.Join("/", sourcePrefix, GetNewObjectName());
+            await CreateObjectInSourceAsync(source.Container, size, itemName1);
+
+            string itemName2 = string.Join("/", sourcePrefix, GetNewObjectName());
+            await CreateObjectInSourceAsync(source.Container, size, itemName2);
+
+            string subDirName = string.Join("/", sourcePrefix, "bar");
+            await CreateDirectoryInSourceAsync(source.Container, subDirName);
+            string itemName3 = string.Join("/", subDirName, GetNewObjectName());
+            await CreateObjectInSourceAsync(source.Container, size, itemName3);
+
+            string subDirName2 = string.Join("/", sourcePrefix, "pik");
+            await CreateDirectoryInSourceAsync(source.Container, subDirName2);
+            string itemName4 = string.Join("/", subDirName2, GetNewObjectName());
+            await CreateObjectInSourceAsync(source.Container, size, itemName4);
+
+            await CopyDirectoryAndVerifyAsync(
+                oauthSourceContainer,
+                oauthDestinationContainer,
+                sourcePrefix,
+                destinationPrefix,
+                4,
+                waitTimeInSec).ConfigureAwait(false);
+        }
+
+        #region Single Concurrency
+        private async Task CreateDirectoryTree(
+            TSourceContainerClient client,
+            string sourcePrefix,
+            int size)
+        {
+            string itemName1 = string.Join("/", sourcePrefix, _firstItemName);
+            string itemName2 = string.Join("/", sourcePrefix, "item2");
+            await CreateObjectInSourceAsync(client, size, itemName1);
+            await CreateObjectInSourceAsync(client, size, itemName2);
+
+            string subDirPath = string.Join("/", sourcePrefix, "bar");
+            await CreateDirectoryInSourceAsync(client, subDirPath);
+            string itemName3 = string.Join("/", subDirPath, "item3");
+            await CreateObjectInSourceAsync(client, size, itemName3);
+
+            string subDirPath2 = string.Join("/", sourcePrefix, "pik");
+            await CreateDirectoryInSourceAsync(client, subDirPath2);
+            string itemName4 = string.Join("/", subDirPath2, "item4");
+            await CreateObjectInSourceAsync(client, size, itemName4);
+        }
+
+        private async Task<DataTransfer> CreateStartTransfer(
+            TSourceContainerClient sourceContainer,
+            TDestinationContainerClient destinationContainer,
+            int concurrency,
+            bool createFailedCondition = false,
+            DataTransferOptions options = default,
+            int size = Constants.KB)
+        {
+            // Arrange
+            string sourcePrefix = "sourceFolder";
+            string destPrefix = "destFolder";
+            await CreateDirectoryInSourceAsync(sourceContainer, sourcePrefix);
+            await CreateDirectoryInDestinationAsync(destinationContainer, destPrefix);
+            await CreateDirectoryTree(sourceContainer, sourcePrefix, size);
+
+            // Create storage resource containers
+            StorageResourceContainer sourceResource = GetSourceStorageResourceContainer(sourceContainer, sourcePrefix);
+            StorageResourceContainer destinationResource = GetDestinationStorageResourceContainer(destinationContainer, destPrefix);
+
+            if (createFailedCondition)
+            {
+                // To create an expected failure, create an item that is supposed to be transferred over.
+                // If we don't enable overwrite, a failure should be thrown.
+                string fullDestPath = string.Join("/", destPrefix, _firstItemName);
+                await CreateObjectInDestinationAsync(destinationContainer, size, fullDestPath);
+            }
+
+            // Create Transfer Manager with single threaded operation
+            TransferManagerOptions managerOptions = new TransferManagerOptions()
+            {
+                MaximumConcurrency = concurrency,
+            };
+            TransferManager transferManager = new TransferManager(managerOptions);
+
+            // Start transfer and await for completion.
+            return await transferManager.StartTransferAsync(
+                sourceResource,
+                destinationResource,
+                options).ConfigureAwait(false);
+        }
+
+        [Test]
+        [LiveOnly] // https://github.com/Azure/azure-sdk-for-net/issues/33082
+        public async Task StartTransfer_AwaitCompletion()
+        {
+            // Arrange
+            await using IDisposingContainer<TSourceContainerClient> source = await GetSourceDisposingContainerAsync();
+            await using IDisposingContainer<TDestinationContainerClient> destination = await GetDestinationDisposingContainerAsync();
+
+            // Create transfer to do a AwaitCompletion
+            DataTransferOptions options = new DataTransferOptions();
+            TestEventsRaised testEventsRaised = new TestEventsRaised(options);
+            DataTransfer transfer = await CreateStartTransfer(
+                source.Container,
+                destination.Container,
+                1,
+                options: options);
+
+            // Act
+            CancellationTokenSource cancellationTokenSource = new CancellationTokenSource(TimeSpan.FromSeconds(30));
+            await transfer.WaitForCompletionAsync(cancellationTokenSource.Token).ConfigureAwait(false);
+
+            // Assert
+            testEventsRaised.AssertUnexpectedFailureCheck();
+            Assert.NotNull(transfer);
+            Assert.IsTrue(transfer.HasCompleted);
+            Assert.AreEqual(DataTransferState.Completed, transfer.TransferStatus.State);
+        }
+
+        [Test]
+        [LiveOnly] // https://github.com/Azure/azure-sdk-for-net/issues/33082
+        public async Task StartTransfer_AwaitCompletion_Failed()
+        {
+            // Arrange
+            await using IDisposingContainer<TSourceContainerClient> source = await GetSourceDisposingContainerAsync();
+            await using IDisposingContainer<TDestinationContainerClient> destination = await GetDestinationDisposingContainerAsync();
+
+            DataTransferOptions options = new DataTransferOptions()
+            {
+                CreationPreference = StorageResourceCreationPreference.FailIfExists
+            };
+            TestEventsRaised testEventsRaised = new TestEventsRaised(options);
+
+            // Create transfer to do a AwaitCompletion
+            DataTransfer transfer = await CreateStartTransfer(
+                source.Container,
+                destination.Container,
+                1,
+                createFailedCondition: true,
+                options: options);
+
+            // Act
+            CancellationTokenSource cancellationTokenSource = new CancellationTokenSource(TimeSpan.FromSeconds(30));
+            await transfer.WaitForCompletionAsync(cancellationTokenSource.Token).ConfigureAwait(false);
+
+            // Assert
+            Assert.NotNull(transfer);
+            Assert.IsTrue(transfer.HasCompleted);
+            Assert.AreEqual(DataTransferState.Completed, transfer.TransferStatus.State);
+            Assert.AreEqual(true, transfer.TransferStatus.HasFailedItems);
+            await testEventsRaised.AssertContainerCompletedWithFailedCheck(1);
+            Assert.IsTrue(testEventsRaised.FailedEvents.First().Exception.Message.Contains(_expectedOverwriteExceptionMessage));
+        }
+
+        [Test]
+        [LiveOnly] // https://github.com/Azure/azure-sdk-for-net/issues/33082
+        public async Task StartTransfer_AwaitCompletion_Skipped()
+        {
+            // Arrange
+            await using IDisposingContainer<TSourceContainerClient> source = await GetSourceDisposingContainerAsync();
+            await using IDisposingContainer<TDestinationContainerClient> destination = await GetDestinationDisposingContainerAsync();
+
+            // Create transfer options with Skipping available
+            DataTransferOptions options = new DataTransferOptions()
+            {
+                CreationPreference = StorageResourceCreationPreference.SkipIfExists
+            };
+            TestEventsRaised testEventsRaised = new TestEventsRaised(options);
+
+            // Create transfer to do a AwaitCompletion
+            DataTransfer transfer = await CreateStartTransfer(
+                source.Container,
+                destination.Container,
+                1,
+                createFailedCondition: true,
+                options: options);
+
+            // Act
+            CancellationTokenSource cancellationTokenSource = new CancellationTokenSource(TimeSpan.FromSeconds(30));
+            await transfer.WaitForCompletionAsync(cancellationTokenSource.Token).ConfigureAwait(false);
+
+            // Assert
+            Assert.NotNull(transfer);
+            Assert.IsTrue(transfer.HasCompleted);
+            Assert.AreEqual(DataTransferState.Completed, transfer.TransferStatus.State);
+            Assert.AreEqual(true, transfer.TransferStatus.HasSkippedItems);
+            await testEventsRaised.AssertContainerCompletedWithSkippedCheck(1);
+        }
+
+        [Test]
+        [LiveOnly] // https://github.com/Azure/azure-sdk-for-net/issues/33082
+        public async Task StartTransfer_EnsureCompleted()
+        {
+            // Arrange
+            await using IDisposingContainer<TSourceContainerClient> source = await GetSourceDisposingContainerAsync();
+            await using IDisposingContainer<TDestinationContainerClient> destination = await GetDestinationDisposingContainerAsync();
+
+            // Create transfer to do a EnsureCompleted
+            DataTransferOptions options = new DataTransferOptions();
+            TestEventsRaised testEventsRaised = new TestEventsRaised(options);
+
+            DataTransfer transfer = await CreateStartTransfer(
+                source.Container,
+                destination.Container,
+                1,
+                options: options);
+
+            // Act
+            CancellationTokenSource cancellationTokenSource = new CancellationTokenSource(TimeSpan.FromSeconds(30));
+            transfer.WaitForCompletion(cancellationTokenSource.Token);
+
+            // Assert
+            testEventsRaised.AssertUnexpectedFailureCheck();
+            Assert.NotNull(transfer);
+            Assert.IsTrue(transfer.HasCompleted);
+            Assert.AreEqual(DataTransferState.Completed, transfer.TransferStatus.State);
+        }
+
+        [Test]
+        [LiveOnly] // https://github.com/Azure/azure-sdk-for-net/issues/33082
+        public async Task StartTransfer_EnsureCompleted_Failed()
+        {
+            // Arrange
+            await using IDisposingContainer<TSourceContainerClient> source = await GetSourceDisposingContainerAsync();
+            await using IDisposingContainer<TDestinationContainerClient> destination = await GetDestinationDisposingContainerAsync();
+
+            DataTransferOptions options = new DataTransferOptions()
+            {
+                CreationPreference = StorageResourceCreationPreference.FailIfExists
+            };
+            TestEventsRaised testEventsRaised = new TestEventsRaised(options);
+
+            // Create transfer to do a AwaitCompletion
+            DataTransfer transfer = await CreateStartTransfer(
+                source.Container,
+                destination.Container,
+                1,
+                createFailedCondition: true,
+                options: options);
+
+            // Act
+            CancellationTokenSource cancellationTokenSource = new CancellationTokenSource(TimeSpan.FromSeconds(30));
+            transfer.WaitForCompletion(cancellationTokenSource.Token);
+
+            // Assert
+            Assert.NotNull(transfer);
+            Assert.IsTrue(transfer.HasCompleted);
+            Assert.AreEqual(DataTransferState.Completed, transfer.TransferStatus.State);
+            Assert.AreEqual(true, transfer.TransferStatus.HasFailedItems);
+            await testEventsRaised.AssertContainerCompletedWithFailedCheck(1);
+            Assert.IsTrue(testEventsRaised.FailedEvents.First().Exception.Message.Contains(_expectedOverwriteExceptionMessage));
+        }
+
+        [Test]
+        [LiveOnly] // https://github.com/Azure/azure-sdk-for-net/issues/33082
+        public async Task StartTransfer_EnsureCompleted_Skipped()
+        {
+            // Arrange
+            await using IDisposingContainer<TSourceContainerClient> source = await GetSourceDisposingContainerAsync();
+            await using IDisposingContainer<TDestinationContainerClient> destination = await GetDestinationDisposingContainerAsync();
+
+            // Create transfer options with Skipping available
+            DataTransferOptions options = new DataTransferOptions()
+            {
+                CreationPreference = StorageResourceCreationPreference.SkipIfExists
+            };
+            TestEventsRaised testEventsRaised = new TestEventsRaised(options);
+
+            // Create transfer to do a EnsureCompleted
+            DataTransfer transfer = await CreateStartTransfer(
+                source.Container,
+                destination.Container,
+                1,
+                createFailedCondition: true,
+                options: options);
+
+            // Act
+            CancellationTokenSource cancellationTokenSource = new CancellationTokenSource(TimeSpan.FromSeconds(30));
+            transfer.WaitForCompletion(cancellationTokenSource.Token);
+
+            // Assert
+            testEventsRaised.AssertUnexpectedFailureCheck();
+            Assert.NotNull(transfer);
+            Assert.IsTrue(transfer.HasCompleted);
+            Assert.AreEqual(DataTransferState.Completed, transfer.TransferStatus.State);
+            Assert.AreEqual(true, transfer.TransferStatus.HasSkippedItems);
+        }
+
+        [Test]
+        [LiveOnly] // https://github.com/Azure/azure-sdk-for-net/issues/33082
+        public async Task StartTransfer_EnsureCompleted_Failed_SmallChunks()
+        {
+            // Arrange
+            await using IDisposingContainer<TSourceContainerClient> source = await GetSourceDisposingContainerAsync();
+            await using IDisposingContainer<TDestinationContainerClient> destination = await GetDestinationDisposingContainerAsync();
+
+            DataTransferOptions options = new DataTransferOptions()
+            {
+                CreationPreference = StorageResourceCreationPreference.FailIfExists,
+                InitialTransferSize = 512,
+                MaximumTransferChunkSize = 512
+            };
+            TestEventsRaised testEventsRaised = new TestEventsRaised(options);
+
+            // Create transfer to do a AwaitCompletion
+            DataTransfer transfer = await CreateStartTransfer(
+                source.Container,
+                destination.Container,
+                1,
+                createFailedCondition: true,
+                options: options,
+                size: Constants.KB * 4);
+
+            // Act
+            CancellationTokenSource cancellationTokenSource = new CancellationTokenSource(TimeSpan.FromSeconds(30));
+            transfer.WaitForCompletion(cancellationTokenSource.Token);
+
+            // Assert
+            Assert.NotNull(transfer);
+            Assert.IsTrue(transfer.HasCompleted);
+            Assert.AreEqual(DataTransferState.Completed, transfer.TransferStatus.State);
+            Assert.AreEqual(true, transfer.TransferStatus.HasFailedItems);
+            Assert.IsTrue(testEventsRaised.FailedEvents.First().Exception.Message.Contains(_expectedOverwriteExceptionMessage));
+            await testEventsRaised.AssertContainerCompletedWithFailedCheck(1);
+        }
+        #endregion
+    }
+}
diff --git a/sdk/storage/Azure.Storage.Files.Shares/src/ShareFileClient.cs b/sdk/storage/Azure.Storage.Files.Shares/src/ShareFileClient.cs
index fa9720fe9c94d..1d5c4cec38de1 100644
--- a/sdk/storage/Azure.Storage.Files.Shares/src/ShareFileClient.cs
+++ b/sdk/storage/Azure.Storage.Files.Shares/src/ShareFileClient.cs
@@ -523,7 +523,7 @@ protected static async Task<HttpAuthorization> GetCopyAuthorizationHeaderAsync(
             if (client.ClientConfiguration.TokenCredential != default)
             {
                 AccessToken accessToken = await client.ClientConfiguration.TokenCredential.GetTokenAsync(
-                    new TokenRequestContext(new string[] { client.ClientConfiguration.Audience.ToString() }),
+                    new TokenRequestContext(new string[] { client.ClientConfiguration.Audience.CreateDefaultScope() }),
                     cancellationToken).ConfigureAwait(false);
                 return new HttpAuthorization(
                     Constants.CopyHttpAuthorization.BearerScheme,