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

PixelProviderSaveDownloadedTaskTest.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 KSData = require 'KSData'
local KSIdGenerator = require 'KSIdGenerator'
local KSPathUtils = require 'KSPathUtils'
local KSTableUtils = require 'KSTableUtils'
local WFCatMockCatalog = require 'WFCatMockCatalog'
local WFFileUtils = require 'WFFileUtils'
local WFFiniteStateMachine = require 'WFFiniteStateMachine'
local WFRxTestCase = require 'WFRxTestCase'
local WFCloseSupport = require 'WFCloseSupport'

	-- Rx
local Rx = require 'Rx'

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

local PixelPathProviderScheduler = require 'PixelPathProviderScheduler'
local PixelPathProviderRequestTypes = require 'PixelPathProviderRequestTypes'
local PixelPathProviderStateMachine = require 'PixelPathProviderStateMachine'
local PixelProviderSharedBlobCache = require 'PixelProviderSharedBlobCache'

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

local PixelPathProviderRequestTypeNames = PixelPathProviderRequestTypes.names

local testPath = KSPathUtils.getStandardFilePath( 'carouselDocuments' )
local volumeGuid = WFFileUtils.getVolumeGuidForPath( testPath )

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

local function makeSchedulerForTest( rxTest )

	local catalog = WFCatMockCatalog.createFakeWFCatalog( rxTest )

	local scheduler = PixelPathProviderScheduler( catalog )

	rxTest:finally( function() scheduler:close() end )

	return scheduler, catalog

end

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

local function localAssetId( i )

	return i

end

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

local function pixStreamId( i )

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

end

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

local function makeSaveDataStr( i )

	return string.format( "PixelPathProviderStateMachineTest fake data %d", i )

end

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

local function justReturn( ... )

	return ...

end

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

local function getNFromTestInfoStream( testStream, n, filterFunc, 
										maybeTransformFunc )

	local transformTestData = maybeTransformFunc or justReturn

	return testStream:
			where( filterFunc ):
			map( transformTestData ):
			take( n):
			toArray()

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( msg )

	return function( testMsg, ... )

		return testMsg == msg

	end

end

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

local runningSaveFileCount = 0

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

local function makeUniqueSavePath( i )

	runningSaveFileCount = runningSaveFileCount + 1

	local fname = string.format( "PixelPathProviderStateMachineTest%d_%d", 
									runningSaveFileCount, i )

	local realFileDir = KSPathUtils.getStandardFilePath( "temp" )

	return WFFileUtils.ensureFileNameIsUnique( 
						KSPathUtils.child( realFileDir, fname ) )

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 extractPixStreamIdLocalAssetIdState( request )

	local aPixStreamId, aLocalAssetId, _, _, state = unpack( request )

	return { aPixStreamId, aLocalAssetId, state }

end

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

local function addRecordToDocStore( rxTest, catalog, i )

	rxTest:expectSingleValue(

		catalog:docStore():bulkSetDocValue {
			{
				_id = KSIdGenerator.uuidInLowercaseHex(),
				_localId = localAssetId( i ),
				subtype = "image",
				_localOnly = {
					files = {
						original = {
							master = {
								relativePath = string.format( 'local path %s', i ),
								volumeGuid = volumeGuid
							}
						}
					}
				},
				develop = {
					croppedHeight = 100 * i,
					croppedWidth = 100 * i,
					xmpCameraRaw = '<xmp />',
					userOrientation = i % 4,
				},
				rating = i % 5,
			},
		},

		':ignore:' )

end

--------------------------------------------------------------------------------
-- rxTestSaveDownloadedNegative 
-- 
-- 		test a real save
--
--------------------------------------------------------------------------------

