[Storage] [DataMovement] [Tests] Core Resume mocked tests (Azure#47431)

* initial commit

* change to internal

* Added checkpointer verify

* Refaactored out jobs and parts completion asserts

* Refactored and added more assertions for pause/resume during chunk processor - container

Author: nickliu-msft

sdk/storage/Azure.Storage.DataMovement/src/JobBuilder.cs

@@ -135,7 +135,10 @@ private async Task<TransferJobInternal> BuildSingleTransferJob(
             arrayPool: _arrayPool,
             clientDiagnostics: ClientDiagnostics);
 
-        if (resumeJob)
+        int jobPartCount = await checkpointer.GetCurrentJobPartCountAsync(
+                transferId: dataTransfer.Id,
+                cancellationToken: cancellationToken).ConfigureAwait(false);
+        if (resumeJob && jobPartCount > 0)
         {
             JobPartPlanHeader part = await checkpointer.GetJobPartAsync(dataTransfer.Id, partNumber: 0).ConfigureAwait(false);
             job.AppendJobPart(

sdk/storage/Azure.Storage.DataMovement/tests/MockStorageResourceProvider.cs

@@ -0,0 +1,58 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
+
+using System;
+using System.Threading;
+using System.Threading.Tasks;
+using Azure.Storage.Common;
+
+namespace Azure.Storage.DataMovement.Tests
+{
+    /// <summary>
+    /// Provider for a <see cref="StorageResource"/> configured for mocking.
+    /// </summary>
+    internal class MockStorageResourceProvider : StorageResourceProvider
+    {
+        /// <inheritdoc/>
+        protected internal override string ProviderId => "mock";
+
+        internal MemoryTransferCheckpointer checkpointer;
+
+        /// <summary>
+        /// Default constructor.
+        /// </summary>
+        internal MockStorageResourceProvider(
+            MemoryTransferCheckpointer checkpter)
+        {
+            checkpointer = checkpter;
+        }
+
+        /// <inheritdoc/>
+        protected internal override Task<StorageResource> FromSourceAsync(DataTransferProperties properties, CancellationToken cancellationToken)
+            => Task.FromResult(FromTransferProperties(properties, getSource: true));
+
+        /// <inheritdoc/>
+        protected internal override Task<StorageResource> FromDestinationAsync(DataTransferProperties properties, CancellationToken cancellationToken)
+            => Task.FromResult(FromTransferProperties(properties, getSource: false));
+
+        private StorageResource FromTransferProperties(DataTransferProperties properties, bool getSource)
+        {
+            Argument.AssertNotNull(properties, nameof(properties));
+
+            string transferId = properties.TransferId;
+            if (!checkpointer.Jobs.TryGetValue(transferId, out var job))
+            {
+                throw new Exception("Job does not exist.");
+            }
+
+            if (getSource)
+            {
+                return job.Source;
+            }
+            else
+            {
+                return job.Destination;
+            }
+        }
+    }
+}