Test_LargeDataUpload.cs

Go to the documentation of this file.

using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Diagnostics;
using System.Globalization;
using System.IO;
using System.Linq;
using System.Reflection;
using System.Text;
using System.Threading;
using System.Net;
 
using AnalyzeRe;
using AnalyzeRe.APITypes;
using AnalyzeRe.LargeDataUpload;
using AnalyzeRe.LossSets;
using AnalyzeReTesting.Base;
using AnalyzeReTesting.Utilities;
using AnalyzeRe.Rest;
 
#if MSTEST
using Microsoft.VisualStudio.TestTools.UnitTesting;
#elif NUNIT
using NUnit.Framework;
using TestClass = NUnit.Framework.TestFixtureAttribute;
using TestMethod = NUnit.Framework.TestAttribute;
using TestCategory = NUnit.Framework.CategoryAttribute;
#endif
 
namespace AnalyzeReTesting.Tests.AnalyzeReAPI
{
    [TestClass]
    public class TestSuite_LargeDataUpload : BaseServerTestSuite
    {
        private const string TypeName = "LargeDataUpload";
 
        #region String Test Properties
        private readonly string _sampleCatalogData = Samples.CSV.Event_Catalog_Data;
 
        private Stream SampleCatalogDataStream =>
            AnalyzeRe.LargeDataUpload.Utilities.GetStringStream(_sampleCatalogData);
        #endregion String Test Properties
 
        #region File Test Properties
        private const string RootSamplesCsvPath = @"..\..\Samples\csv\";
 
        private const string SampleCatalogEndPath = RootSamplesCsvPath + @"event_catalog_data.csv";
        private static readonly string SampleCatalogFilePath = GetTestFilePath(SampleCatalogEndPath);
 
        private static readonly string SampleCatalogFileUtf8;
        #endregion File Test Properties
 
        static TestSuite_LargeDataUpload()
        {
            try
            {
                SampleCatalogFileUtf8 = Encoding.UTF8.GetString(File.ReadAllBytes(SampleCatalogFilePath));
            }
            catch (Exception ex)
            {
                Console.WriteLine("The Test_LargeDataUpload.SampleCatalogFilePath references a " +
                    "location that does not exist in the current context. Tests that rely on " +
                    "this resource will not be able to run. (" + ex.Message + ")");
            }
        }
 
        #region Helper Methods
        private static string GetTestFilePath(string samplesPath)
        {
            char separator = Path.DirectorySeparatorChar;
            // Linux has '/' as a directory separator, Windows '\'
            samplesPath = samplesPath.Replace('\\', separator);
            string basePath = Path.GetDirectoryName(Assembly.GetExecutingAssembly().CodeBase);
            basePath = basePath != null ? Path.Combine(basePath, samplesPath) : samplesPath;
            // in Linux directory name has "file:" in the beginning, which cannot be parsed by Uri method
#if NUNIT
            return new Uri(basePath.Replace("#", "%23").Replace("file:","")).LocalPath;
#elif MSTEST
            return new Uri(basePath.Replace("#", "%23")).LocalPath;
#endif
        }
 
        private static void Test_EventCatalogDataUpload(Action<LargeDataSubResource> uploadAction) =>
            GenericTest.POST_ThenDoAction(Samples.EventCatalog.Unposted, posted => uploadAction(posted.data));
 
        private static void AssertContentsMatch(string original, SubResource<string> dataEndpoint)
        {
            string retrieved = null;
            AssertApi.MethodIsAllowed(() => retrieved = dataEndpoint.Get(), "Get Data");
            Assert.AreEqual(original, retrieved, "Downloaded data did not match uploaded data." +
                $"\nOriginal Contents:\n{original}\nRetrieved Contents:\n{retrieved}");
            Debug.WriteLine($"Asserted that the retrieved file contents matched the original:\n{retrieved}");
        }
 
        private static void PartiallyUploadCatalog(LargeDataSubResource dataEndpoint,
            string data, int patchBytes = 0)
        {
            dataEndpoint.Post(data.Length);
            if (patchBytes > 0)
                dataEndpoint.Patch(data.Substring(0, patchBytes), 0);
        }
        #endregion Helper Methods
 
        #region Simple String Tests
        // Test uploading a small string as a single segment.
        [TestMethod, TestCategory(TypeName)]
        public void Test_LargeDataUpload_OneSegmentString()
        {
            Test_EventCatalogDataUpload(delegate (LargeDataSubResource data)
            {
                data.LargeStreamUpload(SampleCatalogDataStream);
                AssertContentsMatch(_sampleCatalogData, data);
            });
        }
 
        // Test uploading a small string in multiple segments.
        [TestMethod, TestCategory(TypeName)]
        public void Test_LargeDataUpload_MultiSegmentString()
        {
            Test_EventCatalogDataUpload(delegate (LargeDataSubResource data)
            {
                data.LargeStreamUpload(SampleCatalogDataStream,
                    new Parameters(max_chunk_size: 50, enable_compression: false));
                AssertContentsMatch(_sampleCatalogData, data);
            });
        }
 
        // Test uploading a small string in multiple segments with compression
        [TestMethod, TestCategory(TypeName)]
        public void Test_LargeDataUpload_MultiSegmentStringCompressed()
        {
            Test_EventCatalogDataUpload(delegate (LargeDataSubResource data)
            {
                string expectedContents = new StreamReader(SampleCatalogDataStream).ReadToEnd();
                Debug.WriteLine($"Expected contents are:\n{expectedContents}" +
                    $"\nTotal Content Length: {expectedContents.Length}");
                data.LargeStreamUpload(SampleCatalogDataStream,
                    new Parameters(max_chunk_size: 10, enable_compression: true));
                AssertContentsMatch(_sampleCatalogData, data);
            });
        }
        #endregion Simple String Tests
 
        #region Reference Existing Files
        // Test setting the data to a reference to an existing file on POST
        [TestMethod, TestCategory(TypeName)]
        public void Test_LargeDataUpload_ReferenceExistingFile_OnPOST()
        {
            if (EnvironmentSettings.RUN_OFFLINE)
                Assert.Inconclusive("RUN_OFFLINE = true");
            EventCatalog other = Samples.EventCatalog.Posted;
            EventCatalog newCatalog = Samples.EventCatalog.Unposted;
            newCatalog.data_file = other.data_file;
            newCatalog = newCatalog.Post();
            newCatalog = newCatalog.PollUntilReady(Samples.DataPollingOptions);
            Assert.AreEqual(newCatalog.status, TaskStatus.Success);
            AssertContentsMatch(other.data.Get(), newCatalog.data);
        }
 
        // Test setting the data to a reference to an existing file on PUT
        [TestMethod, TestCategory(TypeName)]
        public void Test_LargeDataUpload_ReferenceExistingFile_OnPUT()
        {
            Skip.Indefinitely("ARE-3694 - Setting data reference on PUT does not kick off validation.");
            EventCatalog other = Samples.EventCatalog.Posted;
            GenericTest.POST_ThenDoAction(Samples.EventCatalog.Unposted, posted =>
            {
                Assert.IsNull(posted.data_file);
                posted.data_file = other.data_file;
                posted = posted.Put();
                posted = posted.PollUntilReady(Samples.DataPollingOptions);
                Assert.AreEqual(posted.status, TaskStatus.Success);
                AssertContentsMatch(other.data.Get(), posted.data);
            });
        }
        #endregion Reference Existing Files
 
