--[[----------------------------------------------------------------------------

PixelProviderDownloadToFileTaskTest.lua

--------------------------------------------------------------------------------

ADOBE CONFIDENTIAL
------------------
Copyright 2015 Adobe Systems Incorporated
All Rights Reserved.

NOTICE: All information contained herein is, and remain the property of
Adobe Systems Incorporated and its suppliers, if any. The intellectual and
technical concepts contained herein are proprietary to Adobe Systems
Incorporated and its suppliers and may be covered by U.S. and Foreign
Patents, patents in process, and are protected by trade secret or copyright
law. Dissemination of this information or reproduction of this material is
strictly forbidden unless prior written permission is obtained from
Adobe Systems Incorporated.

------------------------------------------------------------------------------]]

	-- If ACR is not supported, disable test before we attempt to
	-- require WFAcr. (That will fail.)
if not require 'WFPlatformInfo'.hasACR then return {} end

	-- wf.core
local WFCloseSupport = require 'WFCloseSupport'
local KSIdGenerator = require 'KSIdGenerator'
local KSTableUtils = require 'KSTableUtils'
local WFCatMockCatalog = require 'WFCatMockCatalog'
local WFFunctionUtils = require 'WFFunctionUtils'
local WFRxTestCase = require 'WFRxTestCase'

	-- Rx
local Rx = require 'Rx'

--------------------------------------------------------------------------------

local PixelPathProviderScheduler = require 'PixelPathProviderScheduler'
local PixelPathProviderRequestTypes = require 'PixelPathProviderRequestTypes'
local PixelPathProviderStateMachine = require 'PixelPathProviderStateMachine'
local PixelProviderDownloadToFileTask = require 'PixelProviderDownloadToFileTask'
local PixelProviderDownloadToFileUtils = require 'PixelProviderDownloadToFileUtils'

--------------------------------------------------------------------------------

local PixelPathProviderRequestTypeNames = PixelPathProviderRequestTypes.names

--------------------------------------------------------------------------------

local function pixStreamId( i, maybePrefix )

	return string.format( "%s%d", maybePrefix or "pixStreamId", i )

end

--------------------------------------------------------------------------------

local function localAssetId( i )

	return i

end

--------------------------------------------------------------------------------

local function bootstrapScheduler( catalog )

	local scheduler = PixelPathProviderScheduler( catalog )

	local SchedulerTestAPI = scheduler:_expose_test_API()

	SchedulerTestAPI.addRunLoopTaskToAllowlist( 
		scheduler, "PixelProviderDownloadToFileTask" )

	-- adding and removing a dummy request makes sure that all tasks are
	-- bootstrapped
	scheduler:addRequest( pixStreamId( 1 ), localAssetId( 1 ),
							PixelPathProviderRequestTypeNames.loupe_next, 1 )

	SchedulerTestAPI.runScheduledRequests( scheduler )

	scheduler:abandonAllLocalAssetIdRequests( localAssetId ( 1 ) )

	return scheduler

end

--------------------------------------------------------------------------------

local function makeSchedulerForTest( rxTest )

	local catalog = WFCatMockCatalog.createFakeWFCatalog( rxTest )

	local scheduler = bootstrapScheduler( catalog )

	rxTest:finally( 
		function() 

			scheduler:close() 

		end )

	return scheduler, catalog

end

--------------------------------------------------------------------------------

local function digest( i )

	return string.format( "digest%s", i )

end

--------------------------------------------------------------------------------

local _ozAssetIds = {}

local function makeOzAssetId( i )

	_ozAssetIds[ i ] = _ozAssetIds[ i ] or KSIdGenerator.uuidInLowercaseHex()

	return _ozAssetIds[ i ]

end

--------------------------------------------------------------------------------

local function addRecordToDocStore( rxTest, catalog, i )

	rxTest:expectSingleValue(

		catalog:docStore():bulkSetDocValue {
			{
				_id = makeOzAssetId( i ),
				_localId = localAssetId( i ),
				subtype = "image",
				develop = {
					croppedHeight = 100 * i,
					croppedWidth = 100 * i,
					xmpCameraRaw = '<xmp />',
					userOrientation = i % 4,
				},
				rating = i % 5,
			},
		},

		':ignore:' )

end

--------------------------------------------------------------------------------

local function addBaseRenderInfoCacheEntry( scheduler, i )

	scheduler._testData.runLoopCaches.sharedCaches.baseRenderInfoCache
		:pushCacheItem( 
			localAssetId( i ), 
			{ 
				ozAssetId = makeOzAssetId( i ), 
				changeCounter = i, 
				revId = digest( i ),
				digest = digest( i ), 
				proxies = { { size="2560", href="notAProxy" } },
				ozValidOriginalLink = "yeahNope",
				ozXmpDevelopLink = 'uh_huh_NOPE',
				ozXmpLink = 'justNope',
			} )

end

--------------------------------------------------------------------------------

local function stripTestMessageAndMakeTuple( _, ... )

	return { ... }

end

--------------------------------------------------------------------------------

local function sortArray( arr, maybeSortFn )

	local sorted = KSTableUtils.deepCopy( arr )

	table.sort( sorted, maybeSortFn )

	return sorted

end

--------------------------------------------------------------------------------