local function rxTestSaveDownloadedNegative( rxTest )

	rxTest:setTimeLimit( 5 )
	
	local scheduler, catalog = makeSchedulerForTest( rxTest )

	local SchedulerTestAPI = scheduler:_expose_test_API()

	SchedulerTestAPI.addRunLoopTaskToAllowlist( scheduler, 
												"PixelProviderSaveDownloadedTask" )
	
	for i = 1,PixelPathProviderStateMachine.SaveDownloadRetryCount do

		addRecordToDocStore( rxTest, catalog, i )

	end

	local expectedHappyPathEndState = {

			[ PixelPathProviderStateMachine.SaveDownloadStates.NeedToSaveOriginalForRender ]=
				PixelPathProviderStateMachine.RenderStates.NeedsRender,

			[ PixelPathProviderStateMachine.SaveDownloadStates.NeedToSaveProxyForRender ]=
				PixelPathProviderStateMachine.RenderStates.NeedsRender,

			[ PixelPathProviderStateMachine.SaveDownloadStates.NeedToSaveOriginal ]=
				PixelPathProviderStateMachine.SendDataStates.HaveLocalNegativePath,

			[ PixelPathProviderStateMachine.SaveDownloadStates.NeedToSaveProxy ]=
				PixelPathProviderStateMachine.SendDataStates.HaveLocalNegativePath,

		}

	for _, needSaveNegativeState in ipairs( 
					{
						PixelPathProviderStateMachine.SaveDownloadStates.NeedToSaveOriginalForRender,
						PixelPathProviderStateMachine.SaveDownloadStates.NeedToSaveProxyForRender,
						PixelPathProviderStateMachine.SaveDownloadStates.NeedToSaveOriginal,
						PixelPathProviderStateMachine.SaveDownloadStates.NeedToSaveProxy 
					} ) do

		local needSaveDownloadStates =
				{ WFFiniteStateMachine.makeRetryableStates(
							PixelPathProviderStateMachine.SaveDownloadRetryCount, 
								needSaveNegativeState ) }

		for i = 1, #needSaveDownloadStates do

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

		end

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

			local expectedTestDataTuples = {}

			local expectedErrorStateTuples = {}

			local expectedHaveRenditionPathStateTuples = {}

			local strings = {}

			local ksDatas = {}

			local expectedTestMessage = testWithNegativeBlob and 
											"didPixelCacheSave" or "skipSaveNoBlob"

			local watchNegativeSaves = rxTest:observe( 
											getNFromTestInfoStream(
													scheduler._testData.testInfoStream,
													#needSaveDownloadStates,
													isTestMessage( expectedTestMessage ),
													stripTestMessageAndMakeTuple )
												:map( sortTupleArray ) )

			for i, needSaveDownloadState in ipairs( needSaveDownloadStates ) do

				local ozBinaryType = PixelProviderSharedBlobCache.isProxyDownloadState( 
											needSaveDownloadState ) and
												'proxy' or 'original'
										
				local ozBinarySubtype = ozBinaryType == 'original' and 'master' or '2560'

				if testWithNegativeBlob then

					strings[ i ] = makeSaveDataStr( i )

					ksDatas[ i ] = KSData.make( strings[ i ] )

					scheduler._testData.runLoopCaches.sharedCaches.pendingSaveBlobCache:pushBlob(
						localAssetId( i ), ksDatas[ i ], ozBinaryType, ozBinarySubtype )

				end

				SchedulerTestAPI.setState( scheduler, pixStreamId( i ), localAssetId( i ), 
											needSaveDownloadState )

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

				expectedErrorStateTuples[ #expectedErrorStateTuples + 1 ] = 
					not( testWithNegativeBlob ) and
						{ pixStreamId( i ), localAssetId( i ) } or nil

				expectedHaveRenditionPathStateTuples[ #expectedHaveRenditionPathStateTuples + 1 ] = 
					testWithNegativeBlob and 
							{ pixStreamId( i ), localAssetId( i ) } or nil

			end

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

			SchedulerTestAPI.runScheduledRequests( scheduler )

			watchNegativeSaves:expectData( sortTupleArray( expectedTestDataTuples ) )

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

			rxTest:expectEqual( 
				sortTupleArray( expectedErrorStateTuples ),
				sortTupleArray( 
					scheduler:_mapAllRequestsWithState( 
								PixelPathProviderStateMachine.SendDataStates.SendPermanentError,
								takeFirstN( 2 ) ) ) )

			rxTest:expectEqual( 
				sortTupleArray( expectedHaveRenditionPathStateTuples ),
				sortTupleArray( 
					scheduler:_mapAllRequestsWithState( 
								expectedHappyPathEndState[ needSaveNegativeState ],
								takeFirstN( 2 ) ) ) )

			if testWithNegativeBlob then

				-- check to make sure all the render blobs are cleaned up
				for i, needSaveDownloadState in ipairs( needSaveDownloadStates ) do

					local ozBinaryType = PixelProviderSharedBlobCache.isProxyDownloadState( 
												needSaveDownloadState ) and
													'proxy' or 'original'
										
					local ozBinarySubtype = ozBinaryType == 'original' and 'master' or '2560'

					rxTest:expectEqual( nil,
						scheduler._testData.runLoopCaches.sharedCaches.pendingSaveBlobCache:peekBlob(
							localAssetId( i ), ozBinaryType, ozBinarySubtype ) )

					local path = unpack(
						scheduler._testData.runLoopCaches.sharedCaches.localBinaryPathCache:popCacheItem( 
											pixStreamId( i ), localAssetId( i ) ) or {} )

					rxTest:expectEqual( true, WFCloseSupport.hasBeenOrWillBeClosed( ksDatas[ i ] ) )

					rxTest:log( "Making sure saved negative exists at " .. path )

					rxTest:expectEqual( "file", WFFileUtils.exists( path ) )

					local f = io.open( path, 'r' )

					local contents = f:read( "*all" )

					f:close()

					WFFileUtils.delete( path )

					rxTest:expectEqual( strings[ i ], contents )

				end

			end

		end

	end

end


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

local function makeFakeSaveThread( fakeSaveResultsStreams )

	local i

	return {

		savePixels = 

			function( _, _, blob ) 

				blob:close()

				i = ( i or 0 ) + 1

				return fakeSaveResultsStreams[ i ] or Rx.Subject()
				
			end,

		close = function() end,

	}

end

--------------------------------------------------------------------------------
-- rxTestSaveDownloadedNegativeFail 
-- 
-- 	test a save failure (failures are faked by mocking the save thread)
--
--------------------------------------------------------------------------------

local function rxTestSaveDownloadedNegativeFail( rxTest )

	local scheduler = makeSchedulerForTest( rxTest )

	local SchedulerTestAPI = scheduler:_expose_test_API()

	SchedulerTestAPI.addRunLoopTaskToAllowlist( scheduler, 
												"PixelProviderSaveDownloadedTask" )

	for _, needSaveNegativeState in ipairs( 
				{
					PixelPathProviderStateMachine.SaveDownloadStates.NeedToSaveOriginalForRender,
					PixelPathProviderStateMachine.SaveDownloadStates.NeedToSaveProxyForRender,
					PixelPathProviderStateMachine.SaveDownloadStates.NeedToSaveOriginal,
					PixelPathProviderStateMachine.SaveDownloadStates.NeedToSaveProxy 
				} ) do

		local needSaveDownloadStates =
				{ WFFiniteStateMachine.makeRetryableStates(
							PixelPathProviderStateMachine.SaveDownloadRetryCount, 
							needSaveNegativeState ) }

		local k 

		local fakeSaveResultsStreams = {}

		local expectedTestDataTuples = {}

		for i = 1, #needSaveDownloadStates do

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

			for _ = i, #needSaveDownloadStates do

				k = ( k or 0 ) + 1

				expectedTestDataTuples[ k ] = { pixStreamId( i ), localAssetId( i ) }

				fakeSaveResultsStreams[ k ] = Rx.Subject()

			end

		end

		scheduler._testData.runLoopTaskStates.PixelProviderSaveDownloadedTask.saveToDiskThread =
			makeFakeSaveThread( fakeSaveResultsStreams )

		local expectedSendPermanentErrorTuples = {}

		local strings = {}

		local ksDatas = {}

		local paths = {}

		local watchNegativeSaves = rxTest:observe( 
										getNFromTestInfoStream(
												scheduler._testData.testInfoStream,
												#fakeSaveResultsStreams,
												isTestMessage( 'saveFailed' ),
												stripTestMessageAndMakeTuple )
											:map( sortTupleArray ) )

		for i, needSaveDownloadState in ipairs( needSaveDownloadStates ) do

			local ozBinaryType = PixelProviderSharedBlobCache.isProxyDownloadState( 
										needSaveDownloadState ) and
											'proxy' or 'original'
									
			local ozBinarySubtype = ozBinaryType == 'original' and 'master' or '2560'

			strings[ i ] = makeSaveDataStr( i )

			ksDatas[ i ] = KSData.make( strings[ i ] )

			scheduler._testData.runLoopCaches.sharedCaches.pendingSaveBlobCache:pushBlob(
				localAssetId( i ), ksDatas[ i ], ozBinaryType, ozBinarySubtype )

			paths[ i ] = makeUniqueSavePath( i )

			rxTest:log( "Making sure no saved negative exists at " .. tostring( paths[ i ] ) )

			rxTest:expectEqual( false, WFFileUtils.exists( paths[ i ] ) )

			SchedulerTestAPI.setState( scheduler, pixStreamId( i ), localAssetId( i ), 
										needSaveDownloadState )

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

		end

		rxTest:log( "Scheduling a test expecting saveFailed" )

		SchedulerTestAPI.runScheduledRequests( scheduler )

		for i = 1, #fakeSaveResultsStreams do

			fakeSaveResultsStreams[ i ]:onError( string.format( "fale wail #%s!", i ) )

		end

		watchNegativeSaves:expectData( sortTupleArray( expectedTestDataTuples ) )

		rxTest:log( "Finishing a test expecting saveFailed" )

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

		rxTest:expectEqual( 
			sortTupleArray( expectedSendPermanentErrorTuples ),
			sortTupleArray( 
				scheduler:_mapAllRequestsWithState( 
							PixelPathProviderStateMachine.SendDataStates.SendPermanentError,
							takeFirstN( 2 ) ) ) )

		-- check to make sure all the render blobs are cleaned up
		for i = 1, #needSaveDownloadStates do

			local digest = string.format( "digest%s", i )

			local renderSize = 100 + i

			rxTest:expectEqual( nil,
				scheduler._testData.runLoopCaches.sharedCaches.pendingSaveBlobCache:peekBlob(
					localAssetId( i ), 'rendition', renderSize, digest ) )

			rxTest:expectEqual( true, WFCloseSupport.hasBeenOrWillBeClosed( ksDatas[ i ] ) )

		end

	end

end

--------------------------------------------------------------------------------
-- rxTestQuitPendingSaveThread
--
-- 	test quitting the scheduler main loop before the save thread finished (save
-- 	thread is mocked)
--------------------------------------------------------------------------------

local function rxTestQuitPendingSaveThread( rxTest )

	-- quit the task before we've finished saving the blob

	local catalog = WFCatMockCatalog.createFakeWFCatalog( rxTest )

	local scheduler = PixelPathProviderScheduler( catalog )

	local SchedulerTestAPI = scheduler:_expose_test_API()

	SchedulerTestAPI.addRunLoopTaskToAllowlist( scheduler, 
												"PixelProviderSaveDownloadedTask" )

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

	local fakeSaveResultsStreams = { Rx.Subject() }

	scheduler._testData.runLoopTaskStates.PixelProviderSaveDownloadedTask.saveToDiskThread =
		makeFakeSaveThread( fakeSaveResultsStreams )

	local strings = {}

	local ksDatas = {}

	strings[ 1 ] = makeSaveDataStr( 1 )

	ksDatas[ 1 ] = KSData.make( strings[ 1 ] )

	scheduler._testData.runLoopCaches.sharedCaches.pendingSaveBlobCache:pushBlob(
		localAssetId( 1 ), ksDatas[ 1 ], 'original', 'master' )

	SchedulerTestAPI.setState( scheduler, pixStreamId( 1 ), localAssetId( 1 ), 
								PixelPathProviderStateMachine.SaveDownloadStates.NeedToSaveOriginal )

	rxTest:expectEqual( true,
		scheduler._testData.runLoopCaches.sharedCaches.pendingSaveBlobCache:peekBlob(
			localAssetId( 1 ), 'original', 'master' ) ~= nil )

	SchedulerTestAPI.runScheduledRequests( scheduler )

	scheduler:close()

	-- check to make sure all the render blobs are cleaned up
	rxTest:expectEqual( nil,
		scheduler._testData.runLoopCaches.sharedCaches.pendingSaveBlobCache:peekBlob(
			localAssetId( 1 ), 'original', 'master' ) )

	fakeSaveResultsStreams[ 1 ]:onCompleted()

	rxTest:expectEqual( false, WFCloseSupport.hasBeenOrWillBeClosed( ksDatas[ 1 ] ) )

end

--------------------------------------------------------------------------------
-- rxTestPurgeAbandoned
--
-- test that abandoned blobs are purged
--------------------------------------------------------------------------------

local function rxTestPurgeAbandoned( rxTest )

	local scheduler = makeSchedulerForTest( rxTest )

	local SchedulerTestAPI = scheduler:_expose_test_API()

	SchedulerTestAPI.addRunLoopTaskToAllowlist( scheduler, 
												"PixelProviderSaveDownloadedTask" )

	SchedulerTestAPI.runScheduledRequests( scheduler )

	local strings = {}

	local ksDatas = {}

	strings[ 1 ] = makeSaveDataStr( 1 )

	ksDatas[ 1 ] = KSData.make( strings[ 1 ] )

	scheduler._testData.runLoopCaches.sharedCaches.pendingSaveBlobCache:pushBlob(
		localAssetId( 1 ), ksDatas[ 1 ], 'original', 'master' )

	strings[ 2 ] = makeSaveDataStr( 2 )

	ksDatas[ 2 ] = KSData.make( strings[ 2 ] )

	scheduler._testData.runLoopCaches.sharedCaches.pendingSaveBlobCache:pushBlob(
		localAssetId( 2 ), ksDatas[ 2 ], 'proxy', '2560' )

	rxTest:expectEqual( true,
		scheduler._testData.runLoopCaches.sharedCaches.pendingSaveBlobCache:peekBlob(
			localAssetId( 1 ), 'original', 'master' ) ~= nil )

	rxTest:expectEqual( true,
		scheduler._testData.runLoopCaches.sharedCaches.pendingSaveBlobCache:peekBlob(
			localAssetId( 2 ), 'proxy', '2560' ) ~= nil )

	SchedulerTestAPI.runScheduledRequests( scheduler )

	-- check to make sure all the render blobs are cleaned up

	rxTest:expectEqual( nil,
		scheduler._testData.runLoopCaches.sharedCaches.pendingSaveBlobCache:peekBlob(
			localAssetId( 1 ), 'original', 'master' ) )

	rxTest:expectEqual( nil,
		scheduler._testData.runLoopCaches.sharedCaches.pendingSaveBlobCache:peekBlob(
			localAssetId( 2 ), 'proxy', '2560' ) )

	rxTest:expectEqual( true, WFCloseSupport.hasBeenOrWillBeClosed( ksDatas[ 1 ] ) )

	rxTest:expectEqual( true, WFCloseSupport.hasBeenOrWillBeClosed( ksDatas[ 2 ] ) )

end

--------------------------------------------------------------------------------
-- rxTestJoinActiveSave
-- 
-- Test that all appropriate requests join an active save (save thread is
-- mocked)
--------------------------------------------------------------------------------

local function isOzBinaryTypeState( ozBinaryType, state )

	return string.find( string.lower( state ), 
						string.lower( ozBinaryType ) ) ~= nil

end

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

local function rxTestJoinActiveSave( rxTest )

	local scheduler, catalog  = makeSchedulerForTest( rxTest )

	local SchedulerTestAPI = scheduler:_expose_test_API()

	SchedulerTestAPI.addRunLoopTaskToAllowlist( scheduler, 
												"PixelProviderSaveDownloadedTask" )

	-- run the runLoop once to bootstrap test data
	SchedulerTestAPI.runScheduledRequests( scheduler )

	addRecordToDocStore( rxTest, catalog, 1 )

	for _, ozBinaryType in ipairs( { 'original', 'proxy' } ) do

		local ozBinarySubtype = ozBinaryType == 'proxy' and '2650' or 'master'

		local needToSaveBinaryStates =
				ozBinaryType == 'proxy' and 
					PixelProviderSharedBlobCache.needToSaveProxyStates or
					PixelProviderSharedBlobCache.needToSaveOriginalStates

		for _, needToSaveBinaryState in ipairs( needToSaveBinaryStates ) do

			local expectedJoinedSaveTuples = {}

			local expectedSameStateTuples = {}

			-- set up a pending save request

			local fakeSaveResultsStreams = { Rx.Subject() }

			scheduler._testData.runLoopTaskStates.PixelProviderSaveDownloadedTask.saveToDiskThread =
				makeFakeSaveThread( fakeSaveResultsStreams )

			-- suspend runLoop to make sure it doesn't advance our reqeust
			-- state until we want it to do so

			SchedulerTestAPI.suspendRunLoop( scheduler )

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

			local ksData = KSData.make( makeSaveDataStr( 1 ) )

			scheduler._testData.runLoopCaches.sharedCaches.pendingSaveBlobCache:pushBlob(
				localAssetId( 1 ), ksData, ozBinaryType, ozBinarySubtype )

			SchedulerTestAPI.setState( scheduler, pixStreamId( 1 ), localAssetId( 1 ), 
										needToSaveBinaryState )
					
			rxTest:logf( "expecting that %s.%s.%s will join save", 
					 				pixStreamId( 1 ), localAssetId( 1 ), needToSaveBinaryState )

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

			-- resume runLoop long enough to set up the save
			
			SchedulerTestAPI.resumeRunLoop( scheduler )
			
			SchedulerTestAPI.runScheduledRequests( scheduler )

			-- suspend runLoop as we add the rest of these requests
			SchedulerTestAPI.suspendRunLoop( scheduler )

			-- add a bunch requests that should join this one
			for i, state in ipairs( PixelProviderSharedBlobCache.fulfilledBySuccessfulNegativeSaveStates ) do

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

				SchedulerTestAPI.setState( scheduler, pixStreamId( i + 1 ), localAssetId( 1 ), state ) 

				if isOzBinaryTypeState( ozBinaryType, state ) then

					-- these are "eventually needs saved negative states" that
					-- we expect to join this one.
					
					rxTest:logf( "expecting that %s.%s.%s will join save", 
					 				pixStreamId( i + 1 ), localAssetId( 1 ), state )

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

				else

					-- these are "eventually needs saved negative states" that
					-- are a different negative type, so we don't expect the state
					-- to change

					expectedSameStateTuples[ #expectedSameStateTuples + 1 ] = { pixStreamId( i + 1 ), localAssetId( 1 ), state }

				end

			end

			local watchNegativeSaves = rxTest:observe( 
											getNFromTestInfoStream(
													scheduler._testData.testInfoStream,
													#expectedJoinedSaveTuples,
													isTestMessage( "didPixelCacheSave" ),
													stripTestMessageAndMakeTuple )
												:map( sortTupleArray ) )


			fakeSaveResultsStreams[ 1 ]:onCompleted()

			rxTest:log( "Waiting to process save for " .. needToSaveBinaryState )

			watchNegativeSaves:expectData( sortTupleArray( expectedJoinedSaveTuples ) )

			rxTest:log( "Finished process save for " .. needToSaveBinaryState )

			rxTest:expectEqual( true, WFCloseSupport.hasBeenOrWillBeClosed( ksData ) )

			rxTest:expectEqual( 
				{},
				sortTupleArray( 
					scheduler:_mapAllRequestsWithState( 
								PixelPathProviderStateMachine.SendDataStates.SendPermanentError,
								takeFirstN( 1 ) ) ) )

			rxTest:expectEqual( 
				sortTupleArray( expectedJoinedSaveTuples ),
				sortTupleArray( 
					scheduler:_mapAllRequestsWithStates( 
								{ PixelPathProviderStateMachine.SendDataStates.HaveLocalNegativePath,
									PixelPathProviderStateMachine.RenderStates.NeedsRender,
									ozBinaryType ~= 'proxy' and
										PixelPathProviderStateMachine.SendDataStates.HaveOriginalForBestPossibleLocalNegativePath or
										PixelPathProviderStateMachine.SendDataStates.HaveProxyForBestPossibleLocalNegativePath
									},
								takeFirstN( 2 ) ) ) )

			rxTest:expectEqual( 
				sortTupleArray( expectedSameStateTuples ),
				sortTupleArray( 
					scheduler:_mapAllRequestsWithStates( 
								PixelProviderSharedBlobCache.fulfilledBySuccessfulNegativeSaveStates,
								extractPixStreamIdLocalAssetIdState ) ) )

			scheduler:abandonAllLocalAssetIdRequests( localAssetId( 1 ) )

			-- resume runLoop, this test is done, and suspend/resume nest
			SchedulerTestAPI.resumeRunLoop( scheduler )

		end

	end

end

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

return WFRxTestCase.wrapSuite {

	rxTestSaveDownloadedNegative = rxTestSaveDownloadedNegative,

	rxTestSaveDownloadedNegativeFail = rxTestSaveDownloadedNegativeFail,

	rxTestQuitPendingSaveThread = rxTestQuitPendingSaveThread,

	rxTestPurgeAbandoned = rxTestPurgeAbandoned,

	rxTestJoinActiveSave = rxTestJoinActiveSave,

}