        #region Simple File Tests
        // Test uploading a small file as a single segment.
        [TestMethod, TestCategory(TypeName)]
        public void Test_LargeDataUpload_OneSegmentFile()
        {
            Test_EventCatalogDataUpload(delegate (LargeDataSubResource data)
            {
                data.LargeFileUpload(SampleCatalogFilePath);
                AssertContentsMatch(SampleCatalogFileUtf8, data);
            });
        }
 
        // Test uploading a small file as a single segment.
        [TestMethod, TestCategory(TypeName)]
        public void Test_LargeDataUpload_MultiSegmentFile()
        {
            Test_EventCatalogDataUpload(delegate (LargeDataSubResource data)
            {
                data.LargeFileUpload(SampleCatalogFilePath,
                    new Parameters(max_chunk_size: 50, enable_compression: false));
                AssertContentsMatch(SampleCatalogFileUtf8, data);
            });
        }
 
        // Test the error when supplying a bad file path to the file upload
        [TestMethod, TestCategory(TypeName)]
        public void Test_LargeDataUpload_FilePathError()
        {
            Test_EventCatalogDataUpload(data => AssertApi.ExceptionThrown<FileNotFoundException>(
                () => data.LargeFileUpload("bad path")));
        }
 
        // Test being able to open a file in read-only mode if something else is using it.
        [TestMethod, TestCategory(TypeName)]
        public void Test_LargeDataUpload_FileOpenElsewhere()
        {
            // Open the file we plan to test with elsewhere
            using (FileStream busyFile = new FileStream(SampleCatalogFilePath,
                    FileMode.Open, FileAccess.ReadWrite, FileShare.Read))
            {
                // Try using the already opened file
                Test_EventCatalogDataUpload(delegate (LargeDataSubResource data)
                {
                    data.LargeFileUpload(SampleCatalogFilePath);
                    AssertContentsMatch(SampleCatalogFileUtf8, data);
                });
            }
        }
 
        // Test being able to open a file in read-only mode if something else is using it.
        [TestMethod, TestCategory(TypeName)]
        public void Test_LargeDataUpload_FileLockedExternally()
        {
            // Test this using a completely different file than usual, because
            // this test might run concurrently with other tests and prevent them
            // from opening the event catalog too.
            string fakeCatalogEndPath = RootSamplesCsvPath + @"event_catalog_data_fake.csv";
            string fakeCatalogFilePath = GetTestFilePath(fakeCatalogEndPath);
 
            using (FileStream temp = File.Create(fakeCatalogFilePath)) { }
 
            try
            {
                // Open the file we plan to test with elsewhere, but lock it
                using (FileStream busyFile = new FileStream(fakeCatalogFilePath,
                    FileMode.Open, FileAccess.ReadWrite, FileShare.None))
                {
                    // Using the locked file will fail. Nothing we can do about that.
                    Test_EventCatalogDataUpload(data => AssertApi.ExceptionThrown<IOException>(
                        () => data.LargeFileUpload(fakeCatalogFilePath)));
                }
            }
            finally
            {
                // Clean up after our new file.
                File.Delete(fakeCatalogFilePath);
            }
        }
        #endregion Simple File Tests
 
        #region Upload Conflict Tests
        #region With Compression
        // Test raising an error when trying to upload data when it already exists.
        [TestMethod, TestCategory(TypeName)]
        public void Test_LargeDataUpload_Compression_Conflict_Error()
        {
            Test_EventCatalogDataUpload(delegate (LargeDataSubResource data)
            {
                PartiallyUploadCatalog(data, _sampleCatalogData, _sampleCatalogData.Length / 2);
                AssertApi.ExceptionThrown<UploadInProgressException>(
                    () => data.LargeStreamUpload(SampleCatalogDataStream,
                        new Parameters(enable_compression: true, handle_existing_upload_strategy:
                            HandleUploadInSessionStrategy.RaiseError)),
                    ex => Assert.AreEqual("Could not upload to this endpoint, " +
                        "a large upload is already in progress. To attempt to resume or cancel " +
                        "the existing upload session, set the 'handle_existing_upload_strategy' " +
                        "parameter appropriately.", ex.Message));
            });
        }
 
        // Test setting the upload conflict strategy to delete what is there.
        [TestMethod, TestCategory(TypeName)]
        public void Test_LargeDataUpload_Compression_Conflict_CancelPrior()
        {
            Test_EventCatalogDataUpload(delegate (LargeDataSubResource data)
            {
                PartiallyUploadCatalog(data, _sampleCatalogData, _sampleCatalogData.Length / 2);
                data.LargeStreamUpload(SampleCatalogDataStream,
                    new Parameters(enable_compression: true, handle_existing_upload_strategy:
                        HandleUploadInSessionStrategy.CancelPriorUpload));
                AssertContentsMatch(_sampleCatalogData, data);
            });
        }
 
        // Test setting the upload conflict strategy to resume the upload
        [TestMethod, TestCategory(TypeName)]
        public void Test_LargeDataUpload_Compression_Conflict_Resume_PartiallyUploaded()
        {
            Test_EventCatalogDataUpload(delegate (LargeDataSubResource data)
            {
                PartiallyUploadCatalog(data, _sampleCatalogData, _sampleCatalogData.Length / 2);
                data.LargeStreamUpload(SampleCatalogDataStream,
                    new Parameters(enable_compression: true, handle_existing_upload_strategy:
                        HandleUploadInSessionStrategy.AttemptResume));
                AssertContentsMatch(_sampleCatalogData, data);
            });
        }
        
        // ARE-4982: Excel DTUV2 layer status can get stuck after upload is complete.
        // Error: The server reported the following error: Data has been uploaded and processed successfully.
        [TestMethod, TestCategory(TypeName)]
        public void Test_LargeDataUpload_Patch_MethodNotAllowed()
        {
            Test_EventCatalogDataUpload(delegate (LargeDataSubResource data)
            {
                // Upload some data
                PartiallyUploadCatalog(data, _sampleCatalogData, _sampleCatalogData.Length);
                data.LargeStreamUpload(SampleCatalogDataStream,
                    new Parameters(enable_compression: true, handle_existing_upload_strategy:
                        HandleUploadInSessionStrategy.AttemptResume));
                
                // Try to call Patch on the uploaded resource, confirm API returns MethodNotAllowed
                AssertApi.ExceptionThrown<APIRequestException>(
                    () => data.Patch("some_data", 1024),
                    e =>
                    {
                        // Get the status from the server, verify processing is complete
                        Assert.AreEqual(data.status.Get().status.IsProcessingComplete(), true);
                        // Verify API returned status code 405 (method not allowed)
                        Assert.AreEqual(e.RestResponse?.StatusCode, HttpStatusCode.MethodNotAllowed);
                        // Verify the error message returned by the API
                        Assert.AreEqual(e.ServerError.message, "Data has been uploaded and processed successfully.");
                    });
                
                // Call UploadChunk to verify it's handling the exception when calling Patch
                // Without the fix, this call would throw APIRequestException with HTTP status MethodNotAllowed
                AsyncTestState async_state = new AsyncTestState();
                object[] parameters = new object[] {
                    data,
                    Encoding.ASCII.GetBytes("some_data"),
                    1024,
                    28,
                    GetDefaultAsyncParameters(async_state) };
                typeof( LargeDataUpload_ExtensionMethods ).GetMethod("UploadChunk",
                     BindingFlags.NonPublic | BindingFlags.Static).Invoke(null, parameters);
            });
        }
 