local function compareArrays( a1, a2 )

	if #a1 ~= #a2 then

		return #a1 > #a2

	end

	for i = 1, math.min( #a1, #a2 ) do

		if a1[ i ] ~= a2[ i ] then

			return a1[ i ] > a2 [i ]

		end

	end

	return false

end

--------------------------------------------------------------------------------

local function sortTupleArray( tupleArray )

	return sortArray( tupleArray, compareArrays )

end

--------------------------------------------------------------------------------

local function isTestMessage( ... )

	local wantedTestMessages = KSTableUtils.mapArrayToSet( { ... } )

	return function( testMsg )

		return wantedTestMessages[ testMsg ]

	end

end

--------------------------------------------------------------------------------

local function takeFirstN( n )

	return function( t )

		local ret = {}

		for i = 1, n do

			ret[ i ] = t[ i ]

		end

		return ret

	end

end

--------------------------------------------------------------------------------

local function makeMockBinaryRequester( scheduler, getBinaryRequestNumFunc )

	local mockDownloadResultStream = Rx.Subject()

	local mockDownloadRequestStream = Rx.Subject()

	if type( getBinaryRequestNumFunc ) ~= 'function' then

		getBinaryRequestNumFunc = 

			function( self )
				
				return #self._binaryRequests + 1

			end 

	end

	local mockBinaryRequester = {

		_binaryRequests = {},

		getBinaryRequestToFileStream =
			function() 

				return mockDownloadResultStream

			end,
		
		maybeRequestBinaryAtUrlToFile =
			function( self, destPath, ozAssetId, ozBinaryType, ozBinarySubtype, ozBinaryUrl, priority ) 

				local i = getBinaryRequestNumFunc( self )

				local fakeDownloadSubscr

				local function cancelFn()

					if fakeDownloadSubscr then

						fakeDownloadSubscr:dispose()

						fakeDownloadSubscr = nil

					end

					if not WFCloseSupport.hasBeenOrWillBeClosed( scheduler ) then

						mockDownloadResultStream
							:onNext( 
								false, 
								"cancelFn called",
								ozAssetId,
								ozBinaryType,
								ozBinarySubtype )

					end

					self._binaryRequests[ i ] = nil

					return true

				end

				self._binaryRequests[ i ] =
					{
						destPath = destPath,
						ozAssetId = ozAssetId,
						ozBinaryType = ozBinaryType,
						ozBinarySize = ozBinarySubtype,
						ozBinaryUrl = ozBinaryUrl,
						requestPriority = priority,
						cancelFn = cancelFn,
					}

				fakeDownloadSubscr =
					Rx.Observable.timer( 0.1, require 'WFRxScheduler'.sharedScheduler() )
						:subscribe( 
							function() 
								
								mockDownloadRequestStream
									:onNext( 
										destPath, 
										ozAssetId,
										ozBinaryType,
										ozBinarySubtype,
										ozBinaryUrl,
										priority ) 

							end )

				return cancelFn

			end,
			
		close = 
			function( self )

				self._closed = true

			end,

		getInFlightRequests =
			function( self )

				return self._binaryRequests
				
			end,

	}

	return mockBinaryRequester, mockDownloadResultStream, 
			mockDownloadRequestStream

end

--------------------------------------------------------------------------------

local function getBinaryRequestInfo( binaryRequest )

	if not binaryRequest then
		return nil
	end

	return binaryRequest.ozAssetId, binaryRequest.ozBinaryType, binaryRequest.ozBinarySize

end

--------------------------------------------------------------------------------

local function waitForNDownloadRequests( rxTest, scheduler,
											mockDownloadRequestStream, n )

	local sawNDownloadRequestsStream =
			rxTest:observe( 
					mockDownloadRequestStream
						:take( n )
						:takeLast( 0 ) )

	local SchedulerTestAPI = scheduler:_expose_test_API()

	SchedulerTestAPI.runScheduledRequests( scheduler )

	sawNDownloadRequestsStream:expectCompleted()

end

--------------------------------------------------------------------------------

--[[
local function dumpRequest( rxTest, maybeDumpPrefix )

	return function( request )

		rxTest:logf( "%s%s" .. string.rep( ", %s", #request - 1 ), 
						maybeDumpPrefix or "", unpack( request ) )

	end

end

--------------------------------------------------------------------------------

local function dumpActiveRequestsForLocalId( rxTest, scheduler, localId, 
												maybeDumpPrefix )

	local SchedulerTestAPI = scheduler:_expose_test_API()

	WFFunctionUtils.map(
		dumpRequest( rxTest, maybeDumpPrefix ),
		SchedulerTestAPI.getAllLocalAssetIdRequests( scheduler, localId ) )

end

--]]

--------------------------------------------------------------------------------

local function getCurrState( scheduler, wantedPixStreamId, wantedLocalAssetId )

	local currState

	scheduler:_mapAllRequestsWithPixStreamId( 
		wantedPixStreamId,
		function( request )

			local actualLocalAssetId = request[ 2 ]
			local state = request[ 5 ]

			if actualLocalAssetId == wantedLocalAssetId then

				currState = state

			end

		end )

	return currState

end

--==============================================================================
-- rxTestDownloadToFile
--==============================================================================

local function rxTestDownloadToFile( rxTest )

	local scheduler, catalog = makeSchedulerForTest( rxTest )

	local SchedulerTestAPI = scheduler:_expose_test_API()

	local needDownloadToFileStates = 
			PixelProviderDownloadToFileUtils.getNeedsDownloadToFileStates(
				scheduler )

	addRecordToDocStore( rxTest, catalog, 1 )

	for _, needDownloadToFileState in ipairs( needDownloadToFileStates ) do

		scheduler
			:addRequest( 
				pixStreamId( 1 ), localAssetId( 1 ),
				PixelPathProviderRequestTypeNames.grid_visible, 1 )

		local mockBinaryRequester, mockDownloadResultStream, mockDownloadRequestStream = 
				makeMockBinaryRequester( scheduler )

		scheduler._testData.runLoopTaskStates.PixelProviderDownloadToFileTask =
			PixelProviderDownloadToFileTask.privateUnitTestExport.injectMockBinaryRequester(
				scheduler._testData.runLoopTaskStates.PixelProviderDownloadToFileTask,
				2, mockBinaryRequester )

		for _, testWithBasicDownloadInfo in ipairs( { false, true } ) do

			local expectedTestMessage = 
					not testWithBasicDownloadInfo and
						"skipDownloadNeedInfo" or "download_to_file_finished"

			local sawTestMessage = 
					rxTest
						:observe( 
							scheduler._testData.testInfoStream
								:where( isTestMessage( expectedTestMessage ) )
								:take( 1 )
								:takeLast( 0 ) )

			local ozAssetId

			if testWithBasicDownloadInfo then

				ozAssetId = string.format( "ozAssetIdFor%s", localAssetId( 1 ) )

				scheduler._testData.runLoopCaches.sharedCaches.baseRenderInfoCache
					:pushCacheItem( 
						localAssetId( 1 ), 
						{ 
							ozAssetId = ozAssetId,
							changeCounter = 1, 
							revId = digest( 1 ),
							digest = digest( 1 ), 
							croppedLongSide = 1234,
							proxies = { { size="2560", href="notAProxy" } },
							ozValidOriginalLink = "yeahNope",
						} )

			end

			SchedulerTestAPI.setState( 
				scheduler, pixStreamId( 1 ), localAssetId( 1 ),
				needDownloadToFileState )

			rxTest:logf( "Running test of %s with%s basic download info expecting %s",
							needDownloadToFileState,
							testWithBasicDownloadInfo and "" or "out",
							expectedTestMessage )

			local initiatedDownloadRequestStream =
					rxTest:observe( mockDownloadRequestStream:take( 1 ):takeLast( 0 ) )

			SchedulerTestAPI.runScheduledRequests( scheduler )

			if testWithBasicDownloadInfo then

				initiatedDownloadRequestStream:expectCompleted()

				for _, requested in ipairs( mockBinaryRequester._binaryRequests ) do

					local ozAssetId2, ozBinaryType, ozBinarySubtype = 
							getBinaryRequestInfo( requested )

					mockDownloadResultStream
						:onNext( 
							true, 
							"fakePath",
							ozAssetId2, 
							ozBinaryType,
							ozBinarySubtype )

				end

			end

			sawTestMessage:expectCompleted()

		end

		local localBinaryPathCache =
			scheduler._testData.runLoopCaches.sharedCaches.localBinaryPathCache

		rxTest:log( "verify we have a path" )

		local path = 
			unpack( 
				localBinaryPathCache
					:popCacheItem( pixStreamId( 1 ), localAssetId( 1 ) ) or
					{} )

		rxTest:expectEqual( path, "fakePath" )

		scheduler._testData.runLoopCaches.sharedCaches.baseRenderInfoCache:destroyAll()

		scheduler:abandonAllPixStreamRequests( pixStreamId( 1 ) )

	end

end

--==============================================================================
-- rxTestDownloadFail
--==============================================================================

local function rxTestDownloadFail( rxTest )

	local scheduler, catalog = makeSchedulerForTest( rxTest )

	local SchedulerTestAPI = scheduler:_expose_test_API()

	local needDownloadOzBinaryStates = 
			PixelProviderDownloadToFileUtils.getNeedsDownloadToFileStates(
				scheduler )

	addRecordToDocStore( rxTest, catalog, 1 )

	for _, needDownloadOzBinaryState in ipairs( needDownloadOzBinaryStates ) do

		scheduler:addRequest( pixStreamId( 1 ), localAssetId( 1 ),
									PixelPathProviderRequestTypeNames.grid_visible, 1 )

		local mockBinaryRequester, mockDownloadResultStream, mockDownloadRequestStream = 
				makeMockBinaryRequester( scheduler )

		scheduler._testData.runLoopTaskStates.PixelProviderDownloadToFileTask =
			PixelProviderDownloadToFileTask.privateUnitTestExport.injectMockBinaryRequester(
				scheduler._testData.runLoopTaskStates.PixelProviderDownloadToFileTask,
				2, mockBinaryRequester )

		local expectedErrorStateTuples = {}

		local expectedTestMessage = "download_to_file_transient_fail"
		
		local sawTestMessage = 
				rxTest
					:observe( 
						scheduler._testData.testInfoStream
							:where( isTestMessage( expectedTestMessage ) )
							:take( 1 )
							:takeLast( 0 ) )

		addBaseRenderInfoCacheEntry( scheduler, 1 )

		SchedulerTestAPI.setState( scheduler, pixStreamId( 1 ), localAssetId( 1 ), 
									needDownloadOzBinaryState )

		expectedErrorStateTuples[ #expectedErrorStateTuples + 1 ] = { pixStreamId( 1 ), localAssetId( 1 ) }

		waitForNDownloadRequests( 
			rxTest, scheduler, mockDownloadRequestStream, 1 )

		for _, binaryRequest in ipairs( mockBinaryRequester._binaryRequests ) do

			local ozAssetId, ozBinaryType, ozBinarySubtype = 
										getBinaryRequestInfo( binaryRequest )

			local errStr = string.format( "fale wail #%s %s, %s, %s)", 1,
											ozAssetId, ozBinaryType,
											ozBinarySubtype )

			mockDownloadResultStream
				:onNext( 
					false, 
					errStr, 
					ozAssetId,
					ozBinaryType,
					ozBinarySubtype,
					{ url = 'something' } )

		end

		sawTestMessage:expectCompleted()

		local expectedErrorState =
			( needDownloadOzBinaryState == 
					PixelPathProviderStateMachine.DownloadToFileStates.NeedProxyDownloadToFileForBestPossibleLocalNegativePath and
						PixelPathProviderStateMachine.DownloadToFileStates.DownloadingOriginalToFileForBestPossibleLocalNegativePath ) or
			( needDownloadOzBinaryState == 
					PixelPathProviderStateMachine.DownloadToFileStates.NeedOriginalDownloadToFileForBestPossibleLocalNegativePath and
						PixelPathProviderStateMachine.SendDataStates.TransientDownloadBestPossibleNegativeToFileError ) or
				PixelPathProviderStateMachine.SendDataStates.TransientDownloadToFileError

		rxTest:logf( "Got %s test message expect to be in %s for state %s",
						expectedTestMessage,
						expectedErrorState,
						needDownloadOzBinaryState )

		rxTest:expectEqual( 
			sortTupleArray( expectedErrorStateTuples ),
			sortTupleArray( 
				scheduler:_mapAllRequestsWithState( 
							expectedErrorState,
							takeFirstN( 2 ) ) ) )

		rxTest:expectEqual( 
			{},
			sortTupleArray( 
				scheduler:_mapAllRequestsWithState( 
							PixelPathProviderStateMachine.SendDataStates.HaveLocalNegativePath,
							takeFirstN( 2 ) ) ) )

		scheduler._testData.runLoopCaches.sharedCaches.baseRenderInfoCache:destroyAll()

		scheduler:abandonAllPixStreamRequests( pixStreamId( 1 ) )

	end

end

--==============================================================================
-- rxTestDownloadCanceledByNewRequest
--==============================================================================

local function rxTestDownloadCanceledByNewRequest( rxTest )

	rxTest:setTimeLimit( 5 )

	local scheduler, catalog = makeSchedulerForTest( rxTest )

	local SchedulerTestAPI = scheduler:_expose_test_API()

	local needDownloadOzBinaryStates = 
			PixelProviderDownloadToFileUtils.getNeedsDownloadToFileStates(
				scheduler )

	addRecordToDocStore( rxTest, catalog, 1 )

	for _, needDownloadOzBinaryState in ipairs( needDownloadOzBinaryStates ) do

		local expectedTestDataTuples = {}

		scheduler:addRequest( pixStreamId( 1 ), localAssetId( 1 ),
									PixelPathProviderRequestTypeNames.grid_visible, 1 )

		expectedTestDataTuples[ #expectedTestDataTuples + 1 ] = { pixStreamId( 1 ), localAssetId( 1 ) }

		local mockBinaryRequester, mockDownloadResultStream, mockDownloadRequestStream = 
				makeMockBinaryRequester( scheduler )

		scheduler._testData.runLoopTaskStates.PixelProviderDownloadToFileTask =
			PixelProviderDownloadToFileTask.privateUnitTestExport.injectMockBinaryRequester(
				scheduler._testData.runLoopTaskStates.PixelProviderDownloadToFileTask,
				2, mockBinaryRequester )

		local expectedFinalTuples = {}

		local expectedTestMessage = "reschedule_download"
		
		local watchPixelDownloads = 
			rxTest
				:observe( 
					scheduler._testData.testInfoStream
						:where( isTestMessage( expectedTestMessage ) )
						:map( stripTestMessageAndMakeTuple )
						:take( #expectedTestDataTuples )
						:toArray()
						:map( sortTupleArray ) )

		addBaseRenderInfoCacheEntry( scheduler, 1 )

		SchedulerTestAPI.setState( scheduler, pixStreamId( 1 ), localAssetId( 1 ), 
									needDownloadOzBinaryState )

		expectedFinalTuples[ #expectedFinalTuples + 1 ] = { pixStreamId( 1 ), localAssetId( 1 ) }

		local testDesc = string.format( "expecting %s for %s",
											expectedTestMessage,
											needDownloadOzBinaryState )

		rxTest:log( "Scheduling a test expecting " .. testDesc )

		waitForNDownloadRequests( 
			rxTest, scheduler, mockDownloadRequestStream, 1 )

		local binaryRequest =  mockBinaryRequester._binaryRequests[ 1 ]

		local ozAssetId, ozBinaryType, ozBinarySubtype = 
									getBinaryRequestInfo( binaryRequest )

		local errStr = "canceledByNewRequest"

		mockDownloadResultStream:onNext( false, errStr, ozAssetId,
										ozBinaryType, ozBinarySubtype, { url = 'something' } )

		watchPixelDownloads:expectData( sortTupleArray( expectedTestDataTuples ) )

		rxTest:log( "Finishing a test expecting " .. testDesc )

		scheduler._testData.runLoopCaches.sharedCaches.baseRenderInfoCache:destroyAll()

		scheduler:abandonAllPixStreamRequests( pixStreamId( 1 ) )

	end

end

--==============================================================================
-- rxTestAbandonedDownload
--==============================================================================

local function rxTestAbandonedDownload( rxTest )

	rxTest:setTimeLimit( 5 )

	local scheduler, catalog = makeSchedulerForTest( rxTest )

	local SchedulerTestAPI = scheduler:_expose_test_API()

	local needDownloadOzBinaryStates = 
			PixelProviderDownloadToFileUtils.getNeedsDownloadToFileStates(
				scheduler )

	addRecordToDocStore( rxTest, catalog, 1 )

	for _, needDownloadOzBinaryState in ipairs( needDownloadOzBinaryStates ) do

		scheduler:addRequest( pixStreamId( 1 ), localAssetId( 1 ),
									PixelPathProviderRequestTypeNames.grid_visible, 1 )

		local mockBinaryRequester, _, mockDownloadRequestStream = 
				makeMockBinaryRequester( scheduler )

		scheduler._testData.runLoopTaskStates.PixelProviderDownloadToFileTask =
			PixelProviderDownloadToFileTask.privateUnitTestExport.injectMockBinaryRequester(
				scheduler._testData.runLoopTaskStates.PixelProviderDownloadToFileTask,
				2, mockBinaryRequester )

		local expectedDownloadingTuples = {}
		
		addBaseRenderInfoCacheEntry( scheduler, 1 )

		SchedulerTestAPI.setState( scheduler, pixStreamId( 1 ), localAssetId( 1 ), 
									needDownloadOzBinaryState )

		expectedDownloadingTuples[ #expectedDownloadingTuples + 1 ] = { pixStreamId( 1 ), localAssetId( 1 ) }

		local testDesc = string.format( "restart on abandon for %s",
											needDownloadOzBinaryState )

		rxTest:log( "Scheduling a test expecting " .. testDesc )

		waitForNDownloadRequests( 
			rxTest, scheduler, mockDownloadRequestStream, 1 )

		-- makes sure in flight requests exist
		rxTest:
			expectEqual( 
				false,
				not next( mockBinaryRequester._binaryRequests ) )

		scheduler:abandonAllPixStreamRequests( pixStreamId( 1 ) )

		-- now step through the runLoop again
		SchedulerTestAPI.runScheduledRequests( scheduler )

		-- makes sure in-flight requests are canceled
		rxTest:
			expectEqual( 
				true,
				not next( mockBinaryRequester._binaryRequests ) )

		scheduler._testData.runLoopCaches.sharedCaches.baseRenderInfoCache:destroyAll()

	end

end

--==============================================================================
-- rxTestExtraDownloads
--==============================================================================

local function rxTestExtraDownloads( rxTest )

	rxTest:setTimeLimit( 5 )

	local scheduler, catalog = makeSchedulerForTest( rxTest )

	local SchedulerTestAPI = scheduler:_expose_test_API()

	local needDownloadOzBinaryStates = 
			PixelProviderDownloadToFileUtils.getNeedsDownloadToFileStates(
				scheduler )

	addRecordToDocStore( rxTest, catalog, 1 )

	for _, needDownloadOzBinaryState in ipairs( needDownloadOzBinaryStates ) do

		-- add a request
		scheduler:addRequest( pixStreamId( 1 ), localAssetId( 1 ),
									PixelPathProviderRequestTypeNames.grid_visible, 1 )

		local mockBinaryRequester, mockDownloadResultStream, mockDownloadRequestStream = 
				makeMockBinaryRequester( scheduler )

		scheduler._testData.runLoopTaskStates.PixelProviderDownloadToFileTask =
			PixelProviderDownloadToFileTask.privateUnitTestExport.injectMockBinaryRequester(
				scheduler._testData.runLoopTaskStates.PixelProviderDownloadToFileTask,
				2, mockBinaryRequester )
		
		addBaseRenderInfoCacheEntry( scheduler, 1 )

		SchedulerTestAPI.setState( scheduler, pixStreamId( 1 ), localAssetId( 1 ), 
									needDownloadOzBinaryState )

		local expectedDownloadingTuples = {}

		expectedDownloadingTuples[ #expectedDownloadingTuples + 1 ] = { pixStreamId( 1 ), localAssetId( 1 ) }

		-- check if we are in expected start state
		rxTest:expectEqual( 
			sortTupleArray( expectedDownloadingTuples ),
			sortTupleArray( 
				WFFunctionUtils.map(
					takeFirstN( 2 ),
						scheduler:_getRequestsWithStatesInPriorityOrder( 
									needDownloadOzBinaryState  ) ) ) )

		waitForNDownloadRequests( 
			rxTest, scheduler, mockDownloadRequestStream, 1 )

		local currState = 
				getCurrState( scheduler, pixStreamId( 1 ), localAssetId( 1 ) )

		-- now send non matching success result
		for _, binaryRequest in ipairs( mockBinaryRequester._binaryRequests ) do

			local _, ozBinaryType, ozBinarySubtype = 
										getBinaryRequestInfo( binaryRequest )

			mockDownloadResultStream:onNext( true, 'fakePath', 'bogusOzAssetId',
											ozBinaryType, ozBinarySubtype )

		end

		-- now step through the runLoop again
		SchedulerTestAPI.runScheduledRequests( scheduler )

		-- make sure our tuples didn't change
		rxTest:expectEqual( 
			sortTupleArray( expectedDownloadingTuples ),
			sortTupleArray( 
				WFFunctionUtils.map(
					takeFirstN( 2 ),
						scheduler:_getRequestsWithStatesInPriorityOrder( currState ) ) ) )

		scheduler._testData.runLoopCaches.sharedCaches.baseRenderInfoCache:destroyAll()

		scheduler:abandonAllPixStreamRequests( pixStreamId( 1 ) )

	end

end

--==============================================================================
-- rxTestDownloadTaskBacklog
--==============================================================================

local function rxTestDownloadTaskBacklog( rxTest )

	rxTest:setTimeLimit( 5 )

	local scheduler, catalog = makeSchedulerForTest( rxTest )

	local SchedulerTestAPI = scheduler:_expose_test_API()

	local needsDownloadStates = 
			PixelProviderDownloadToFileUtils.getNeedsDownloadToFileStates(
				scheduler )

	SchedulerTestAPI.runScheduledRequests( scheduler )

	local mockBinaryRequester, _, mockDownloadRequestStream = 
			makeMockBinaryRequester( scheduler )

	scheduler._testData.runLoopTaskStates.PixelProviderDownloadToFileTask =
		PixelProviderDownloadToFileTask.privateUnitTestExport.injectMockBinaryRequester(
			scheduler._testData.runLoopTaskStates.PixelProviderDownloadToFileTask,
			#needsDownloadStates, mockBinaryRequester )

	for i = 1, #needsDownloadStates + 1 do

		addRecordToDocStore( rxTest, catalog, i )

		scheduler:addRequest( pixStreamId( i ), localAssetId( i ),
									PixelPathProviderRequestTypeNames.grid_visible, 1 )

		addBaseRenderInfoCacheEntry( scheduler, i )

		if i <= #needsDownloadStates then

			SchedulerTestAPI.setState( scheduler, pixStreamId( i ), localAssetId( i ), 
										needsDownloadStates[ i ] )

		end

	end

	-- run the loop long enough to set up a download for all of the types
	waitForNDownloadRequests( 
		rxTest, scheduler, mockDownloadRequestStream, #needsDownloadStates )

	rxTest:logf( "All %s downloads initiated, going to try to start another and expecting it to fail to start", #needsDownloadStates )

	local downloadWasSkipped = 
		rxTest
			:observe( 
				scheduler._testData.testInfoStream
					:where( isTestMessage( "skipDownloadDueToBacklog" ) ) 
					:take( 1 )
					:takeLast( 0 ) )

	-- add another download to file request and verify its stuck in backlog
	SchedulerTestAPI.setState( scheduler, 
								pixStreamId( #needsDownloadStates + 1 ), 
								localAssetId( #needsDownloadStates + 1 ), 
								needsDownloadStates[ 1 ] )

	SchedulerTestAPI.runScheduledRequests( scheduler )

	downloadWasSkipped:expectCompleted()

end

--==============================================================================
-- rxTestDownloadTaskRescheduling
--==============================================================================
	
local function extractLocalIdsFromRequests( requests )

	local localIds = {}
	
	for _, request in pairs( requests ) do
		localIds[ #localIds + 1 ] = { request[ 2 ] }
	end
	
	return localIds

end

--------------------------------------------------------------------------------

local function rxTestDownloadTaskRescheduling( rxTest )

	rxTest:setTimeLimit( 5 )

	local scheduler, catalog = makeSchedulerForTest( rxTest )

	local SchedulerTestAPI = scheduler:_expose_test_API()

	local needDownloadOzBinaryStates = 
			PixelProviderDownloadToFileUtils.getNeedsDownloadToFileStates(
				scheduler )

	local mockBinaryRequester, _, mockDownloadRequestStream = 
			makeMockBinaryRequester( scheduler )

	scheduler._testData.runLoopTaskStates.PixelProviderDownloadToFileTask =
		PixelProviderDownloadToFileTask.privateUnitTestExport.injectMockBinaryRequester(
			scheduler._testData.runLoopTaskStates.PixelProviderDownloadToFileTask,
			1, mockBinaryRequester )

	for _, needDownloadOzBinaryState in ipairs( needDownloadOzBinaryStates ) do

		local function pixStreamIdForDownloadState( i )

			return pixStreamId( i )

		end

		local maxDownloads = 6

		local function getRequestPriority( i )

			if i == 1 then

				return PixelPathProviderRequestTypeNames.loupe_next

			end

			if i == 2 then

				return PixelPathProviderRequestTypeNames.highest

			end

			if i == 3 then

				return PixelPathProviderRequestTypeNames.download_soon

			end

			if i == 4 then

				return PixelPathProviderRequestTypeNames.grid_prefetch

			end

			if i == 5 then

				return PixelPathProviderRequestTypeNames.grid_visible

			end

			return PixelPathProviderRequestTypeNames.idle

		end

		local expectedLocalIds = {}

		for i = 1, maxDownloads do

			addRecordToDocStore( rxTest, catalog, i )

			local requestPriority = getRequestPriority( i )

			if i == 1 then

				expectedLocalIds[ #expectedLocalIds + 1 ] = { localAssetId( i ) }

			end

			scheduler:addRequest( pixStreamIdForDownloadState( i ), localAssetId( i ), requestPriority, 1 )

			addBaseRenderInfoCacheEntry( scheduler, i )

			if i == 1 then

				SchedulerTestAPI.setState( 
					scheduler, pixStreamIdForDownloadState( i ), localAssetId( i ), needDownloadOzBinaryState )

			end

		end
		
		waitForNDownloadRequests( 
			rxTest, scheduler, mockDownloadRequestStream, 1 )

		local currState = 
				getCurrState( scheduler, pixStreamId( 1 ), localAssetId( 1 ) )

		rxTest:logf( "Checking that we have %s local ids in the %s state when starting from %s",
						#expectedLocalIds, currState, needDownloadOzBinaryState )

		rxTest:expectEqual( 
			sortTupleArray( expectedLocalIds, compareArrays ),
			sortTupleArray( 
				extractLocalIdsFromRequests( 
					SchedulerTestAPI.getRequestsWithStatesInPriorityOrder( scheduler,
						currState ),
				compareArrays ) ) )

		-- preempt the lowest priority in flight download:
		-- 1 is the lowest, 2 second lowest, 3 third lowest

		local extraRequestNum = 2

		addBaseRenderInfoCacheEntry( scheduler, extraRequestNum )

		rxTest:logf( "Testing preempt request priority %s", getRequestPriority( 1 ), 1 )

		expectedLocalIds[ 1 ] = { localAssetId( 2 ) }

		local watchPixelDownloads = 
			rxTest
				:observe( 
					scheduler._testData.testInfoStream
						:where( isTestMessage( "reschedulingDownload" ) ) 
						:take( 1 )
						:takeLast( 0 ) )

		SchedulerTestAPI.setState( scheduler, 
									pixStreamIdForDownloadState( 2 ), 
									localAssetId( 2 ), 
									needDownloadOzBinaryState )

		waitForNDownloadRequests( 
			rxTest, scheduler, mockDownloadRequestStream, 1 )

		watchPixelDownloads:expectCompleted()

		rxTest:expectEqual( 
			sortTupleArray( 
				expectedLocalIds, 
				compareArrays ),
			sortTupleArray( 
				extractLocalIdsFromRequests( 
					SchedulerTestAPI.getRequestsWithStatesInPriorityOrder( 
						scheduler, currState ),
				compareArrays ) ) )

		-- don't preempt a saturated download queue with a request that is of lower priority
		
		local watchPixelDownloads2 = 
			rxTest
				:observe( 
					scheduler._testData.testInfoStream
						:where( isTestMessage( "skipDownloadDueToBacklog" ) ) 
						:take( 1 )
						:takeLast( 0 ) )

		SchedulerTestAPI.setState( scheduler, 
									pixStreamIdForDownloadState( maxDownloads ), 
									localAssetId( maxDownloads ), 
									PixelPathProviderStateMachine.DownloadStates.NeedRenderDownload )

		SchedulerTestAPI.runScheduledRequests( scheduler )

		watchPixelDownloads2:expectCompleted()

		for i = 1, maxDownloads do

			scheduler:abandonAllLocalAssetIdRequests( localAssetId ( i ) )
	
		end

	end

end

--==============================================================================
-- rxTestDeletedPixelRequestsCauseCancelation
--==============================================================================

local function rxTestDeletedPixelRequestsCauseCancelation( rxTest )

	rxTest:setTimeLimit( 5 )

	local scheduler, catalog = makeSchedulerForTest( rxTest )

	local SchedulerTestAPI = scheduler:_expose_test_API()

	local needsDownloadStates = 
			PixelProviderDownloadToFileUtils.getNeedsDownloadToFileStates(
				scheduler )

	local binaryRequestNum 

	local function generateNextBinaryRequestNum()

		binaryRequestNum = ( binaryRequestNum or 0 ) + 1

		return binaryRequestNum 

	end

	local mockBinaryRequester, _, mockDownloadRequestStream = 
		makeMockBinaryRequester( scheduler, generateNextBinaryRequestNum )

	scheduler._testData.runLoopTaskStates.PixelProviderDownloadToFileTask =
		PixelProviderDownloadToFileTask.privateUnitTestExport.injectMockBinaryRequester(
			scheduler._testData.runLoopTaskStates.PixelProviderDownloadToFileTask,
			1, mockBinaryRequester )

	addRecordToDocStore( rxTest, catalog, 1 )

	addRecordToDocStore( rxTest, catalog, 2 )

	for i, needsDownloadState in ipairs( needsDownloadStates ) do

		rxTest:logf( "--------------------------------------------------Run with state: %s", needsDownloadState )

		rxTest:log( "add two requests for same localAssetId with highest and lowest priorities" )

		scheduler:addRequest( pixStreamId( 1 ), localAssetId( 1 ),
									PixelPathProviderRequestTypeNames.highest, 1 )

		scheduler:addRequest( pixStreamId( 2 ), localAssetId( 1 ),
									PixelPathProviderRequestTypeNames.idle, 1 )

		addBaseRenderInfoCacheEntry( scheduler, 1 )

		rxTest:log( "add a request for a different localAssetId at grid_visible" )

		scheduler:addRequest( pixStreamId( 1 ), localAssetId( 2 ),
									PixelPathProviderRequestTypeNames.grid_visible, 1 )

		addBaseRenderInfoCacheEntry( scheduler, 2 )

		rxTest:log( "start downloading at highest priority" )

		SchedulerTestAPI.setState( scheduler, pixStreamId( 1 ), localAssetId( 1 ), 
									needsDownloadState )
		
		waitForNDownloadRequests( 
			rxTest, scheduler, mockDownloadRequestStream, 1 )

		rxTest:logf( "verify scheduler thinks it is downloading %s.%s",
						pixStreamId( 1 ), localAssetId( 1 ) )

		local downloadingToFileStates =
				PixelProviderDownloadToFileUtils.getIsDownloadingToFileStates( scheduler )

		rxTest:expectEqual( 
			sortTupleArray( { { pixStreamId( 1 ), localAssetId( 1 ) }, } ),
			sortTupleArray( WFFunctionUtils.map(
								takeFirstN( 2 ),
								SchedulerTestAPI.getRequestsWithStatesInPriorityOrder( 
									scheduler,
									unpack( downloadingToFileStates ) ) ) ) )

		rxTest:log( "verify mockDownloader is downloading localAssetId( 1 )" )

		rxTest:expectEqual( 
			mockBinaryRequester._binaryRequests[ 3*i - 2 ].ozAssetId,
			makeOzAssetId( 1 ) )
		
		rxTest:logf( "other request now should join download for download state: %s", needsDownloadState )

		SchedulerTestAPI.setState( scheduler, pixStreamId( 2 ), localAssetId( 1 ), 
									needsDownloadState )

		SchedulerTestAPI.runScheduledRequests( scheduler )

		rxTest:log( "verify scheduler thinks both are downloading now" )

		rxTest:expectEqual( 
			sortTupleArray( { { pixStreamId( 1 ), localAssetId( 1 ) },
								{ pixStreamId( 2 ), localAssetId( 1 ) } } ),
			sortTupleArray( WFFunctionUtils.map(
								takeFirstN( 2 ),
								SchedulerTestAPI.getRequestsWithStatesInPriorityOrder( 
									scheduler,
									unpack( downloadingToFileStates ) ) ) ) )

		rxTest:log( "verify mockDownloader is downloading localAssetId( 1 )" )

		rxTest:expectEqual( 
			mockBinaryRequester._binaryRequests[ 3*i - 2 ].ozAssetId,
			makeOzAssetId( 1 ) )

		rxTest:log( "add another assetId, which should be backlogged" )
	
		local downloadWasSkipped = 
			rxTest
				:observe( 
					scheduler._testData.testInfoStream
						:where( isTestMessage( "skipDownloadDueToBacklog" ) ) 
						:take( 1 )
						:takeLast( 0 ) )

		SchedulerTestAPI.setState( scheduler, pixStreamId( 1 ), localAssetId( 2 ), 
									needsDownloadState )

		SchedulerTestAPI.runScheduledRequests( scheduler )

		rxTest:log( "verify this causes backlog" )

		downloadWasSkipped:expectCompleted()

		rxTest:log( "verify scheduler thinks both local asset id 1 requests are downloading now" )

		rxTest:expectEqual( 
			sortTupleArray( { { pixStreamId( 1 ), localAssetId( 1 ) },
								{ pixStreamId( 2 ), localAssetId( 1 ) } } ),
			sortTupleArray( WFFunctionUtils.map(
								takeFirstN( 2 ),
								SchedulerTestAPI.getRequestsWithStatesInPriorityOrder( 
									scheduler,
									unpack( downloadingToFileStates ) ) ) ) )

		rxTest:log( "verify scheduler thinks other assetId is still queued" )

		rxTest:expectEqual( 
			sortTupleArray( { { pixStreamId( 1 ), localAssetId( 2 ) }, } ),
			sortTupleArray( WFFunctionUtils.map(
								takeFirstN( 2 ),
								SchedulerTestAPI.getRequestsWithStatesInPriorityOrder( 
									scheduler, needsDownloadState ) ) ) )

		rxTest:log( "verify mockDownloader is downloading localAssetId( 1 )" )

		rxTest:expectEqual( 
			mockBinaryRequester._binaryRequests[  3*i - 2 ].ozAssetId,
			makeOzAssetId( 1 ) )

		local existingDownloadRescheduled = 
			rxTest
				:observe( 
					scheduler._testData.testInfoStream
						:where( isTestMessage( "reschedulingDownload" ) )
						:take( 1 )
						:takeLast( 0 ) )

		rxTest:logf( "remove request %s, %s" ,pixStreamId( 1 ), localAssetId( 1 ) )

		scheduler:removeRequest( pixStreamId( 1 ), localAssetId( 1 ) )

		rxTest:logf( "verify active other request %s, %s" ,pixStreamId( 1 ), localAssetId( 2 ) )

		rxTest:expectEqual( 
			sortTupleArray( { { pixStreamId( 2 ), localAssetId( 1 ) } } ),
			sortTupleArray( WFFunctionUtils.map(
								takeFirstN( 2 ),
								SchedulerTestAPI.getRequestsWithStatesInPriorityOrder( 
									scheduler,
									unpack( downloadingToFileStates ) ) ) ) )

		SchedulerTestAPI.runScheduledRequests( scheduler )

		rxTest:log( "wait for test message confirming we canceled the deleted pixel download" )

		existingDownloadRescheduled:expectCompleted()

		rxTest:log( "wait until we've started the new download after cancelation" )

		waitForNDownloadRequests( 
			rxTest, scheduler, mockDownloadRequestStream, 1 )

		rxTest:log( "verify mockDownloader is downloading localAssetId( 2 )" )

		rxTest:expectEqual( 
			mockBinaryRequester._binaryRequests[  3*i - 1 ].ozAssetId,
			makeOzAssetId( 2 ) )

		SchedulerTestAPI.runScheduledRequests( scheduler )

		rxTest:log( "verify scheduler thinks localAssetId( 2 ) is downloading" )

		rxTest:expectEqual( 
			sortTupleArray( { { pixStreamId( 1 ), localAssetId( 2 ) }, } ),
			sortTupleArray( WFFunctionUtils.map(
								takeFirstN( 2 ),
								SchedulerTestAPI.getRequestsWithStatesInPriorityOrder( 
									scheduler,
									unpack( downloadingToFileStates ) ) ) ) )

		rxTest:log( "verify scheduler thinks localAssetId( 1 ) is queued" )

		rxTest:expectEqual( 
			sortTupleArray( { { pixStreamId( 2 ), localAssetId( 1 ) }, } ),
			sortTupleArray( WFFunctionUtils.map(
								takeFirstN( 2 ),
								SchedulerTestAPI.getRequestsWithStatesInPriorityOrder( 
									scheduler, unpack( needsDownloadStates ) ) ) ) )

		rxTest:log( "verify mockDownloader is downloading localAssetId( 2 )" )

		rxTest:expectEqual( 
			mockBinaryRequester._binaryRequests[ 3*i - 1 ].ozAssetId,
			makeOzAssetId( 2 ) )

		rxTest:log( "verify mockDownloader has discarded original download" )

		rxTest:expectEqual( 
			mockBinaryRequester._binaryRequests[  3*i - 2 ],
			nil )

		rxTest:log( "delete grid_visible request" )

		scheduler:removeRequest( pixStreamId( 1 ), localAssetId( 2 ) )

		waitForNDownloadRequests( 
			rxTest, scheduler, mockDownloadRequestStream, 1 )

		rxTest:log( "verify scheduler thinks other localAssetId( 1 ) is downloading" )

		rxTest:expectEqual( 
			sortTupleArray( { { pixStreamId( 2 ), localAssetId( 1 ) }, } ),
			sortTupleArray( WFFunctionUtils.map(
								takeFirstN( 2 ),
								SchedulerTestAPI.getRequestsWithStatesInPriorityOrder( 
									scheduler,
									unpack( downloadingToFileStates ) ) ) ) )

		rxTest:log( "verify scheduler thinks nothing is queued" )

		rxTest:expectEqual( 
			sortTupleArray( {} ),
			sortTupleArray( WFFunctionUtils.map(
								takeFirstN( 2 ),
								SchedulerTestAPI.getRequestsWithStatesInPriorityOrder( 
									scheduler, unpack( needsDownloadStates ) ) ) ) )

		rxTest:log( "verify mockDownloader is downloading localAssetId( 1 ) again" )

		rxTest:expectEqual( 
			mockBinaryRequester._binaryRequests[ 3*i ].ozAssetId,
			makeOzAssetId( 1 ) )

		rxTest:log( "verify mockDownloader has discarded second download" )

		rxTest:expectEqual( 
			mockBinaryRequester._binaryRequests[ 3*i - 1 ],
			nil )

		rxTest:log( "delete idle request" )

		scheduler:removeRequest( pixStreamId( 2 ), localAssetId( 1 ) )

		SchedulerTestAPI.runScheduledRequests( scheduler )

		rxTest:log( "verify scheduler thinks nothing is downloading" )

		rxTest:expectEqual( 
			{},
			sortTupleArray( WFFunctionUtils.map(
								takeFirstN( 2 ),
								SchedulerTestAPI.getRequestsWithStatesInPriorityOrder( 
									scheduler,
									unpack( downloadingToFileStates ) ) ) ) )

		rxTest:log( "verify scheduler thinks nothing is queued" )

		rxTest:expectEqual( 
			sortTupleArray( {} ),
			sortTupleArray( WFFunctionUtils.map(
								takeFirstN( 2 ),
								SchedulerTestAPI.getRequestsWithStatesInPriorityOrder( 
									scheduler, needsDownloadState ) ) ) )

		rxTest:log( "verify mockDownloader is not downloding" )

		rxTest:expectEqual( 
			mockBinaryRequester._binaryRequests[ 3*i ],
			nil )

	end

end

--==============================================================================
-- rxTestHigherHTTPPrioritySameDownload
--==============================================================================

local function rxTestHigherHTTPPrioritySameDownload( rxTest )

	rxTest:setTimeLimit( 5 )

	local scheduler, catalog = makeSchedulerForTest( rxTest )

	local SchedulerTestAPI = scheduler:_expose_test_API()

	addRecordToDocStore( rxTest, catalog, 1 )

	local needsDownloadStates = 
			PixelProviderDownloadToFileUtils.getNeedsDownloadToFileStates(
				scheduler )

	local binaryRequestNum 

	local function generateNextBinaryRequestNum()

		binaryRequestNum = ( binaryRequestNum or 0 ) + 1

		return binaryRequestNum 

	end

	local mockBinaryRequester, _, mockDownloadRequestStream = 
		makeMockBinaryRequester( scheduler, generateNextBinaryRequestNum )

	scheduler._testData.runLoopTaskStates.PixelProviderDownloadToFileTask =
		PixelProviderDownloadToFileTask.privateUnitTestExport.injectMockBinaryRequester(
			scheduler._testData.runLoopTaskStates.PixelProviderDownloadToFileTask,
			1, mockBinaryRequester )

	for i, needsDownloadState in ipairs( needsDownloadStates ) do

		rxTest:logf( "--------------------------------------------------Run with state: %s", needsDownloadState )

		rxTest:log( "add two requests for same localAssetId with highest and lowest priorities" )

		scheduler:addRequest( pixStreamId( 1 ), localAssetId( 1 ),
									PixelPathProviderRequestTypeNames.idle, 1 )

		scheduler:addRequest( pixStreamId( 2 ), localAssetId( 1 ),
									PixelPathProviderRequestTypeNames.highest, 1 )

		addBaseRenderInfoCacheEntry( scheduler, 1 )

		rxTest:log( "start downloading at idle priority" )

		SchedulerTestAPI.setState( scheduler, pixStreamId( 1 ), localAssetId( 1 ), 
									needsDownloadState )

		waitForNDownloadRequests( 
			rxTest, scheduler, mockDownloadRequestStream, 1 )

		rxTest:log( "verify scheduler thinks it is downloading" )

		local downloadingToFileStates =
				PixelProviderDownloadToFileUtils.getIsDownloadingToFileStates( scheduler )

		rxTest:expectEqual( 
			sortTupleArray( { { pixStreamId( 1 ), localAssetId( 1 ) }, } ),
			sortTupleArray( WFFunctionUtils.map(
								takeFirstN( 2 ),
								SchedulerTestAPI.getRequestsWithStatesInPriorityOrder( 
									scheduler,
									unpack( downloadingToFileStates ) ) ) ) )

		rxTest:log( "verify mockDownloader is downloading localAssetId( 1 )" )

		rxTest:expectEqual( 
			mockBinaryRequester._binaryRequests[ 2*i - 1 ].ozAssetId,
			makeOzAssetId( 1 ) )
		
		rxTest:log( "other higher priority should cancel download and restart it at higher priority" )

		SchedulerTestAPI.setState( scheduler, pixStreamId( 2 ), localAssetId( 1 ), 
									needsDownloadState )

		SchedulerTestAPI.runScheduledRequests( scheduler )

		waitForNDownloadRequests( 
			rxTest, scheduler, mockDownloadRequestStream, 1 )

		rxTest:log( "verify scheduler thinks both are downloading now" )

		rxTest:expectEqual( 
			sortTupleArray( { { pixStreamId( 1 ), localAssetId( 1 ) },
								{ pixStreamId( 2 ), localAssetId( 1 ) } } ),
			sortTupleArray( WFFunctionUtils.map(
								takeFirstN( 2 ),
								SchedulerTestAPI.getRequestsWithStatesInPriorityOrder( 
									scheduler,
									unpack( downloadingToFileStates ) ) ) ) )

		rxTest:log( "verify mockDownloader is downloading localAssetId( 1 ) on different request" )

		rxTest:expectEqual( 
			mockBinaryRequester._binaryRequests[ 2*i ].ozAssetId,
			makeOzAssetId( 1 ) )

		rxTest:log( "verify mockDownloader orig request is gone" )

		rxTest:expectEqual( 
			mockBinaryRequester._binaryRequests[ 2*i - 1 ],
			nil )

		rxTest:log( "delete highest requests" )

		scheduler:removeRequest( pixStreamId( 1 ), localAssetId( 1 ) )

		scheduler:removeRequest( pixStreamId( 2 ), localAssetId( 1 ) )

		SchedulerTestAPI.runScheduledRequests( scheduler )

		rxTest:log( "verify scheduler thinks nothing is downloading" )

		rxTest:expectEqual( 
			{},
			sortTupleArray( WFFunctionUtils.map(
								takeFirstN( 2 ),
								SchedulerTestAPI.getRequestsWithStatesInPriorityOrder( 
									scheduler,
									unpack( downloadingToFileStates ) ) ) ) )

		rxTest:log( "verify scheduler thinks nothing is queued" )

		rxTest:expectEqual( 
			sortTupleArray( {} ),
			sortTupleArray( WFFunctionUtils.map(
								takeFirstN( 2 ),
								SchedulerTestAPI.getRequestsWithStatesInPriorityOrder( 
									scheduler, needsDownloadState ) ) ) )

		rxTest:log( "verify mockDownloader is not downloding" )

		rxTest:expectEqual( 
			mockBinaryRequester._binaryRequests[ 2*i ],
			nil )

	end

end

--==============================================================================
-- rxTestJoiningInFlightDownload
--==============================================================================

local currNewPixStreamId

--------------------------------------------------------------------------------

local function getNewPixStreamId( maybeState )

	currNewPixStreamId = ( currNewPixStreamId or 0 ) + 1

	local pixStreamIdPrefix =
			string.format( "%sNewPixStreamId", maybeState or "" )

	return pixStreamId( currNewPixStreamId, pixStreamIdPrefix )

end

--------------------------------------------------------------------------------

local function makeStateToPixStreamIdMap( stateArray )

	return WFFunctionUtils.reduce( {}, 
				function( stateMap, state )

					stateMap[ state ] = getNewPixStreamId( state )

					return stateMap 

				end,
				stateArray )

end

--------------------------------------------------------------------------------

local function setRequestState( scheduler, pixStreamIdForState, state )

	local SchedulerTestAPI = scheduler:_expose_test_API()

	SchedulerTestAPI.setState( 
		scheduler, pixStreamIdForState, localAssetId( 1 ), state )

end

--------------------------------------------------------------------------------

local function rxTestJoiningInFlightDownload( rxTest )

	rxTest:setTimeLimit( 5 )

	local scheduler, catalog = makeSchedulerForTest( rxTest )

	local SchedulerTestAPI = scheduler:_expose_test_API()

	addRecordToDocStore( rxTest, catalog, 1 )

	local needsDownloadSourceStates =
			PixelProviderDownloadToFileUtils.privateUnitTestExports.getNeedsDownloadToFileSourceStateArray( scheduler )

	local needsDownloadSourceStatesPixStreamIdMap =
			makeStateToPixStreamIdMap( needsDownloadSourceStates )

	local needsDownloadStates =
			PixelProviderDownloadToFileUtils.getNeedsDownloadToFileStates( scheduler )

	local needsDownloadStatesPixStreamIdMap =
			makeStateToPixStreamIdMap( needsDownloadStates )

	for _, needsDownloadState in ipairs( needsDownloadStates ) do

		local mockBinaryRequester, _, mockDownloadRequestStream = 
				makeMockBinaryRequester( scheduler )

		scheduler._testData.runLoopTaskStates.PixelProviderDownloadToFileTask =
			PixelProviderDownloadToFileTask.privateUnitTestExport.injectMockBinaryRequester(
				scheduler._testData.runLoopTaskStates.PixelProviderDownloadToFileTask,
				1, mockBinaryRequester )

		addBaseRenderInfoCacheEntry( scheduler, 1 )

		local mainRequestBinaryType = 
				PixelProviderDownloadToFileUtils.maybeGetBinaryTypeForDownloadToFileState(
					scheduler, needsDownloadState )

		local mainRequestPixStreamId = getNewPixStreamId( "MainRequest" )

		rxTest:logf( "--------------------------------------------------Run with state: %s", needsDownloadState )

		rxTest:expectEqual( 
			true,
			not next( mockBinaryRequester._binaryRequests ) )

		local initialStateRequestTuples = {}

		scheduler
			:addRequest( 
				mainRequestPixStreamId, localAssetId( 1 ), 
				PixelPathProviderRequestTypeNames.highest, 1 )

		initialStateRequestTuples[ #initialStateRequestTuples + 1 ] =
				{ needsDownloadState, mainRequestPixStreamId, localAssetId( 1 ) }

		setRequestState( scheduler, mainRequestPixStreamId, needsDownloadState )

		rxTest:logf( "wait for download to start for request %s", mainRequestPixStreamId )

		waitForNDownloadRequests( 
			rxTest, scheduler, mockDownloadRequestStream, 1 )

		rxTest:log( "verify mockDownloader is downloading localAssetId( 1 )" )

		rxTest:expectEqual( 
			mockBinaryRequester._binaryRequests[ 1 ].ozAssetId,
			makeOzAssetId( 1 ) )
		
		-- now add requests for all the needs download source states, we want to verify 
		-- that those requests transition to downloading if we start downloading the
		-- right binary type

		-- first suspend the download task so we can control when the scheduler runs
		SchedulerTestAPI.suspendRunLoop( scheduler )

		for initialState, pixStrmId in pairs( needsDownloadSourceStatesPixStreamIdMap ) do

			initialStateRequestTuples[ #initialStateRequestTuples + 1 ] = 
				{ initialState, pixStrmId, localAssetId( 1 ) }

			scheduler
				:addRequest( 
					pixStrmId, localAssetId( 1 ), 
					PixelPathProviderRequestTypeNames.grid_visible, 1 )

		end

		for initialState, pixStrmId in pairs( needsDownloadStatesPixStreamIdMap ) do

			initialStateRequestTuples[ #initialStateRequestTuples + 1 ] = 
				{ initialState, pixStrmId, localAssetId( 1 ) }

			scheduler
				:addRequest( 
					pixStrmId, localAssetId( 1 ), 
					PixelPathProviderRequestTypeNames.grid_visible, 1 )

		end
		
		local expectedIsDownloadingTuples = {}

		local expectedNeedsDownloadTuples = {}

		WFFunctionUtils.map(
			function( initialStateRequestTuple )

				local initialState, pixStreamIdForState, localId = 
					unpack( initialStateRequestTuple )

				local hasSameBinaryTypeAsMainRequest =
						PixelProviderDownloadToFileUtils.privateUnitTestExports.isBinaryTypeForDownloadToFileState( 
							scheduler, initialState, mainRequestBinaryType )

				if hasSameBinaryTypeAsMainRequest then

					expectedIsDownloadingTuples[ #expectedIsDownloadingTuples + 1 ] = 
						{ pixStreamIdForState, localId }

				else

					expectedNeedsDownloadTuples[ #expectedNeedsDownloadTuples + 1 ] = 
						{ pixStreamIdForState, localId }

				end

				SchedulerTestAPI.setState( 
					scheduler, pixStreamIdForState, localId, initialState )

			end,
			initialStateRequestTuples )

		rxTest:log( "run the loop so that requests can join" )

		SchedulerTestAPI.resumeRunLoop( scheduler )

		SchedulerTestAPI.runScheduledRequests( scheduler )

		rxTest:log( "verify all expected requests are downloading" )

		local downloadingToFileStates =
				PixelProviderDownloadToFileUtils.getIsDownloadingToFileStates( scheduler )

		local actualPostDownloadTuples =
			WFFunctionUtils.map(
				takeFirstN( 2 ),
				SchedulerTestAPI.getRequestsWithStatesInPriorityOrder( 
					scheduler,
					unpack( downloadingToFileStates ) ) )

		rxTest:expectEqual( 
			sortTupleArray( expectedIsDownloadingTuples ),
			sortTupleArray( actualPostDownloadTuples ) )

		rxTest:log( "verify all expected requests are still in needs download source states" )

		local actualNeedsDownloadSourceStateTuples =
			WFFunctionUtils.map(
				takeFirstN( 2 ),
				SchedulerTestAPI.getRequestsWithStatesInPriorityOrder( 
					scheduler,
					unpack( needsDownloadSourceStates ) ) )

		rxTest:expectEqual( 
			sortTupleArray( {} ),
			sortTupleArray( actualNeedsDownloadSourceStateTuples ) )

		rxTest:log( "verify all expected requests are still in needs download states" )

		local actualNeedsDownloadTuples =
			WFFunctionUtils.map(
				takeFirstN( 2 ),
				SchedulerTestAPI.getRequestsWithStatesInPriorityOrder( 
					scheduler,
					unpack( needsDownloadStates ) ) )

		rxTest:expectEqual(
			sortTupleArray( expectedNeedsDownloadTuples ),
			sortTupleArray( actualNeedsDownloadTuples ) )

		scheduler:abandonAllLocalAssetIdRequests( localAssetId ( 1 ) )

		SchedulerTestAPI.runScheduledRequests( scheduler )

	end

end

--==============================================================================
-- rxTestJoiningJustCompletedDownload
--==============================================================================

local function rxTestJoiningJustCompletedDownload( rxTest )

	rxTest:setTimeLimit( 5 )

	local scheduler, catalog = makeSchedulerForTest( rxTest )

	local SchedulerTestAPI = scheduler:_expose_test_API()

	addRecordToDocStore( rxTest, catalog, 1 )

	local needsDownloadSourceStates =
			PixelProviderDownloadToFileUtils.privateUnitTestExports.getNeedsDownloadToFileSourceStateArray( scheduler )

	local needsDownloadSourceStatesPixStreamIdMap =
			makeStateToPixStreamIdMap( needsDownloadSourceStates )

	local needsDownloadStates =
			PixelProviderDownloadToFileUtils.getNeedsDownloadToFileStates( scheduler )

	local needsDownloadStatesPixStreamIdMap =
			makeStateToPixStreamIdMap( needsDownloadStates )

	for _, downloadOk in ipairs( { false, true } ) do

		for _, needsDownloadState in ipairs( needsDownloadStates ) do

			local mockBinaryRequester, mockDownloadResultStream, mockDownloadRequestStream = 
					makeMockBinaryRequester( scheduler )

			scheduler._testData.runLoopTaskStates.PixelProviderDownloadToFileTask =
				PixelProviderDownloadToFileTask.privateUnitTestExport.injectMockBinaryRequester(
					scheduler._testData.runLoopTaskStates.PixelProviderDownloadToFileTask,
					1, mockBinaryRequester )

			addBaseRenderInfoCacheEntry( scheduler, 1 )

			local mainRequestBinaryType = 
					PixelProviderDownloadToFileUtils.maybeGetBinaryTypeForDownloadToFileState(
						scheduler, needsDownloadState )

			local mainRequestPixStreamId = getNewPixStreamId( "MainRequest" )

			rxTest:logf( "----Test join download %s for state: %s", 
							downloadOk and "success" or "failure", needsDownloadState )

			rxTest:expectEqual( 
				true,
				not next( mockBinaryRequester._binaryRequests ) )

			local initialStateRequestTuples = {}

			initialStateRequestTuples[ #initialStateRequestTuples + 1 ] =
					{ needsDownloadState, mainRequestPixStreamId, localAssetId( 1 ) }

			scheduler
				:addRequest( 
					mainRequestPixStreamId, localAssetId( 1 ), 
					PixelPathProviderRequestTypeNames.highest, 1 )

			setRequestState( scheduler, mainRequestPixStreamId, needsDownloadState )

			rxTest:logf( "wait for download to start for request %s", mainRequestPixStreamId )

			waitForNDownloadRequests( 
				rxTest, scheduler, mockDownloadRequestStream, 1 )

			rxTest:log( "verify mockDownloader is downloading localAssetId( 1 )" )

			local binaryRequestOzAssetId, binaryRequestOzBinaryType,
					binaryRequestOzBinarySubtype = 
						getBinaryRequestInfo( 
							mockBinaryRequester._binaryRequests[ 1 ] )

			rxTest:expectEqual( binaryRequestOzAssetId, makeOzAssetId( 1 ) )
			
			-- now add requests for all the needs download source states, we want to verify 
			-- that those requests transition to downloading if we start downloading the
			-- right binary type

			-- first suspend the download task so we can control when the scheduler runs
			SchedulerTestAPI.suspendRunLoop( scheduler )

			for initialState, pixStrmId in pairs( needsDownloadSourceStatesPixStreamIdMap ) do

				initialStateRequestTuples[ #initialStateRequestTuples + 1 ] = 
					{ initialState, pixStrmId, localAssetId( 1 ) }

				scheduler
					:addRequest( 
						pixStrmId, localAssetId( 1 ), 
						PixelPathProviderRequestTypeNames.grid_visible, 1 )

			end

			for initialState, pixStrmId in pairs( needsDownloadStatesPixStreamIdMap ) do

				initialStateRequestTuples[ #initialStateRequestTuples + 1 ] = 
					{ initialState, pixStrmId, localAssetId( 1 ) }

				scheduler
					:addRequest( 
						pixStrmId, localAssetId( 1 ), 
						PixelPathProviderRequestTypeNames.grid_visible, 1 )

			end

			rxTest:log( "download has started, set all non main requests to their initial states" )

			WFFunctionUtils.map(
				function( initialStateRequestTuple )

					local initialState, pixStreamIdForState, localId = 
						unpack( initialStateRequestTuple )

					if pixStreamIdForState == mainRequestPixStreamId then

						return

					end

					SchedulerTestAPI.setState( 
						scheduler, pixStreamIdForState, localId, initialState )

				end,
				initialStateRequestTuples )

			local downloadToFileFailureDestStatesMap =
					KSTableUtils.mapArrayToSet( 
						PixelProviderDownloadToFileUtils.privateUnitTestExports.getDownloadToFileFailureDestStates( 
							scheduler ) )

			local shouldHavePostDownloadStateTuples = 
				WFFunctionUtils.reduce(
					{},
					function( tuples, initialStateRequestTuple )

						local initialState, pixStreamIdForState, localId = 
							unpack( initialStateRequestTuple )

						local hasSameBinaryTypeAsMainRequest =
								PixelProviderDownloadToFileUtils.privateUnitTestExports.isBinaryTypeForDownloadToFileState( 
									scheduler, initialState, mainRequestBinaryType )

						local initialStateIsAlsoDownloadToFileFailureDestState =
								downloadToFileFailureDestStatesMap[ initialState ]

						if hasSameBinaryTypeAsMainRequest or
								( not downloadOk and mainRequestBinaryType == 'proxy' and
									initialStateIsAlsoDownloadToFileFailureDestState ) then

							tuples[ #tuples + 1 ] = 
								{ pixStreamIdForState, localId }

						end

						return tuples

					end,
					initialStateRequestTuples )

			local finishedHandlingDownloadResultStream =
				rxTest
					:observe( 
						scheduler._testData.testInfoStream
							:where( isTestMessage( "finished handling download result" ) )
							:take( 1 )
							:takeLast( 0 ) )

			rxTest:logf( "emulate download %s", downloadOk and "success" or "failure" )

			if downloadOk then

				mockDownloadResultStream
					:onNext( 
						true, 
						"fakePath",
						binaryRequestOzAssetId,
						binaryRequestOzBinaryType,
						binaryRequestOzBinarySubtype )

			else

				mockDownloadResultStream
					:onNext( 
						false, 
						"fake error",
						binaryRequestOzAssetId,
						binaryRequestOzBinaryType,
						binaryRequestOzBinarySubtype )

			end

			rxTest:log( "wait for download result handling completion" )

			finishedHandlingDownloadResultStream:expectCompleted()

			rxTest:log( "verify all expected requests were updated by download result" )

			local postDownloadFileStates =
					downloadOk and
						PixelProviderDownloadToFileUtils.privateUnitTestExports.getDownloadToFileSuccessDestStates( 
							scheduler ) or
						PixelProviderDownloadToFileUtils.privateUnitTestExports.getDownloadToFileFailureDestStates( 
							scheduler ) 

			local actualPostDownloadTuples =
				WFFunctionUtils.map(
					takeFirstN( 2 ),
					SchedulerTestAPI.getRequestsWithStatesInPriorityOrder( 
						scheduler,
						unpack( postDownloadFileStates ) ) )

			rxTest:expectEqual( 
				sortTupleArray( shouldHavePostDownloadStateTuples ),
				sortTupleArray( actualPostDownloadTuples ) )

			if downloadOk then

				local localBinaryPathCache =
					scheduler._testData.runLoopCaches.sharedCaches.localBinaryPathCache

				rxTest:log( "verify all expected requests have a path" )

				for _, tuple in ipairs( shouldHavePostDownloadStateTuples ) do

					local tuplePixStreamId, tupleLocalAssetId = unpack( tuple )

					local path = 
						unpack( 
							localBinaryPathCache
								:popCacheItem( 
									tuplePixStreamId, tupleLocalAssetId ) or
								{} )

					rxTest:expectEqual( path, "fakePath" )

				end

			end

			scheduler:abandonAllLocalAssetIdRequests( localAssetId ( 1 ) )

			SchedulerTestAPI.resumeRunLoop( scheduler )

			SchedulerTestAPI.runScheduledRequests( scheduler )

		end

	end

end

--==============================================================================

local testCases = {

	rxTestDownloadToFile = rxTestDownloadToFile,

	rxTestDownloadFail = rxTestDownloadFail,

	rxTestDownloadCanceledByNewRequest = rxTestDownloadCanceledByNewRequest,

	rxTestAbandonedDownload = rxTestAbandonedDownload,

	rxTestExtraDownloads = rxTestExtraDownloads,

	rxTestDownloadTaskBacklog = rxTestDownloadTaskBacklog,

	rxTestDownloadTaskRescheduling = rxTestDownloadTaskRescheduling,

	rxTestDeletedPixelRequestsCauseCancelation = 
		rxTestDeletedPixelRequestsCauseCancelation,

	rxTestHigherHTTPPrioritySameDownload = 
		rxTestHigherHTTPPrioritySameDownload,

	rxTestJoiningInFlightDownload = rxTestJoiningInFlightDownload,

	rxTestJoiningJustCompletedDownload = rxTestJoiningJustCompletedDownload,

}

--[[
testCases = {

	rxTestJoiningInFlightDownload = rxTestJoiningInFlightDownload,

}
--]]

return WFRxTestCase.wrapSuite( 'PixelProviderDownloadToFileTaskTest', testCases )