        // Test a resumed upload where no data has been uploaded yet
        [TestMethod, TestCategory(TypeName)]
        public void Test_LargeDataUpload_Compression_Conflict_Resume_ZeroBytesUploaded()
        {
            Test_EventCatalogDataUpload(delegate (LargeDataSubResource data)
            {
                PartiallyUploadCatalog(data, _sampleCatalogData);
                data.LargeStreamUpload(SampleCatalogDataStream,
                    new Parameters(enable_compression: true, handle_existing_upload_strategy:
                        HandleUploadInSessionStrategy.AttemptResume));
                AssertContentsMatch(_sampleCatalogData, data);
            });
        }
 
        // Test a resumed upload where all that's left to do is commit.
        [TestMethod, TestCategory(TypeName)]
        public void Test_LargeDataUpload_Compression_Conflict_Resume_Uncommitted()
        {
            Test_EventCatalogDataUpload(delegate (LargeDataSubResource data)
            {
                PartiallyUploadCatalog(data, _sampleCatalogData, _sampleCatalogData.Length);
                data.LargeStreamUpload(SampleCatalogDataStream,
                    new Parameters(enable_compression: true, handle_existing_upload_strategy:
                        HandleUploadInSessionStrategy.AttemptResume));
                AssertContentsMatch(_sampleCatalogData, data);
            });
        }
 
        // Test raising an error when resuming upload where the files are of different sizes
        [TestMethod, TestCategory(TypeName)]
        public void Test_LargeDataUpload_Compression_Conflict_Resume_FileMismatch()
        {
            Test_EventCatalogDataUpload(delegate (LargeDataSubResource data)
            {
                PartiallyUploadCatalog(data, _sampleCatalogData.Substring(0, _sampleCatalogData.Length - 1));
                AssertApi.ExceptionThrown<UploadInProgressException>(
                    () => data.LargeStreamUpload(SampleCatalogDataStream,
                        new Parameters(enable_compression: true, handle_existing_upload_strategy:
                            HandleUploadInSessionStrategy.AttemptResume)),
                    ex => Assert.AreEqual(String.Format("Attempted to resume an existing upload session with a " +
                        "file of different size. Prior file size: {0} This file size: {1}",
                        _sampleCatalogData.Length - 1, _sampleCatalogData.Length), ex.Message));
            });
        }
        #endregion With Compression
 
        #region Without Compression
        // Test raising an error when trying to upload data when it already exists.
        [TestMethod, TestCategory(TypeName)]
        public void Test_LargeDataUpload_NoCompression_Conflict_Error()
        {
            Test_EventCatalogDataUpload(delegate (LargeDataSubResource data)
            {
                PartiallyUploadCatalog(data, _sampleCatalogData, _sampleCatalogData.Length / 2);
                AssertApi.ExceptionThrown<UploadInProgressException>(
                    () => data.LargeStreamUpload(SampleCatalogDataStream,
                        new Parameters(enable_compression: false,
                            handle_existing_upload_strategy: HandleUploadInSessionStrategy.RaiseError)),
                    ex => Assert.AreEqual("Could not upload to this endpoint, " +
                        "a large upload is already in progress. To attempt to resume or cancel " +
                        "the existing upload session, set the 'handle_existing_upload_strategy' " +
                        "parameter appropriately.", ex.Message));
            });
        }
 
        // Test setting the upload conflict strategy to delete what is there.
        [TestMethod, TestCategory(TypeName)]
        public void Test_LargeDataUpload_NoCompression_Conflict_CancelPrior()
        {
            Test_EventCatalogDataUpload(delegate (LargeDataSubResource data)
            {
                PartiallyUploadCatalog(data, _sampleCatalogData, _sampleCatalogData.Length / 2);
                data.LargeStreamUpload(SampleCatalogDataStream,
                    new Parameters(enable_compression: false,
                        handle_existing_upload_strategy: HandleUploadInSessionStrategy.CancelPriorUpload));
                AssertContentsMatch(_sampleCatalogData, data);
            });
        }
 
        // Test setting the upload conflict strategy to resume the upload
        [TestMethod, TestCategory(TypeName)]
        public void Test_LargeDataUpload_NoCompression_Conflict_Resume_PartiallyUploaded()
        {
            Test_EventCatalogDataUpload(delegate (LargeDataSubResource data)
            {
                PartiallyUploadCatalog(data, _sampleCatalogData, _sampleCatalogData.Length / 2);
                data.LargeStreamUpload(SampleCatalogDataStream,
                    new Parameters(enable_compression: false,
                        handle_existing_upload_strategy: HandleUploadInSessionStrategy.AttemptResume));
                AssertContentsMatch(_sampleCatalogData, data);
            });
        }
 
        // Test a resumed upload where no data has been uploaded yet
        [TestMethod, TestCategory(TypeName)]
        public void Test_LargeDataUpload_NoCompression_Conflict_Resume_ZeroBytesUploaded()
        {
            Test_EventCatalogDataUpload(delegate (LargeDataSubResource data)
            {
                PartiallyUploadCatalog(data, _sampleCatalogData);
                data.LargeStreamUpload(SampleCatalogDataStream,
                    new Parameters(enable_compression: false,
                        handle_existing_upload_strategy: HandleUploadInSessionStrategy.AttemptResume));
                AssertContentsMatch(_sampleCatalogData, data);
            });
        }
 
        // Test a resumed upload where all that's left to do is commit.
        [TestMethod, TestCategory(TypeName)]
        public void Test_LargeDataUpload_NoCompression_Conflict_Resume_Uncommitted()
        {
            Test_EventCatalogDataUpload(delegate (LargeDataSubResource data)
            {
                PartiallyUploadCatalog(data, _sampleCatalogData, _sampleCatalogData.Length);
                data.LargeStreamUpload(SampleCatalogDataStream,
                    new Parameters(enable_compression: false,
                        handle_existing_upload_strategy: HandleUploadInSessionStrategy.AttemptResume));
                AssertContentsMatch(_sampleCatalogData, data);
            });
        }
 
        // Test raising an error when resuming upload where the files are of different sizes
        [TestMethod, TestCategory(TypeName)]
        public void Test_LargeDataUpload_NoCompression_Conflict_Resume_FileMismatch()
        {
            Test_EventCatalogDataUpload(delegate (LargeDataSubResource data)
            {
                PartiallyUploadCatalog(data, _sampleCatalogData.Substring(0, _sampleCatalogData.Length - 1));
                AssertApi.ExceptionThrown<UploadInProgressException>(
                    () => data.LargeStreamUpload(SampleCatalogDataStream,
                        new Parameters(enable_compression: false,
                            handle_existing_upload_strategy: HandleUploadInSessionStrategy.AttemptResume)),
                    ex => Assert.AreEqual(String.Format("Attempted to resume an existing upload session with a " +
                        "file of different size. Prior file size: {0} This file size: {1}",
                        _sampleCatalogData.Length - 1, _sampleCatalogData.Length), ex.Message));
            });
        }
        #endregion Without Compression
 
        // Test an upload where a a file has already been uploaded and finalized.
        [TestMethod, TestCategory(TypeName)]
        public void Test_LargeDataUpload_Conflict_DataExists()
        {
            Test_EventCatalogDataUpload(delegate (LargeDataSubResource data)
            {
                data.LargeStreamUpload(SampleCatalogDataStream);
                Assert.AreEqual(TaskStatus.Success, data.status.Get().status);
                AssertApi.ExceptionThrown<InvalidOperationException>(
                    () => data.LargeStreamUpload(SampleCatalogDataStream),
                    ex => Assert.AreEqual("Cannot upload data for this resource because " +
                        "existing data has already been successfully uploaded.", ex.Message));
            });
        }
 
        // Test an upload where a a file has already been uploaded and finalized.
        [TestMethod, TestCategory(TypeName)]
        public void Test_LargeDataUpload_Conflict_Resume_DataExists()
        {
            Test_EventCatalogDataUpload(delegate (LargeDataSubResource data)
            {
                data.LargeStreamUpload(SampleCatalogDataStream);
                Assert.AreEqual(TaskStatus.Success, data.status.Get().status);
                // This should return successfully, even though the file is already
                // completely uploaded and finalized.
                data.LargeStreamUpload(SampleCatalogDataStream,
                    new Parameters(handle_existing_upload_strategy:
                        HandleUploadInSessionStrategy.AttemptResume));
            });
        }
        #endregion Upload Conflict Tests
 
        #region Asynchronous Upload
        #region Helper Methods
        public class AsyncTestState
        {
            #region Async Test Properties
            public ConcurrentQueue<StatusResponse> ProgressUpdates { get; set; }
 
            public Exception Error { get; set; }
 
            public StatusResponse Response { get; set; }
            #endregion Async Test Properties
 
            #region Async Test Delegates
            public Action<StatusResponse> OnProgress =>
                status => ProgressUpdates.Enqueue(status);
 
            public Action<Exception> OnError => ex => Error = ex;
 
            public Action<StatusResponse> OnSuccess =>
                result => Response = result;
            #endregion Async Test Delegates
 
            public AsyncTestState()
            {
                ProgressUpdates = new ConcurrentQueue<StatusResponse>();
                Error = null;
                Response = null;
            }
        }
 
        public AsyncParameters GetDefaultAsyncParameters(AsyncTestState async_state)
        {
            return new AsyncParameters(
                max_chunk_size: 50,
                // Disable compression when the chunk size is tiny
                enable_compression: false,
                commit_polling_options: Samples.DataPollingOptions,
                onSuccess: async_state.OnSuccess,
                onError: async_state.OnError,
                onProgressUpdate: async_state.OnProgress);
        }
 
        public void SleepUntilResultReady(AsyncTestState async_state, int max_sleep_seconds = 5)
        {
            API.PollUntil(() => async_state.Error != null || async_state.Response != null,
                new PollingOptions(
                    maxPollInterval: EnvironmentSettings.POLLING_INTERVAL,
                    maxPollTotalTime: max_sleep_seconds * 1000));
        }
 
        public void TestAsyncSuccess(AsyncTestState async_state)
        {
            // Verify that no error was received.
            Assert.IsNull(async_state.Error, "An Error was received: " +
                (async_state.Error?.ToString() ?? ""));
            // Verify that a successful response was received.
            Assert.IsNotNull(async_state.Response, "No API response was received.");
            // Verify that progress updates came through in order.
            Debug.Write("\nProgress Updates: ");
            long lastStatusBytes = -1;
            while (async_state.ProgressUpdates.TryDequeue(out StatusResponse status))
            {
                long uploaded = status.bytes_uploaded ?? -1;
                Debug.Write(uploaded + " (" + (uploaded * 100) / status.total_bytes + "%), ");
                Assert.IsTrue(uploaded >= lastStatusBytes, "Expected progress update: " +
                    uploaded + " to be >= last progress update: " + lastStatusBytes);
                // Assert that if there's no change in upload progress, we are done uploading
                // (i.e. that we are not sending redundant progress updates)
                if (uploaded == lastStatusBytes)
                    Assert.AreNotEqual(status.status, TaskStatus.Uploading);
                lastStatusBytes = uploaded;
            }
            Debug.WriteLine("");
        }
        #endregion Helper Methods
 
        #region Asynchronous Upload Tests
        // Test asynchronously uploading a small file in multiple segments.
        [TestMethod, TestCategory(TypeName)]
        public void Test_LargeDataUpload_Async_MultiSegmentFile()
        {
            Test_EventCatalogDataUpload(delegate (LargeDataSubResource data)
            {
                AsyncTestState async_state = new AsyncTestState();
                data.LargeFileUpload_Async(SampleCatalogFilePath,
                    GetDefaultAsyncParameters(async_state));
                SleepUntilResultReady(async_state);
                TestAsyncSuccess(async_state);
                AssertContentsMatch(SampleCatalogFileUtf8, data);
            });
        }
 
        // Test asynchronously uploading a small string in multiple segments.
        [TestMethod, TestCategory(TypeName)]
        public void Test_LargeDataUpload_Async_MultiSegmentString()
        {
            Test_EventCatalogDataUpload(delegate (LargeDataSubResource data)
            {
                AsyncTestState async_state = new AsyncTestState();
                data.LargeStreamUpload_Async(SampleCatalogDataStream,
                    GetDefaultAsyncParameters(async_state));
                SleepUntilResultReady(async_state);
                TestAsyncSuccess(async_state);
                AssertContentsMatch(_sampleCatalogData, data);
            });
        }
 
        // Test that Upload_Async can return in error via Callback
        [TestMethod, TestCategory(TypeName)]
        public void Test_LargeDataUpload_Async_MultiSegmentString_Error()
        {
            Test_EventCatalogDataUpload(delegate (LargeDataSubResource data)
            {
                PartiallyUploadCatalog(data, _sampleCatalogData);
                AsyncTestState async_state = new AsyncTestState();
                data.LargeStreamUpload_Async(SampleCatalogDataStream,
                    GetDefaultAsyncParameters(async_state));
                SleepUntilResultReady(async_state);
                // Assert that only an error response was received
                Assert.IsNull(async_state.Response);
                Assert.IsNotNull(async_state.Error);
                Assert.IsTrue(async_state.Error is UploadInProgressException);
            });
        }
 
        // Assert that if no onError callback is supplied, the upload fails silently.
        [TestMethod, TestCategory(TypeName)]
        public void Test_LargeDataUpload_Async_MultiSegmentString_Error_NoHandler()
        {
            Test_EventCatalogDataUpload(delegate (LargeDataSubResource data)
            {
                AsyncTestState async_state = new AsyncTestState();
                data.LargeFileUpload_Async("bad path");
            });
        }
        #endregion Asynchronous Upload Tests
        #endregion Asynchronous Upload
 
        #region Encoding Tests
        // Test uploading a files with different encodings
        [TestMethod, TestCategory(TypeName)]
        public void Test_LargeDataUpload_DifferentEncodings()
        {
            Test_LargeDataUpload_DifferentEncodings_Helper(new[]
            {
                RootSamplesCsvPath + @"event_catalog_encodings\event_catalog_data_ansi.csv",
                RootSamplesCsvPath + @"event_catalog_encodings\event_catalog_data_exceldefault.csv",
                RootSamplesCsvPath + @"event_catalog_encodings\event_catalog_data_notepadutf8.csv",
                RootSamplesCsvPath + @"event_catalog_encodings\event_catalog_data_win1252.csv"
            });
        }
 
        // Test uploading a files with different
        [TestMethod, TestCategory(TypeName), TestCategory("Skipped")]
        public void Test_LargeDataUpload_DifferentEncodings_Unicode()
        {
            // TODO: This test appears to work locally but not on Jenkins. It's possible that git
            // is interfering with the files when committing them or downloading them remotely.
            Skip.Indefinitely("(No associated ticket)");
            Test_LargeDataUpload_DifferentEncodings_Helper(new[]
            {
               RootSamplesCsvPath + @"event_catalog_encodings\event_catalog_data_unicode.csv",
               RootSamplesCsvPath + @"event_catalog_encodings\event_catalog_data_unicodebigendian.csv"
            });
        }
 
        // Test uploading a files with different encodings
        private void Test_LargeDataUpload_DifferentEncodings_Helper(IEnumerable<string> files)
        {
            if (EnvironmentSettings.RUN_OFFLINE)
                Assert.Inconclusive("RUN_OFFLINE = true");
            List<string> unsupported = new List<string>();
            foreach (string file in files)
            {
                string path = GetTestFilePath(file);
                try
                {
                    Test_EventCatalogDataUpload(data => data.LargeFileUpload(path));
                }
                catch (Exception ex)
                {
                    string encoding = path.Substring(path.LastIndexOf('\\'));
                    unsupported.Add(encoding);
                    Debug.WriteLine("\nUnsupported encoding: " + encoding +
                        "\nError: " + ex + "\n\n");
                }
            }
            if (unsupported.Count > 0)
            {
                Assert.AreEqual(0, unsupported.Count,
                    "One or more unsupported encodings encountered:\n" +
                    String.Join("\n", unsupported));
            }
        }
        #endregion Encoding Tests
 
        #region Binary Upload
        public DateTime LogTimeElapsed(DateTime startTime, string actionName = "Action")
        {
            DateTime now = DateTime.UtcNow;
            Console.WriteLine($"{actionName} took {(now - startTime).TotalSeconds} seconds.");
            return DateTime.UtcNow;
        }
 
        // Time uploading a large volume of data.
        [TestMethod, TestCategory(TypeName)]
        [Obsolete("ARE-5226 Remove support for Binary YELT Loss Set format. ")]
        public void Test_LargeDataUpload_Async_BinaryThroughput()
        {
            if (EnvironmentSettings.RUN_OFFLINE)
                Assert.Inconclusive("RUN_OFFLINE = true");
            const int events_per_trial = 125;
            const int rows = 500000;
 
            long bytes;
            double megabytes;
            TimeSpan elapsed = TimeSpan.MaxValue;
 
            // Post a contrived event catalog
            EventCatalog forYELT = new EventCatalog()
            {
                description = "Minimal Event Catalog",
                source = "Contrived"
            }.Post("\"Event Id\",\"Rate\"\n" + String.Join("\n",
                Enumerable.Range(1, events_per_trial).Select(i => i + "," + "1")));
            // Post the YELTLossSet Metadata
            YELTLossSet posted = new YELTLossSet
            {
                currency = "USD",
                description = "Sample Binary YELT",
                data_type = YELTLossSet.DataType.binary,
                event_catalogs = new List<IReference<EventCatalog>> { forYELT.ToReference() },
                trial_count = (int)Math.Ceiling((double)rows / events_per_trial)
            }.Post();
 
            // Prepare file upload parameters.
            AsyncTestState async_state = new AsyncTestState();
            AsyncParameters async_params =
                new AsyncParameters(
                    commit_polling_options: Samples.DataPollingOptions,
                    onSuccess: async_state.OnSuccess,
                    onError: async_state.OnError,
                    onProgressUpdate: async_state.OnProgress);
 
            DateTime checkpoint = DateTime.UtcNow;
            // Prepare a large YELT sample file.
            string temp_file_name = Path.GetTempFileName();
            Random r = new Random();
            try
            {
                using (StreamWriter writer = new StreamWriter(temp_file_name))
                {
                    foreach (int i in Enumerable.Range(0, rows).ToList())
                        writer.Write(
                            (i % events_per_trial) + "," + // Event
                            (i / events_per_trial + 1) + "," + // Trial
                            ((i % events_per_trial) + r.NextDouble()).ToString("0.##") + "," + // Sequence
                            (r.NextDouble() * 100000d).ToString("0.##") + "\n"); // Loss
                }
                DateTime start = checkpoint = LogTimeElapsed(checkpoint,
                    $"Generating YELT with {rows:#,###} rows.");
                // Upload the generated file
                posted.data.LargeFileUpload_Async(temp_file_name, async_params);
 
                // Monitor the upload, and notice when it changes status
                bool uploading = true;
                StatusResponse lastUpdate = null;
                while (async_state.Error == null && async_state.Response == null &&
                       (DateTime.UtcNow - start).TotalMinutes < 5.0)
                {
                    while (async_state.ProgressUpdates.Any())
                        async_state.ProgressUpdates.TryDequeue(out lastUpdate);
                    // Capture the switch from uploading to processing
                    if (uploading && lastUpdate != null &&
                        lastUpdate.status != TaskStatus.Uploading)
                    {
                        elapsed = DateTime.UtcNow - start;
                        checkpoint = LogTimeElapsed(checkpoint, "Uploading YELT");
                        uploading = false;
                    }
                    Thread.Sleep(50);
                }
                TimeSpan endToEndElapsed = DateTime.UtcNow - start;
                LogTimeElapsed(checkpoint, "Processing YELT on the server");
                if (elapsed > endToEndElapsed) elapsed = endToEndElapsed;
 
                while (async_state.ProgressUpdates.Any())
                    async_state.ProgressUpdates.TryDequeue(out lastUpdate);
                // Get the upload speed based on the size of the compressed file.
                bytes = lastUpdate?.total_bytes ?? 0;
                megabytes = bytes / 1048576d;
                Console.WriteLine($"\nCompressed File size:  {bytes} bytes, {megabytes:0.00}MiB");
                Console.WriteLine($"Upload Speed: {(int)(bytes / elapsed.TotalSeconds)} bytes/s " +
                    $"({(megabytes / elapsed.TotalSeconds):0.00} MiBps)");
 
                // Calculate file size
                bytes = new FileInfo(temp_file_name).Length;
                megabytes = bytes / 1048576d;
                Console.WriteLine($"\nOriginal File size:  {bytes} bytes, {megabytes:0.00}MiB");
 
                // Log the throughput for this upload
                Console.WriteLine($"Effective Throughput: {(int)(bytes / elapsed.TotalSeconds)} " +
                    $"bytes/s ({(megabytes / elapsed.TotalSeconds):0.00} MiBps)\n");
 
                // Get the throughput counting the server validation time.
                Console.WriteLine($"Throughput including server time: " +
                    $"{(int)(bytes / endToEndElapsed.TotalSeconds)} bytes/s " +
                    $"({(megabytes / endToEndElapsed.TotalSeconds):0.00} MiBps)\n");
 
                Assert.IsTrue(async_state.Error == null && !(lastUpdate?.status.IsError() ?? false),
                    $"Async upload appears to have failed: {lastUpdate?.status} - {async_state.Error}");
                Assert.IsNull(async_state.Error,
                    $"Async upload appears to have never failed: {async_state.Error}");
                Assert.AreEqual(TaskStatus.Success, lastUpdate?.status,
                    $"Async upload appears to have never completed: {lastUpdate?.status}");
            }
            finally
            {
                try { File.Delete(temp_file_name); } catch { /* Ignore if cleanup fails. */ }
            }
 
            // Now time the upload of the tiniest file possible against a copy of the resource
            YELTLossSet copy = posted.ShallowCopy().Post();
            async_state = new AsyncTestState();
            async_params = new AsyncParameters(
                commit_polling_options: Samples.DataPollingOptions,
                onSuccess: async_state.OnSuccess,
                onError: async_state.OnError,
                onProgressUpdate: async_state.OnProgress);
 
            checkpoint = DateTime.UtcNow;
            copy.data.LargeStreamUpload_Async(
                AnalyzeRe.LargeDataUpload.Utilities.GetStringStream("1,1,1,1"), async_params);
            SleepUntilResultReady(async_state, 20);
            TimeSpan overhead = DateTime.UtcNow - checkpoint;
            Console.WriteLine($"- {overhead.TotalSeconds:0.0000} " +
                "(Overhead for any upload determined by uploading a teeny tiny file)");
            elapsed -= overhead;
            Console.WriteLine($"= {elapsed.TotalSeconds:0.0000} Adjusted Upload-Only Time");
 
            // Display the adjusted throughput.
            Console.WriteLine($"\nAdjusted Throughput: {(int)(bytes / elapsed.TotalSeconds)} " +
                $"bytes/s ({(megabytes / elapsed.TotalSeconds):0.00} MiBps)");
            Console.WriteLine("Adjusted Throughput is an optimistic measure of the C# " +
                "client's capacity to push volumes of data to the server by ignoring " +
                "fixed overhead and subsequent validation performance of the server.\n");
        }
 
        [TestMethod, TestCategory(TypeName)]
        [Obsolete("ARE-5226 Remove support for Binary YELT Loss Set format. ")]
        public void Test_LargeStreamUpload_AlreadyBinaryYELTFile()
        {
            if (EnvironmentSettings.RUN_OFFLINE)
                Assert.Inconclusive("RUN_OFFLINE = true");
            // Post the YELTLossSet Metadata
            YELTLossSet binaryYelt = Samples.LossSet_YELTLossSet.Unposted
                .Change(ls => ls.data_type, YELTLossSet.DataType.binary).Post();
 
            // Create a stream containing a YELT already in the binary format
            using (MemoryStream ms = new MemoryStream())
            {
                using (BufferedBytesFromStreamProducer bufferProducer = new BufferedBytesFromStreamProducer(
                        new YELTStream(binaryYelt.trial_count, 10, false, new[] { 0, 1, 2 })))
                using (BinaryYELTConverter binaryConverter = new BinaryYELTConverter(bufferProducer))
                {
                    WriteAllProducerConsumerBytesToMemoryStream(binaryConverter, ms);
                }
                // Ensure binary conversion is disabled, but compression is enabled.
                Parameters uploadParams = new Parameters(commit_polling_options: Samples.DataPollingOptions)
                {
                    binary_yelt_options = Parameters.BinaryYELTUploadOptions.AutomaticCompressionOnly
                };
                binaryYelt.data.LargeStreamUpload(ms, uploadParams);
            }
 
            // Assert that processing succeeded
            StatusResponse status = binaryYelt.data.status.Get();
            Assert.IsTrue(status.commit_error == null,
                $"Upload appears to have failed: {status.commit_error}");
            binaryYelt = binaryYelt.Get();
            Assert.AreEqual(TaskStatus.Success, binaryYelt.status, "Async upload appears to have " +
                $"not completed successfully: {binaryYelt.status} - {binaryYelt.status_message}");
        }
 
        [TestMethod, TestCategory(TypeName)]
        [Obsolete("ARE-5226 Remove support for Binary YELT Loss Set format. ")]
        public void Test_LargeStreamUpload_AlreadyBinaryZippedYELTFile()
        {
            if (EnvironmentSettings.RUN_OFFLINE)
                Assert.Inconclusive("RUN_OFFLINE = true");
            // Post the YELTLossSet Metadata
            YELTLossSet binaryYelt = Samples.LossSet_YELTLossSet.Unposted
                .Change(ls => ls.data_type, YELTLossSet.DataType.binary).Post();
 
            // Create a stream containing a YELT already in the binary format
            using (MemoryStream ms = new MemoryStream())
            {
                using (BufferedBytesFromStreamProducer bufferProducer = new BufferedBytesFromStreamProducer(
                        new YELTStream(binaryYelt.trial_count, 10, false, new[] { 0, 1, 2 })))
                using (BinaryYELTConverter binaryConverter = new BinaryYELTConverter(bufferProducer))
                using (GZipBufferedBytesProducer compressor = new GZipBufferedBytesProducer(binaryConverter,
                    compressionStrategy: GZipBufferedBytesProducer.CompressionStrategy.CompressEntireFile))
                {
                    WriteAllProducerConsumerBytesToMemoryStream(compressor, ms);
                }
                // Ensure binary conversion and compression are disabled in the parameters.
                Parameters uploadParams = new Parameters(commit_polling_options: Samples.DataPollingOptions)
                {
                    binary_yelt_options = Parameters.BinaryYELTUploadOptions.None
                };
                binaryYelt.data.LargeStreamUpload(ms, uploadParams);
            }
 
            // Assert that processing succeeded
            StatusResponse status = binaryYelt.data.status.Get();
            Assert.IsTrue(status.commit_error == null,
                $"Upload appears to have failed: {status.commit_error}");
            binaryYelt = binaryYelt.Get();
            Assert.AreEqual(TaskStatus.Success, binaryYelt.status, "Async upload appears to have " +
                $"not completed successfully: {binaryYelt.status} - {binaryYelt.status_message}");
        }
 
        private void WriteAllProducerConsumerBytesToMemoryStream(ProducerConsumerBuffer<BufferedBytes> producer, MemoryStream ms)
        {
            const int maxFailedReads = 5;
            int failedReads = 0;
            producer.Start();
            while (producer.CanTake && failedReads < maxFailedReads)
            {
                BufferedBytes binaryYeltBytes = producer.TryTake(out bool success, null);
                if (!success)
                    failedReads++;
                else
                {
                    failedReads = 0;
                    ms.Write(binaryYeltBytes.Bytes, 0, binaryYeltBytes.LengthFilled);
                }
            }
            Assert.IsTrue(failedReads < maxFailedReads, "The compressor failed to generate any bytes " +
                "following 5 reads in a row - yet CanTake is still true. Infinite loop avoided.");
            // Reset the memory stream position to 0 so it can be consumed by the caller
            ms.Position = 0;
        }
 
        [TestMethod, TestCategory(TypeName)]
        [Obsolete("ARE-5226 Remove support for Binary YELT Loss Set format. ")]
        public void Test_LargeStreamUpload_ReUploadBinaryYELTFile()
        {
            if (EnvironmentSettings.RUN_OFFLINE)
                Assert.Inconclusive("RUN_OFFLINE = true");
            // 1. Post the a binary YELTLossSet
            YELTLossSet binaryYelt = Samples.LossSet_YELTLossSet.Unposted
                .Change(ls => ls.data_type, YELTLossSet.DataType.binary)
                .Post(Samples.CSV.YELTLossSet_10Trials_ForBinary).PollUntilReady(Samples.DataPollingOptions);
            Assert.AreEqual(TaskStatus.Success, binaryYelt.status, binaryYelt.status_message);
 
            // 2. Server should allow us to create a copy of this pretty easily by re-using the same data file
            YELTLossSet copy = binaryYelt.DeepCopy().Post().PollUntilReady(Samples.DataPollingOptions);
            Assert.AreEqual(TaskStatus.Success, copy.status, copy.status_message);
 
            // We should also be able to download the converted binary file and re-use it that way.
            // Ensure binary conversion and compression are disabled in the parameters.
            Parameters uploadParams = new Parameters(commit_polling_options: Samples.DataPollingOptions)
            {
                binary_yelt_options = Parameters.BinaryYELTUploadOptions.None
            };
 
            // 3. Download the data from the server as stream of bytes, then re-upload
            copy = binaryYelt.DeepCopy().Change(r => r.data_file, null).Post();
            using (MemoryStream ms = new MemoryStream())
            {
                byte[] buffer = new byte[4096];
                binaryYelt.data.GetStream(downloadStream =>
                {
                    int count;
                    while ((count = downloadStream.Read(buffer, 0, buffer.Length)) != 0)
                        ms.Write(buffer, 0, count);
                });
                // Reset the memory stream position to 0 so it can be consumed by the uploader
                ms.Position = 0;
                copy.data.LargeStreamUpload(ms, uploadParams);
            }
            copy = copy.Get();
            Assert.AreEqual(TaskStatus.Success, copy.status, copy.status_message);
 
            // 4. We should also be able to cleverly stream a download a new upload directly.
            copy = binaryYelt.DeepCopy().Change(r => r.data_file, null).Post();
            binaryYelt.data.GetStream(
                downloadStream => copy.data.LargeStreamUpload(downloadStream, uploadParams));
            copy = copy.Get();
            Assert.AreEqual(TaskStatus.Success, copy.status, copy.status_message);
 
            // 5? Note: We *cannot* read the data into a string to re-upload it
            //          The binary data is corrupted by the UTF-8 Encoding process.
            //string binaryData = binaryYelt.data.Get();
            //copy = binaryYelt.DeepCopy().Change(r => r.data_file, null)
            //    .Post(binaryData, null, uploadParams).PollUntilReady();
            //Assert.AreEqual(TaskStatus.Success, copy.status, copy.status_message);
        }
 
        [TestMethod, TestCategory(TypeName)]
        [Obsolete("ARE-5226 Remove support for Binary YELT Loss Set format. ")]
        public void Test_LargeDataUpload_Binary_ErrorHandling()
        {
            if (EnvironmentSettings.RUN_OFFLINE)
                Assert.Inconclusive("RUN_OFFLINE = true");
            YELTLossSet yelt = Samples.LossSet_YELTLossSet.Unposted;
            yelt.data_type = YELTLossSet.DataType.binary;
            AssertApi.ExceptionThrown<BufferProducerException>(() =>
                yelt.Post("1,0,1.5,542345.543\n" +
                    "2,98765,-123.456,-9999.9999\n" +
                    "1,0,1.5,1e4"));
        }
        #endregion Binary Upload
 
        #region Stream of indeterminate Length
        private class UnknownLengthStream : Stream
        {
            private const int BytesToReturn = 1000000;
            private int _bytesReturned = 0;
 
            public override bool CanRead => _bytesReturned < BytesToReturn;
            public override bool CanSeek => false;
            public override bool CanWrite => false;
            public override long Length => throw new NotSupportedException();
            public override long Position
            {
                get => throw new NotSupportedException();
                set => throw new NotSupportedException();
            }
 
            public override void Flush() { }
            public override long Seek(long offset, SeekOrigin origin) => throw new NotSupportedException();
            public override void SetLength(long value) => throw new NotSupportedException();
            public override void Write(byte[] buffer, int offset, int count) => throw new NotSupportedException();
            public override int Read(byte[] buffer, int offset, int count)
            {
                int written = 0;
                // Due to ARE-7230, we will generate an internal server error unless we include
                // line breaks in our data periodically - so we add char(10) every 1000 bytes.
                for (int i = offset; i < offset + count; i++)
                    buffer[i] = i % 1000 == 0 ? (byte)(char)10 : (byte)(char)(44 + (i ^ 13) % (58 - 48));
                if (_bytesReturned < BytesToReturn)
                    written = Math.Min(count, BytesToReturn - _bytesReturned);
                _bytesReturned += written;
                return written;
            }
        }
 
        [TestMethod, TestCategory(TypeName)]
        public void Test_LargeDataUpload_Stream_Unknown_Length()
        {
            if (EnvironmentSettings.RUN_OFFLINE)
                Assert.Inconclusive("RUN_OFFLINE = true");
            EventCatalog incomplete = Samples.EventCatalog.Unposted.Post();
            AssertApi.ExceptionThrown<CommitFailedException>(() =>
            {
                using (Stream unknownStream = new UnknownLengthStream())
                {
                    incomplete.data.LargeStreamUpload(unknownStream);
                }
            }, ex => Assert.AreEqual("The uploaded file failed validation " +
                "with the following message: CSV input error: Event ID column not found.", ex.Message));
        }
 
        // A sample stream that produces a YELT CSV stream on unknown length on the fly.
        public class YELTStream : Stream
        {
            private readonly int _eventsPerTrial;
            private readonly int _trialCount;
            private readonly Random _rng = new Random();
            private byte[] _leftOver;
            private int _rowsGenerated;
            private long _bytesReturned;
            private readonly bool _includeHeader;
            private readonly int[] _validEventIds;
 
            public bool HeaderGenerated { get; private set; } = false;
            public int RowsGenerated => _rowsGenerated;
            public long BytesReturned => _bytesReturned;
 
            public override bool CanRead => RowsGenerated < _eventsPerTrial * _trialCount || _leftOver != null;
            public override bool CanSeek => false;
            public override bool CanWrite => false;
            public override long Length => throw new NotSupportedException();
 
            public override long Position
            {
                get => throw new NotSupportedException();
                set => throw new NotSupportedException();
            }
 
            public YELTStream(int trial_count, int events_per_trial, bool includeHeader = false,
                IEnumerable<int> validEventIds = null)
            {
                _trialCount = trial_count;
                _eventsPerTrial = events_per_trial;
                _includeHeader = includeHeader;
                _validEventIds = validEventIds?.ToArray();
            }
 
            public override void Flush() { }
            public override void Write(byte[] buffer, int offset, int count) => throw new NotSupportedException();
            public override long Seek(long offset, SeekOrigin origin) => throw new NotSupportedException();
            public override void SetLength(long value) => throw new NotSupportedException();
            public override int Read(byte[] buffer, int offset, int count)
            {
                if (!CanRead)
                    return 0;
 
                const string yeltFormat = "{0},{1},{2:f2},{3:f2}\n";
                int bytesPrepared = 0;
                int currentOffset = offset;
                CultureInfo culture = CultureInfo.InvariantCulture;
 
                if (_leftOver != null)
                { 
                    int to_copy = Math.Min(count, _leftOver.Length);
                    Buffer.BlockCopy(_leftOver, 0, buffer, offset, to_copy);
                    // If we didn't get to use all leftovers, return now
                    if (count < _leftOver.Length)
                    {
                        byte[] leftOverleft = new byte[_leftOver.Length - count];
                        Buffer.BlockCopy(_leftOver, count, leftOverleft, 0, leftOverleft.Length);
                        _leftOver = leftOverleft;
                        _bytesReturned += count;
                        return count;
                    }
                    bytesPrepared += to_copy;
                    currentOffset += bytesPrepared;
                }
                _leftOver = null;
 
                // Keep generating rows until we've reached the number of requested bytes or
                // we've reached the number of rows we've been configured to generate.
                while (bytesPrepared < count && RowsGenerated < _eventsPerTrial * _trialCount)
                {
                    // Produce a random YELT row
                    string nextRow;
                    if (RowsGenerated == 0 && _includeHeader && !HeaderGenerated)
                    {
                        nextRow = "Event Id,Trial Id,Day,Loss\n";
                        HeaderGenerated = true;
                    }
                    else
                    {
                        int event_id = _validEventIds?[_rng.Next(_validEventIds.Length)] ??
                            _rng.Next(_eventsPerTrial);
                        nextRow = String.Format(culture, yeltFormat,
                            event_id,
                            RowsGenerated / _eventsPerTrial + 1 /* Trial */,
                            RowsGenerated % _eventsPerTrial + _rng.NextDouble() /* Day */,
                            _rng.NextDouble() * 100000d /* Loss */);
                    }
                    // Increment the row counter and total bytes counter
                    // Get the byte encoding for this row
                    byte[] bytes = Encoding.UTF8.GetBytes(nextRow);
                    _rowsGenerated++;
                    int byte_count = bytes.Length;
                    int buffer_room = count - bytesPrepared;
                    // Copy as many bytes as we have or as many as will fit into the current buffer
                    int bytes_copied = buffer_room < byte_count ? buffer_room : byte_count;
                    Buffer.BlockCopy(bytes, 0, buffer, currentOffset, bytes_copied);
                    bytesPrepared += bytes_copied;
                    currentOffset += bytes_copied;
                    // If the whole row didn't fit into the buffer, save what's left for later.
                    if (bytes_copied < byte_count)
                    {
                        _leftOver = new byte[byte_count - bytes_copied];
                        Buffer.BlockCopy(bytes, bytes_copied, _leftOver, 0, _leftOver.Length);
                    }
                }
                _bytesReturned += bytesPrepared;
                // Return the number of bytes we just produced
                return bytesPrepared;
            }
        }
 
        [TestMethod, TestCategory(TypeName)]
        [Obsolete("ARE-5226 Remove support for Binary YELT Loss Set format. ")]
        public void Test_LargeDataUpload_Binary_YELT_Stream_Unknown_Length()
        {
            if (EnvironmentSettings.RUN_OFFLINE)
                Assert.Inconclusive("RUN_OFFLINE = true");
            const int trialCount = 50000;
            YELTLossSet incomplete = Samples.LossSet_YELTLossSet.Unposted;
            incomplete.data_type = YELTLossSet.DataType.binary;
            incomplete.trial_count = trialCount;
            incomplete = incomplete.Post();
            using (YELTStream yeltStream = new YELTStream(trialCount, 4))
            {
                DateTime start = DateTime.UtcNow;
                incomplete.data.LargeStreamUpload(yeltStream);
                double elapsedSeconds = (DateTime.UtcNow - start).TotalSeconds;
                Console.WriteLine($"Upload of {yeltStream.RowsGenerated} row YELT " +
                    $"took {elapsedSeconds:0.0} seconds " +
                    $"({(yeltStream.RowsGenerated / elapsedSeconds):0} rows per second).");
                Console.WriteLine($"Total bytes (uncompressed): {yeltStream.BytesReturned}.");
                Console.WriteLine("Effective Throughput: " +
                    $"{(yeltStream.BytesReturned / 1.049e+6 / elapsedSeconds):0.00} MiB per second.");
            }
        }
        #endregion Stream of indeterminate Length
 
        #region StreamProducers
        [TestMethod, TestCategory(TypeName)]
        public void Test_BufferedBytesFromStreamProducer_BytesToReadRespected()
        {
            const int timeoutMs = 20000; // ms
            using (YELTStream data = new YELTStream(10, 10))
            {
                int bytes_to_read = 666;
                BufferedBytesFromStreamProducer producer =
                    new BufferedBytesFromStreamProducer(data, bytes_to_read, 7, 1234);
                bool success = true;
                int total_buffered = 0;
                producer.Start();
                while (producer.CanTake && success)
                {
                    try
                    {
                        CancellationTokenSource timeout = new CancellationTokenSource(timeoutMs);
                        total_buffered += producer.TryTake(out success, timeout.Token).LengthFilled;
                        if (total_buffered > bytes_to_read || data.BytesReturned > bytes_to_read)
                            Assert.Fail(total_buffered + " bytes of been buffered, and " +
                                data.BytesReturned + " bytes requested from the stream, which " +
                                "exceeds the set number of bytes to read of " + bytes_to_read);
                    }
                    catch (OperationCanceledException)
                    {
                        Assert.Fail("The stream took longer than " + timeoutMs + "ms to respond.");
                    }
                }
                Assert.AreEqual(bytes_to_read, total_buffered, "Expected only bytes to read to " +
                    "be reflected in the total number of bytes returned.");
                Assert.AreEqual(bytes_to_read, data.BytesReturned, "Expected only bytes to read to " +
                    "be reflected in the total number of bytes requested from the stream.");
            }
        }
        #endregion StreamProducers
    }
}