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

WFCatMetadataPusherTest.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.

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

	-- RxLua
local Rx = require 'Rx'

	-- WF.core.crypto
local WFCrypto = require 'WFCrypto'

	-- WF.core.lua
local KSData = require 'KSData'

	-- WF.core.os
local WFFileUtils = require 'WFFileUtils'

	-- WF.core.self_test
local WFRxScheduler = require 'WFRxScheduler'
local WFRxTestCase = require 'WFRxTestCase'

	-- WF.core.utils
local KSDate = require 'KSDate'
local KSIdGenerator = require 'KSIdGenerator'
local KSPathUtils = require 'KSPathUtils'
local KSTableUtils = require 'KSTableUtils'
local WFArguments = require 'WFArguments'
local WFPlatformInfo = require 'WFPlatformInfo'
local WFTableArchiving = require 'WFTableArchiving'
local WFTableMerge = require 'WFTableMerge'

	-- WF.oz_client.managed_catalog
local WFCatAssetAux = require 'WFCatAssetAux'
local WFCatAssetTestUtils = require 'WFCatAssetTestUtils'
local WFCatMockCatalog = require 'WFCatMockCatalog'
local WFOzDocUtils = require 'WFOzDocUtils'

	-- WF.oz_client.managed_catalog.sync
local WFCatBinaryUploader = require 'WFCatBinaryUploader'
local WFCatConflictResolver = require 'WFCatConflictResolver'
local WFCatErrorHandler = require 'WFCatErrorHandler'
local WFCatMetadataPuller = require 'WFCatMetadataPuller'
local WFCatMetadataPusher = require 'WFCatMetadataPusher'
local WFCatSpecMock = require 'WFCatSpecMock'
local WFCatTestSession = require 'WFCatTestSession'

	-- WF.sqlite.index_store
local WFRxIndexStoreSchema = require 'WFRxIndexStoreSchema'

	-- WF.web.http_client
local WFMockHttpRequest = require 'WFMockHttpRequest'

	-- WF.web.json
local KSJsonDecoder = require 'KSJsonDecoder'
local KSJsonEncoder = require 'KSJsonEncoder'

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

local check = assert( WFArguments.check )
local kUserId = KSIdGenerator.uuidInLowercaseHex()
	-- Helpful bash one-liner. Put a new uuid on the clipboard (mac cmdline):
	-- uuidgen  | perl -p -e 's/-//g;s/[A-Z]/lc($&)/ge;' | pbcopy

local kPurgeDate = '2019-11-05T09:07:00.002Z'
local kUserCreated = '2019-09-05T09:07:00.002Z'

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

local function albumAssetsForFilteredAssetsQueryFn( catalog, assetIds )

	return catalog:indexStore():queryOnce {
				sql = [[SELECT album_asset_v2.docId, album_asset_v2.assetId FROM CARRAY( ?, ?, 'char*' ) as ai JOIN album_asset_v2 ON
							ai.value = album_asset_v2.assetId]],
				assetIds,
				#assetIds,
			}
end

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

local function prefetchReqResp( catalogId )

	return {
		url = "/v2/catalogs/" .. catalogId .. "/albums?subtype=collection%3Bcollection_set",
		response = [[{
			"base": "https://photos-stage.adobe.io/v2/catalogs/]]..catalogId ..[[/",
			"resources": []
			}]],
		headers = {
			status_code = '200',
		},
		options = {
			returnHttpErrorsAsData = true,
		}
	}
end

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

local function catalogPrefetchReqResp( catalogId )

	return {
		url = "/v2/catalogs/" .. catalogId,
		response = [[{
			"base": "https://photos-stage.adobe.io/v2/",
			"id": "]] .. catalogId .. [[",
			"type": "catalog"
			}]],
		headers = {
			status_code = '200',
		},
	}
end

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

local function httpMockDataTransformation( data, requestInfoData )

	data = KSJsonDecoder.decodeWebSafeJson( data )
	requestInfoData = KSJsonDecoder.decodeWebSafeJson( requestInfoData )

	return data, requestInfoData

end

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

local function stripUpdateFeedLinkNoise( changedDocsResult )

	local updates = KSTableUtils.derefOpt( changedDocsResult, 'updates' )

	if not updates then
		return changedDocsResult
	end

	local sortedUpdates = {}

	for i, originalUpdateDoc in ipairs( updates ) do

		local updateDoc = KSTableUtils.deepCopy( originalUpdateDoc )

		if updateDoc.album then
			updateDoc.album.links = nil
		end

		if updateDoc.asset then
			updateDoc.asset.links = nil
		end

		updateDoc.created = nil
		updateDoc.links = nil
		updateDoc.updated = nil
		updateDoc.storage = nil

		-- Remove the location as Oz does geo-location decoding which we sometimes
		-- catch as part of the sync feed changes

		if updateDoc.payload and updateDoc.payload.location then
			updateDoc.payload.location = nil
		end

		sortedUpdates[ i ] = updateDoc

	end

	table.sort( sortedUpdates,
		function( a, b )
			return a.type < b.type
		end )

	return changedDocsResult, sortedUpdates

end

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

local function remapSingleDocument( doc, stripMore, stripRevIdAndChangeCounter )

	doc._localId = nil

	if stripMore then
		doc._id = nil
	end

	if stripMore or stripRevIdAndChangeCounter then
		doc._revId = nil
		doc._changeCounter = nil
	end

	if doc.userUpdated then
		doc.userUpdated = ">>userUpdated"
	end

	if doc.userCreated then
		doc.userCreated = ">>userCreated"
	end

	if doc.purgeDate then
		doc.purgeDate = ">>purgeDate"
	end

	local spm = doc._serverProvidedMetadata

	if spm then
		if spm.added_by then
			spm.added_by = ">>added"
		end
		if spm.base then
			spm.base = ">>base"
		end
		if spm.created then
			spm.created = ">>created"
		end
		if spm.created_by then
			spm.created_by = ">>created_by"
		end
		if spm.deleted_by then
			spm.deleted_by = ">>deleted_by"
		end
		if spm.id then
			spm.id = ">>id"
		end
		if spm.links then
			spm.links = ">>links"
		end
		if spm.link_templates then
			spm.link_templates = ">>link_templates"
		end
		if spm.revision_ids then
			spm.revision_ids = ">>revision_ids"
		end
		if spm.storage then
			spm.storage = ">>storage"
		end
		if spm.updated then
			spm.updated = ">>updated"
		end
		if spm.new_id then
			spm.new_id = ">>new_id"
		end
	end

end

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

local function remapChangedDocsFeed( changedDocsResult, mapFn )

	mapFn = mapFn or remapSingleDocument

	for i, doc in ipairs( changedDocsResult ) do

		mapFn( doc, true )

	end

	table.sort( changedDocsResult,
		function( a, b )
			return ( a.type or "" ) < ( b.type or "" )
		end )

	return changedDocsResult

end

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

local function remapChangedDocsFeedForMockTests( docs )

	local function strip( doc )
		doc._localId = nil
		doc._changeCounter = nil
		if doc.type == 'catalog' then
			doc._revId = nil
		end
	end

	return remapChangedDocsFeed( docs, strip )

end

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

local function sortByDocId( docs )

	table.sort( docs, function( a, b )
		return ( a and a._id or "" ) < ( b and b._id or "" )
	end )

	return docs

end

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

local function remapChangedDocsFeedWithDocId( currentRevisionsResult )

	for i, doc in ipairs( currentRevisionsResult ) do

		remapSingleDocument( doc, false, true )

	end

	return sortByDocId( currentRevisionsResult )

end

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

local function stripOzChangesFeed( ozChangesFeed )

	if type( ozChangesFeed ) == 'table' then

		ozChangesFeed.links = nil

		if type( ozChangesFeed.updates ) == 'table' then

			for i, update in ipairs( ozChangesFeed.updates ) do

				if update.deleted_by then
					update.deleted_by = ">>deleted_by"
				end

				if update.id then
					update.id = ">>id"
				end

				if update.updated then
					update.updated = ">>updated"
				end

				if update.links then
					update.links = ">>links"
				end

				if update.created then
					update.created = ">>created"
				end

				if update.original then
					update.original = ">>original"
				end

				if update.new_id then
					update.new_id = ">>new_id"
				end

				if update.payload then
					remapSingleDocument( update.payload )
				end

				if update.revision_ids then
					update.revision_ids = ">>revision_ids"
				end

			end

		end

	end

	return ozChangesFeed

end

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

local function observeHasServerProvidedMetadata( rxTest, catalog, docId )

	rxTest:logf( "Observing %s for Server Provided Metadata", docId )

	return rxTest:observe( catalog:docStore():ongoingDocValue( docId )
		:map( function( docValue )

			local success, strDocValue = pcall( WFTableArchiving.debugTableToString, docValue )

			if success then
				rxTest:logf( "New Doc Value: \n\n%s\n\n", strDocValue )
			end

			local receivedUpdateFromOz = docValue._serverProvidedMetadata ~= nil

			return receivedUpdateFromOz

		end )
	)

end

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

local function observeIsDeletedOnOz( rxTest, catalog, docId )

	rxTest:logf( "Waiting for %s to be deleted on Oz", docId )

	return rxTest:observe( catalog:docStore():ongoingDocValue( docId )
		:map( function( docValue )

			local success, strDocValue = pcall( WFTableArchiving.debugTableToString, docValue )

			if success then
				rxTest:logf( "New Doc Value: \n\n%s\n\n", strDocValue )
			end

			local isDeleted = not not ( docValue._deleted or docValue._empty )
			local isLabeledOz = not not ( docValue._labels and docValue._labels.oz )

			return isDeleted and isLabeledOz

		end )
	)

end

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

local function observeIsOnOz( rxTest, catalog, docId )

	rxTest:logf( "Waiting for %s to exist on Oz", docId )

	return rxTest:observe( catalog:docStore():ongoingDocValue( docId )
		:map( function( docValue )

			local success, strDocValue = pcall( WFTableArchiving.debugTableToString, docValue )

			if success then
				rxTest:logf( "New Doc Value: \n\n%s\n\n", strDocValue )
			end

			local isLabeledOz = not not ( docValue._labels and docValue._labels.oz )

			return isLabeledOz

		end )
	)

end

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

--[[

	\brief Generate mock POST request

	\param type (string) - either "catalog" or "space"
	\param catalogOrSpaceId (string) - guid id for the catalog or space
	\para updates (table) - array of key/value tables describing updates expected to be sent to Oz
	\para fail (table) - optional error information

	\return table

]]
local function pusherMockReqResp( type, catalogOrSpaceId, updates, fail )

	local reqMock = {

		dataTransformation = httpMockDataTransformation,

		data = KSJsonEncoder.encodeOzUploadJson( updates ),

		options = {
			bypassRateLimiter = true,
		},

		options = {
			requestHeaders = {
				{
					header = "Content-Type",
					value = "application/json",
				},
				{
					header = "Accept",
					value = "application/json",
				},
			},
			requestMethod = "POST",
			bypassRateLimiter = true,
		},

	}

	if type == 'space' then
		reqMock.url = string.format( '/v2/%ss/updates', type )
	else
		reqMock.url = string.format( '/v2/%ss/%s/updates', type, catalogOrSpaceId )
	end

	if fail then
		reqMock.fail = fail
	else
		reqMock.headers = {
			status_code = '202',
		}
	end

	return reqMock

end

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

--[[

	\brief Generate mock cat puller response from Oz

	\param type (string) - either "catalog" or "space"
	\param catalogOrSpaceId (string) - guid id for the catalog or space
	\para updates (table) - array of key/value tables describing updates expected to be returned from Oz

	\return table

]]
local function pullerMockReqResp( type, catalogOrSpaceId, updates, includeUpdatedSinceParam )

	local baseUrl = string.format( '/v2/%ss/%s/', type, catalogOrSpaceId )
	local updatesUrl = baseUrl .. 'updates'
	local nextLink = updatesUrl .. '?updated_since=2023-04-27T14%3A37%3A26.880138Z'

	-- mocked calls with are made with the next link with updated_since=...
	if #updates == 0 or includeUpdatedSinceParam then
		updatesUrl = nextLink
	end

	local response = {

		base = 'https://photos-stage.adobe.io' .. baseUrl,
		updates = updates,
		links = { next = { href = nextLink } },

	}

	return {

		url = updatesUrl,

		response = KSJsonEncoder.encodeWebSafeJson( response ),

		headers = {
			status_code = '200',
		},

		options = {
			bypassRateLimiter = true,
		},

	}

end

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

local function addBaseToLinks( links, base )

	for k, v in pairs(links) do
		if k == 'href' then
			links[k] = base .. v
		elseif type( v ) == 'table' then
			addBaseToLinks( v, base )
		end
	end

	return links

end

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

--[[

	\brief Generate mocks, docStore values and expected docStore values after syncing with Oz

	\param type (string) - either "catalog" or "space"
	\param catalogOrSpaceId (string) - guid id for the catalog or space
	\para options (table) - optional options to override default behavior

	\return table

]]
local function generateDocStoreMocks( type, catalogOrSpaceId, options )

	options = options or {}
	if options.excludeAlbum or options.excludeAsset then
		options.excludeAlbumAsset = true
	end

	local baseUrl = string.format( 'https://photos-stage.adobe.io/v2/%ss/%s/', type, catalogOrSpaceId )

	local assetDocId = KSIdGenerator.uuidInLowercaseHex()
	local assetRevisionId = KSIdGenerator.uuidInLowercaseHex()

	local albumDocId = KSIdGenerator.uuidInLowercaseHex()
	local albumRevisionId = KSIdGenerator.uuidInLowercaseHex()

	local albumAssetDocId = KSIdGenerator.uuidInLowercaseHex()
	local albumAssetRevisionId = KSIdGenerator.uuidInLowercaseHex()

	local assetPayload = {
		location = { altitude = 377.9, latitude = 45.951733, longitude = -93.600107 },
		captureDate = "2015-05-17T15:50:12",
		xmp = { dc = { description = "Rum River State Forest, near Onamia, Minnesota", rights = "test copyright string" } },
		importSource = { originalWidth = 1068, originalHeight = 1600, importTimestamp = "2015-05-26T17:34:23-07:00" },
	}

	if options.customDescription then
		assetPayload.xmp.dc.description = options.customDescription
	end

	local albumPayload = {
		name = "Sync me",
		cover = { id = assetDocId },
	}

	local albumAssetPayload = { cover =true, order = "V" }

	local result = {

		pusherUpdates = {

			{
				type = "asset",
				action = "create",
				revision_id = assetRevisionId,
				id = assetDocId,
				payload = assetPayload,
				subtype = "image",
			},

			{
				album = { id = albumDocId },
				type = "album_asset",
				action = "create",
				revision_id = albumAssetRevisionId,
				id = albumAssetDocId,
				payload = albumAssetPayload,
				asset = { id = assetDocId },
			},

			{
				type = "album",
				action = "create",
				revision_id = albumRevisionId,
				id = albumDocId,
				payload = albumPayload,
				subtype = "collection",
			},

		},

		pullerUpdates = {

			{
				id = assetDocId,
				type = "asset",
				updated = "2023-04-27T16:01:10.141012Z",
				subtype = "image",
				created = "2023-04-27T16:01:10.141012Z",
				revision_ids = {
					assetRevisionId,
				},
				payload = assetPayload,
				links = {
					self = {
						href = string.format( "assets/%s", assetDocId ),
					},
					[ "/rels/proxy_create" ] = {
						href = string.format( "assets/%s/revisions/{revision_id}/proxies/{proxy_id}", assetDocId ),
						templated = true,
					},
					[ "/rels/master_create" ] = {
						href = string.format( "assets/%s/revisions/{revision_id}/master", assetDocId ),
						templated = true,
					},
				},
			},

			{
				id = albumAssetDocId,
				type = "album_asset",
				created = "2023-04-27T16:01:12.141012Z",
				updated = "2023-04-27T16:01:12.141012Z",
				revision_ids = {
					albumAssetRevisionId,
				},
				payload = albumAssetPayload,
				links = {
					self = {
						href = string.format( "albums/%s/assets/%s", albumDocId, assetDocId ),
					}
				},
				album = {
					id = albumDocId,
					links = {
						self = {
							href = string.format( "albums/%s", albumDocId ),
						}
					}
				},
				asset = {
					id = assetDocId,
					links = {
						self = {
							href = string.format( "assets/%s", assetDocId ),
						}
					},
				},
			},

			{
				id = albumDocId,
				type = "album",
				subtype = "collection",
				created = "2023-04-27T16:01:11.141012Z",
				updated = "2023-04-27T16:01:11.141012Z",
				revision_ids = {
					albumRevisionId,
				},
				payload = albumPayload,
				links = {
					self = {
						href = "albums/5208067cf2b8489fa133158d14a5b905"
					},
					[ "/rels/web_album" ] = {
						href = string.format( "https://stage.adobelr.com/libraries/%s/albums/%s", catalogOrSpaceId, albumDocId ),
					},
					[ "/rels/album_assets" ] = {
						href = "albums/5208067cf2b8489fa133158d14a5b905/assets?embed=asset"
					},
					[ "/rels/cover_asset" ] = {
						href = string.format( "assets/%s", assetDocId ),
					},
				},
			},

		},

		docStoreValues = {

			KSTableUtils.merge( {},
				{
					_id = assetDocId,
					_newRevId = assetRevisionId,
					subtype = "image",
					type = "asset",
				}, assetPayload
			),

			KSTableUtils.merge( {},
				{
					_id = albumAssetDocId,
					_newRevId = albumAssetRevisionId,
					album = {
						id = albumDocId,
					},
					asset = {
						id = assetDocId,
					},
					type = "album_asset",
				}, albumAssetPayload
			),

			KSTableUtils.merge( {},
				{
					_id = albumDocId,
					_newRevId = albumRevisionId,
					subtype = "collection",
					type = "album",
				}, albumPayload
			),

		},

	}

	local assetExpectedValueIndex = 3

	if options.excludeAlbum then
		assetExpectedValueIndex = assetExpectedValueIndex - 1
		table.remove( result.pusherUpdates )
		table.remove( result.pullerUpdates )
		table.remove( result.docStoreValues )
	else
		result.albumDocId = albumDocId
		result.albumRevisionId = albumRevisionId
	end

	if options.excludeAlbumAsset then
		assetExpectedValueIndex = assetExpectedValueIndex - 1
		table.remove( result.pusherUpdates, 2 )
		table.remove( result.pullerUpdates, 2 )
		table.remove( result.docStoreValues, 2 )
	else
		result.albumAssetDocId = albumAssetDocId
		result.albumAssetRevisionId = albumAssetRevisionId
	end

	if options.excludeAsset then
		table.remove( result.pusherUpdates, 1 )
		table.remove( result.pullerUpdates, 1 )
		table.remove( result.docStoreValues, 1 )
	else
		result.assetDocId = assetDocId
		result.assetRevisionId = assetRevisionId
	end

	if type == 'space' then

		for i, v in ipairs( result.pusherUpdates ) do
			v.space_id = catalogOrSpaceId
			result.docStoreValues[i]._spaceId = catalogOrSpaceId
		end

	end

	local expectedDocStoreValues = { }

	for i, v in ipairs( result.docStoreValues ) do

		local expectedValue = KSTableUtils.merge( {}, v,
			{
				_labels = {
					oz = true,
				},
				_serverProvidedMetadata = {
					base = baseUrl,
					created = result.pullerUpdates[i].created,
					updated = result.pullerUpdates[i].updated,
					links = addBaseToLinks( result.pullerUpdates[i].links, baseUrl ),
					revision_ids = result.pullerUpdates[i].revision_ids,
				},
			}
		)

		expectedValue._revId = expectedValue._newRevId
		expectedValue._newRevId = nil

		table.insert( expectedDocStoreValues, expectedValue )

	end

	if type == 'catalog' then
		table.insert( expectedDocStoreValues,
			{
				_id = catalogOrSpaceId,
				_labels = {
					oz = true,
				},
				_serverProvidedMetadata = {
					base = "https://photos-stage.adobe.io/v2/",
				},
				type = "catalog",
			}
		)
	end

	result.expectedDocStoreValues = remapChangedDocsFeedForMockTests( expectedDocStoreValues )
	result.expectedDocStoreValues.continueFrom = #expectedDocStoreValues * 2
	if type ~= 'catalog' then
		result.expectedDocStoreValues.continueFrom = result.expectedDocStoreValues.continueFrom + 1
	end

	if options.includeAssetUpdate then

		local newRevisionId = KSIdGenerator.uuidInLowercaseHex()

		result.assetUpdateMock = {
			type = "asset",
			action = "update",
			revision_id = newRevisionId,
			id = assetDocId,
			payload = { xmp = { dc = { description = "Rum River State Forest, not near Winona, Minnesota" } } },
			if_match = assetRevisionId,
		}

		if type == 'space' then
			result.assetUpdateMock.space_id = catalogOrSpaceId
		end

		result.assetUpdateOverlay = {
			_id = assetDocId,
			_newRevId = newRevisionId,
			xmp = {
				dc = {
					description = "Rum River State Forest, not near Winona, Minnesota",
				},
			},
		}

		result.assetExpectedValue = KSTableUtils.deepCopy( result.expectedDocStoreValues[ assetExpectedValueIndex ] )
		result.assetExpectedValue._labels = nil
		result.assetExpectedValue._revId = newRevisionId
		result.assetExpectedValue.xmp.dc.description = "Rum River State Forest, not near Winona, Minnesota"

		result.revisionExpectedValue = KSTableUtils.deepCopy( result.assetExpectedValue )
		result.revisionExpectedValue._changeCounter = result.expectedDocStoreValues.continueFrom
		result.revisionExpectedValue._localId = 1

	end

	if options.includeAssetDelete then

		local newRevisionId = KSIdGenerator.uuidInLowercaseHex()

		result.assetDeleteMock = {
			type = "asset",
			action = "delete",
			revision_id = newRevisionId,
			id = assetDocId,
		}

		if type == 'space' then
			result.assetDeleteMock.space_id = catalogOrSpaceId
		end

		result.assetDeleteOverlay = {
			_id = assetDocId,
			_newRevId = newRevisionId,
			_deleted = true,
		}

		result.deletedAssetExpectedValue = KSTableUtils.deepCopy( result.expectedDocStoreValues[ assetExpectedValueIndex ] )
		result.deletedAssetExpectedValue._deleted = true
		result.deletedAssetExpectedValue._labels = nil
		result.deletedAssetExpectedValue._revId = newRevisionId

		result.deletedRevisionExpectedValue = KSTableUtils.deepCopy( result.deletedAssetExpectedValue )
		result.deletedRevisionExpectedValue._changeCounter = result.expectedDocStoreValues.continueFrom
		result.deletedRevisionExpectedValue._localId = 1

	end

	return result

end

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

local WFCatMetadataPusherTest = {}

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

local options = {

	continuousTaskDelays = {
		progressiveBackoff = false,
		noWorkDelay = 0.5,
		maxNoWorkDelay = 1,
		errorDelay = 1,
		maxDelay = 1,
	},

	isCatalogPuller = true,

}


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

local function makeAssetApplyingOverlay( assetId, revId, overlay, subtype )

	local doc = {
			_id = assetId,
			_revId = revId,
			subtype = subtype or 'image',
			type = 'asset',
	}

	doc = WFTableMerge.applyOverlay( doc, overlay )

	return doc

end

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

local function makeAssetWithCreateFrom( assetId, revId, subtype, createFrom )

	local doc = makeAssetApplyingOverlay( assetId, revId, { subtype = subtype } )

	doc.userCreated = kUserCreated

	if subtype == 'deleted_image' or subtype == 'deleted_video' then
		doc.purgeDate = kPurgeDate
	end

	KSTableUtils.setNested( doc, createFrom, '_localOnly', 'universal', 'trash', 'createFrom' )

	return doc

end

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

local function makeAsset( assetId, revId, subtype, createFromId, createFromRevId, createFromSubtype )

	local doc = makeAssetApplyingOverlay( assetId, revId, { subtype = subtype } )

	doc.userCreated = kUserCreated

	if subtype == 'deleted_image' or subtype == 'deleted_video' then
		doc.purgeDate = kPurgeDate
	end

	local createFrom
	if createFromId then

		createFrom = {
			id = createFromId,
			revision_id = createFromRevId,
			subtype = createFromSubtype
		}

		KSTableUtils.setNested( doc, createFrom, '_localOnly', 'universal', 'trash', 'createFrom' )

	end

	return doc

end

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

local function makeDeletedTrashAsset( assetId, revId, subtype, createFromId, createFromRevId, createFromSubtype )

	local doc = makeAsset( assetId, revId, subtype, createFromId, createFromRevId, createFromSubtype )

	doc._deleted = true

	return doc

end

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

local function makePermanentlyDeletedTrashAsset( assetId, revId, subtype, createFromId, createFromRevId, createFromSubtype )

	local doc = makeDeletedTrashAsset( assetId, revId, subtype, createFromId, createFromRevId, createFromSubtype )
	KSTableUtils.setNested( doc, true, '_localOnly', 'universal', 'trash', 'isDeleteTrash' )

	return doc

end

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

local function pack( ... )
	return { ... }
end

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

function WFCatMetadataPusherTest.rxTestConvertOneDocToDelta_deleted_general( rxTest )

	local catalog = WFCatMockCatalog.createFakeWFCatalog( rxTest )
	local pusher = WFCatMetadataPusher( catalog, options )
	rxTest:addCloseTarget( pusher )

	local docId = KSIdGenerator.uuidInLowercaseHex()
	local revisionId = KSIdGenerator.uuidInLowercaseHex()

	local docTypesToTest = {
		asset = true,
		album = true,
		album_asset = true,
		comment = true,
		favorite = true,
		space = true,
		space_album = true,
		space_user = true,
		spaceinvite = true,
		stack_asset = true,
		version = true,
		invalidDocType = true,
	}

	local function makeDocApplyingOverlay( docType, docId, overlay )

		local doc = {
			type = docType,
			_id = docId,
			_deleted = true,
			_revId = revisionId,
		}

		doc = WFTableMerge.applyOverlay( doc, overlay )

		return doc

	end

	for docType, _ in pairs( docTypesToTest ) do

		local ozDocNil = nil

		local ozDoc = {
			_id = docId,
			type = docType,
		}

		local ozDocDeleted = {
			_id = docId,
			_deleted = true,
			type = docType,
		}
		-- deleted doc with conflict
		rxTest:expectEqual( {}, -- Don't send anything as doc has a conflict to be resolved first
			pack( pusher:DEBUG_convertOneDocToDelta( makeDocApplyingOverlay( docType, docId, { _deleted = true, _conflict = true } ), ozDocNil ) ) )

		-- deleted doc, no ozDoc
		rxTest:expectEqual( {}, -- Don't send anything as nothing on oz
			pack( pusher:DEBUG_convertOneDocToDelta( makeDocApplyingOverlay( docType, docId, { _deleted = true } ), ozDocNil ) ) )

		if docType == 'invalidDocType' then
			-- deleted doc, ozDoc
			rxTest:expectEqual( {},
				pack( pusher:DEBUG_convertOneDocToDelta( makeDocApplyingOverlay( docType, docId, { _deleted = true } ), ozDoc ) ) )
		else
			-- deleted doc, ozDoc
			rxTest:expectEqual( {{
				action = "delete",
				id = docId,
				type = docType,
				revision_id = revisionId,
			}}, pack( pusher:DEBUG_convertOneDocToDelta( makeDocApplyingOverlay( docType, docId, { _deleted = true } ), ozDoc ) ) )
		end

		-- deleted doc, deleted ozDoc
		rxTest:expectEqual( {}, -- don't send anything as oz doc already deleted
			pack( pusher:DEBUG_convertOneDocToDelta( makeDocApplyingOverlay( docType, docId, { _deleted = true } ), ozDocDeleted ) ) )

	end

end

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

function WFCatMetadataPusherTest.rxTestConvertOneDocToDelta_deleted_asset( rxTest )

	local catalog = WFCatMockCatalog.createFakeWFCatalog( rxTest )
	local pusher = WFCatMetadataPusher( catalog, options )
	rxTest:addCloseTarget( pusher )

	local docId = KSIdGenerator.uuidInLowercaseHex()
	local revId = KSIdGenerator.uuidInLowercaseHex()

	local createFromAssetId = KSIdGenerator.uuidInLowercaseHex()
	local createFromRevId = KSIdGenerator.uuidInLowercaseHex()
	local ozDocNil = nil

	local ozDoc = {
		_id = docId,
		type = 'asset',
	}

	local ozDocDeleted = {
		_id = docId,
		_deleted = true,
		type = 'asset',
	}

	-- deleted doc with conflict
	rxTest:expectEqual( {}, -- Don't send anything as doc has a conflict to be resolved first
		pack( pusher:DEBUG_convertOneDocToDelta( makeAssetApplyingOverlay( docId, revId, { _conflict = true, _deleted = true } ), ozDocNil ) ) )

	-- deleted doc, no ozDoc
	rxTest:expectEqual( {}, -- Don't send anything as nothing on oz
		pack( pusher:DEBUG_convertOneDocToDelta( makeAssetApplyingOverlay( docId, revId, { _deleted = true } ), ozDocNil ) ) )

	-- deleted doc, ozDoc
	rxTest:expectEqual( {{
		action = "delete",
		id = docId,
		type = "asset",
		revision_id = revId,
	}}, pack( pusher:DEBUG_convertOneDocToDelta( makeAssetApplyingOverlay( docId, revId, { _deleted = true } ), ozDoc ) ) )

	-- deleted doc, deleted ozDoc
	rxTest:expectEqual( {}, -- don't send anything as oz doc already deleted
		pack( pusher:DEBUG_convertOneDocToDelta( makeAssetApplyingOverlay( docId, revId, { _deleted = true } ), ozDocDeleted ) ) )

	-- Cloud Trash doesn't process deletes for assets unless they are marked with isDeleteTrash.
	-- So the commented out cases below should not matter:
--	-- deleted trash doc that existed on oz that was restored
--	rxTest:expectEqual( nil,
--		pusher:DEBUG_convertOneDocToDelta( makeDeletedTrashAsset( docId, revId, 'deleted_image' ), ozDoc ) )
--
--	-- deleted trash doc from deleted asset that didn't exist on oz
--	rxTest:expectEqual( nil,
--		pusher:DEBUG_convertOneDocToDelta( makeDeletedTrashAsset( docId, revId, 'deleted_image' ), ozDocNil ) )
--
--	-- deleted trash doc created from asset that existed on on oz that was restored
--	rxTest:expectEqual( nil,
--		pusher:DEBUG_convertOneDocToDelta( makeDeletedTrashAsset( docId, revId, 'deleted_image', createFromAssetId, createFromRevId, 'image' ), ozDocNil ) )
--
--	-- deleted trash doc created from asset that was created from trash that existed on on oz
--	rxTest:expectEqual( nil,
--		pusher:DEBUG_convertOneDocToDelta( makeDeletedTrashAsset( docId, revId, 'deleted_image', createFromAssetId, createFromRevId, 'deleted_image' ), ozDocNil ) )

	-- permanently deleted trash doc that existed on oz
	rxTest:expectEqual( {{
		action = "delete",
		id = docId,
		type = "asset",
		revision_id = revId,
	}}, pack( pusher:DEBUG_convertOneDocToDelta( makePermanentlyDeletedTrashAsset( docId, revId, 'deleted_image' ), ozDoc ) ) )

	-- permanently deleted trash doc from deleted asset that was created from asset that never existed on oz
	rxTest:expectEqual( {},
		pack( pusher:DEBUG_convertOneDocToDelta( makePermanentlyDeletedTrashAsset( docId, revId, 'deleted_image' ), ozDocNil ) ) )

	-- permanently deleted trash doc created from asset that existed on on oz
	rxTest:expectEqual( {
		{
			action = "delete",
			id = docId,
			type = "asset",
			revision_id = revId,
		},
		{
			action = "delete",
			deleted_asset = {
				id = docId,
				revision_id = revId,
				payload = {
					userCreated = kUserCreated,
					purgeDate = kPurgeDate,
				},
			},
			id = createFromAssetId,
			revision_id = createFromRevId,
			type = "asset",
		},
	}, pack( pusher:DEBUG_convertOneDocToDelta( makePermanentlyDeletedTrashAsset( docId, revId, 'deleted_image', createFromAssetId, createFromRevId, 'image' ), ozDocNil ) ) )

	-- permanently deleted trash doc created from asset that was created from trash asset that existed on on oz
	rxTest:expectEqual( {{ -- just send the delete to the prior trash asset
		action = "delete",
		id = createFromAssetId,
		type = "asset",
		revision_id = createFromRevId,
	}},
	pack( pusher:DEBUG_convertOneDocToDelta( makePermanentlyDeletedTrashAsset( docId, revId, 'deleted_image', createFromAssetId, createFromRevId, 'deleted_image' ), ozDocNil ) ) )

end

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

function WFCatMetadataPusherTest.rxTestConvertOneDocToDelta_changed_general( rxTest )

	local catalog = WFCatMockCatalog.createFakeWFCatalog( rxTest )
	local pusher = WFCatMetadataPusher( catalog, options )
	rxTest:addCloseTarget( pusher )

	local docId = KSIdGenerator.uuidInLowercaseHex()
	local revId = KSIdGenerator.uuidInLowercaseHex()
	local ozRevId = KSIdGenerator.uuidInLowercaseHex()

	local kCreated = '2019-09-05T09:07:01.002Z'
	local kUpdated = '2019-09-05T09:07:02.002Z'

	local docTypesToTest = {
		asset = true,
		album = true,
		album_asset = true,
		catalog = 'nocreate',
		comment = 'noupdate',
		favorite = 'noupdate',
		space = true,
		space_album = 'noupdate',
		space_user = 'nocreate',
		spaceinvite = true,
		stack_asset = true,
		version = true,
		invalidDocType = true,
	}

	local function makeDocApplyingOverlay( docType, docId, revId, overlay )

		local doc = {
			type = docType,
			_id = docId,
			_revId = revId,
		}

		doc = WFTableMerge.applyOverlay( doc, overlay )

		return doc

	end

	for docType, _ in pairs( docTypesToTest ) do

		local ozDocNil = nil -- ozDoc must be nil in order for convertOneDocToDelta to call the deltaForCreateDoc

		local ozDoc = {
			_id = docId,
			type = docType,
			_serverProvidedMetadata = {
				base = 'base_url',
				created = kCreated,
				links = { ['self'] = 'self_url_info' },
				revision_ids = { ozRevId },
				updated = kUpdated,
			},
		}

		if docType == 'invalidDocType' then

			rxTest:expectEqual( {},
				pack( pusher:DEBUG_convertOneDocToDelta( makeDocApplyingOverlay( docType, docId, revId, {} ), ozDocNil ) ) )

		else

			local StorageExceededError = {
				_localOnly = {
					ozError = {
						subtype = 'StorageExceededError'
					}
				}
			}
			rxTest:expectEqual( {}, pack( pusher:DEBUG_convertOneDocToDelta( makeDocApplyingOverlay( docType, docId, revId, StorageExceededError ), ozDocNil ) ) )

			local subExpiredError = {
				_localOnly = {
					ozError = {
						subtype = 'TrialOrSubscriptionExpiredError'
					}
				}
			}
			rxTest:expectEqual( {}, pack( pusher:DEBUG_convertOneDocToDelta( makeDocApplyingOverlay( docType, docId, revId, subExpiredError ), ozDocNil ) ) )

			local resourceNotFoundError = {
				_localOnly = {
					ozError = {
						subtype = 'ResourceNotFoundError'
					}
				}
			}
			rxTest:expectEqual( {}, pack( pusher:DEBUG_convertOneDocToDelta( makeDocApplyingOverlay( docType, docId, revId, resourceNotFoundError ), ozDocNil ) ) )

			if docTypesToTest[ docType ] == 'nocreate' then

				rxTest:expectEqual( {},
					pack( pusher:DEBUG_convertOneDocToDelta( makeDocApplyingOverlay( docType, docId, revId, {} ), ozDocNil ) ) )

			else

				local createDelta = {
					action = 'create',
					id = docId,
					payload = { },
					revision_id = revId,
					type = docType,
				}

				rxTest:expectEqual( { createDelta }, pack( pusher:DEBUG_convertOneDocToDelta( makeDocApplyingOverlay( docType, docId, revId, {} ), ozDocNil ) ) )

				-- test with Atomic Develop Settings enabled

				pusher:TEST_getCapabilitiesCache():setCapabilitiesForTests( {
					[ 'WFCatBinaryUploader.atomicDevelopSettings' ] = true,
				})

				-- only versions should send prefer=update-atomic on create

				if docType == 'version' then
					createDelta.prefer = 'update-atomic'
				end

				rxTest:expectEqual( { createDelta }, pack( pusher:DEBUG_convertOneDocToDelta( makeDocApplyingOverlay( docType, docId, revId, {} ), ozDocNil ) ) )

				pusher:TEST_getCapabilitiesCache():setCapabilitiesForTests( {
					[ 'WFCatBinaryUploader.atomicDevelopSettings' ] = false,
				} )

			end

			if docTypesToTest[ docType ] == 'noupdate' then

				rxTest:expectEqual( {},
					pack( pusher:DEBUG_convertOneDocToDelta( makeDocApplyingOverlay( docType, docId, revId, { userUpdated = kUserCreated } ), ozDoc ) ) )

			else

				local updateDelta = {
					action = 'update',
					id = docId,
					if_match = ozRevId,
					payload = {
						userUpdated = kUserCreated,
					},
					revision_id = revId,
					type = docType,
				}


				rxTest:expectEqual( { updateDelta }, pack( pusher:DEBUG_convertOneDocToDelta( makeDocApplyingOverlay( docType, docId, revId, { userUpdated = kUserCreated } ), ozDoc ) ) )

				-- test with Atomic Develop Settings enabled

				pusher:TEST_getCapabilitiesCache():setCapabilitiesForTests( {
					[ 'WFCatBinaryUploader.atomicDevelopSettings' ] = true,
				} )

				-- image and deleted_iamge assets should send prefer = update-atomic, but document in this
				-- test has no subtype, so even asset documents will not include it.

				rxTest:expectEqual( { updateDelta }, pack( pusher:DEBUG_convertOneDocToDelta( makeDocApplyingOverlay( docType, docId, revId, { userUpdated = kUserCreated } ), ozDoc ) ) )

				pusher:TEST_getCapabilitiesCache():setCapabilitiesForTests( {
					[ 'WFCatBinaryUploader.atomicDevelopSettings' ] = false,
				} )

			end

		end

	end

	-- additional test for album_asset update with album/asset id changes which shouldn't be allowed and shouldn't get to Oz
	-- this is related to trash actions that can update optimistically update the id (assuming it's not yet on Oz)
	rxTest:expectEqual( {}, pack( pusher:DEBUG_convertOneDocToDelta(
		{ 	_id = docId,
			_revId = revId,
			type = 'album_asset',
			album = { id = KSIdGenerator.uuidInLowercaseHex() },
			asset = { id = KSIdGenerator.uuidInLowercaseHex() }
		},
		{ 	_id = docId,
			type = 'album_asset',
			_serverProvidedMetadata = {
				base = 'base_url',
				created = kCreated,
				links = { ['self'] = 'self_url_info' },
				revision_ids = { revId },
				updated = kUpdated,
			}
		} ) ) )
end

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

function WFCatMetadataPusherTest.rxTestConvertOneDocToDelta_changed_asset( rxTest )

	for _, assetType in ipairs( { "image", "video" } ) do

		local catalog = WFCatMockCatalog.createFakeWFCatalog( rxTest )
		local pusher = WFCatMetadataPusher( catalog, options )
		rxTest:addCloseTarget( pusher )

		local docId = KSIdGenerator.uuidInLowercaseHex()
		local revId = KSIdGenerator.uuidInLowercaseHex()
		local ozRevId = KSIdGenerator.uuidInLowercaseHex()

		local createFromAssetId = KSIdGenerator.uuidInLowercaseHex()
		local createFromRevId = KSIdGenerator.uuidInLowercaseHex()
		local ozDocNil = nil

		local kCreated = '2019-09-05T09:07:01.002Z'
		local kUpdated = '2019-09-05T09:07:02.002Z'

		local intermediateTrashId = KSIdGenerator.uuidInLowercaseHex()
		local intermediateTrashRevId = KSIdGenerator.uuidInLowercaseHex()

		local intermediateRestoredId = KSIdGenerator.uuidInLowercaseHex()
		local intermediateRestoredRevId = KSIdGenerator.uuidInLowercaseHex()

		local ozDoc = {
			_id = docId,
			type = 'asset',
			subtype = 'image',
			_serverProvidedMetadata = {
				base = 'base_url',
				created = kCreated,
				links = { ['self'] = 'self_url_info' },
				revision_ids = { ozRevId },
				updated = kUpdated,
			},
		}

		-- Documents Not Sent Scenarios --

		-- changed doc with conflict
		rxTest:expectEqual( {}, -- Don't send anything as doc has a conflict to be resolved first
			pack( pusher:DEBUG_convertOneDocToDelta( makeAssetApplyingOverlay( docId, revId, { _conflict = true } ), ozDoc ) ) )

		-- changed doc with ozError preventing upload (see shouldSkipUploadDueToError)
		for _, ozErrorSubtype in pairs( { 'StorageExceededError', 'TrialOrSubscriptionExpiredError', 'ResourceNotFoundError' } ) do

			local localOnly = { ozError = { subtype = ozErrorSubtype } }

			rxTest:expectEqual( {}, -- Don't send anything as doc has an error to possibly be resolved first
				pack( pusher:DEBUG_convertOneDocToDelta( makeAssetApplyingOverlay( docId, revId, { _localOnly = localOnly } ), ozDoc ) ) )

		end

		-- changed doc with localDoc._revId == ozDoc._revId
		local ozDocRevIdMatchesLocal = KSTableUtils.deepCopy( ozDoc )
		ozDocRevIdMatchesLocal._revId = revId

		-- changed doc, ozDoc
		rxTest:expectEqual( {}, pack( pusher:DEBUG_convertOneDocToDelta( makeAsset( docId, revId, assetType ), ozDocRevIdMatchesLocal ) ) )


		-- Update Scenarios -- cases where deltaForUpdateDoc is called

		-- changed doc, ozDoc
		rxTest:expectEqual( {{
			action = 'update',
			id = docId,
			if_match = ozRevId,
			payload = {
				userCreated = kUserCreated,
			},
			revision_id = revId,
			type = 'asset',
		}}, pack( pusher:DEBUG_convertOneDocToDelta( makeAsset( docId, revId, assetType ), ozDoc ) ) )

		-- changed doc, ozDoc, atomicDevelopSettings capability enabled

		pusher:TEST_getCapabilitiesCache():setCapabilitiesForTests( {
			[ 'WFCatBinaryUploader.atomicDevelopSettings' ] = true,
		} )

		-- will send prefer = update-atomic for image and deleted_image docs

		local expectedDoc = {
			action = 'update',
			id = docId,
			if_match = ozRevId,
			payload = {
				userCreated = kUserCreated,
			},
			revision_id = revId,
			type = 'asset',
			prefer = 'update-atomic',
		}

		rxTest:expectEqual( { expectedDoc }, pack( pusher:DEBUG_convertOneDocToDelta( makeAsset( docId, revId, assetType ), ozDoc ) ) )

		ozDoc.subtype = 'deleted_image'
		rxTest:expectEqual( { expectedDoc }, pack( pusher:DEBUG_convertOneDocToDelta( makeAsset( docId, revId, assetType ), ozDoc ) ) )

		-- will not send prefer = update-atomic for video and other asset subtypes

		expectedDoc.prefer = nil

		ozDoc.subtype = 'video'
		rxTest:expectEqual( { expectedDoc }, pack( pusher:DEBUG_convertOneDocToDelta( makeAsset( docId, revId, assetType ), ozDoc ) ) )

		ozDoc.subtype = 'deleted_video'
		rxTest:expectEqual( { expectedDoc }, pack( pusher:DEBUG_convertOneDocToDelta( makeAsset( docId, revId, assetType ), ozDoc ) ) )

		ozDoc.subtype = 'profile'
		rxTest:expectEqual( { expectedDoc }, pack( pusher:DEBUG_convertOneDocToDelta( makeAsset( docId, revId, assetType ), ozDoc ) ) )

		pusher:TEST_getCapabilitiesCache():setCapabilitiesForTests( {
			[ 'WFCatBinaryUploader.atomicDevelopSettings' ] = false,
		} )

		-- changed doc, ozDoc -- no _serverProvidedMetadata
		local ozDocNoSPM = KSTableUtils.deepCopy( ozDoc )
		ozDocNoSPM._serverProvidedMetadata = nil
		rxTest:expectEqual( {}, pack( pusher:DEBUG_convertOneDocToDelta( makeAsset( docId, revId, assetType ), ozDocNoSPM ) ) )

		-- changed doc, ozDoc -- no data to send in payload
		local ozDocNoPayload = KSTableUtils.deepCopy( ozDoc )
		ozDocNoPayload.userCreated = kUserCreated -- adding userCreated causes ozDoc to match localDoc
		rxTest:expectEqual( {}, pack( pusher:DEBUG_convertOneDocToDelta( makeAsset( docId, revId, assetType ), ozDocNoPayload ) ) )

		-- changed doc, ozDoc, has internal develop settings
		rxTest:expectEqual( {{
			action = 'update',
			id = docId,
			if_match = ozRevId,
			generate_renditions = assetType == 'image' and 'all' or nil,
			payload = {
				develop = { xmpCameraRaw = 'gobbledygook' },
			},
			revision_id = revId,
			type = 'asset',
		}}, pack( pusher:DEBUG_convertOneDocToDelta( makeAssetApplyingOverlay( docId, revId, { develop = { xmpCameraRaw = 'gobbledygook' } }, assetType), ozDoc ) ) )

		-- changed doc, ozDoc, has same develop.xmpCameraRaw.sha256, but with develop.masks section update
		local ozDocXmpCameraRawSha256 = KSTableUtils.deepCopy( ozDoc )
		KSTableUtils.setNested( ozDocXmpCameraRawSha256, 'sha256', 'develop', 'xmpCameraRaw', 'sha256' )
		local developUpdate = {
			masks = {
				auxId = { digest = 'digest' }
			},
			xmpCameraRaw = { sha256 = 'sha256' }
		}
		rxTest:expectEqual( {{
			action = 'update',
			id = docId,
			if_match = ozRevId,
			payload = {
				develop = {
					masks = {
						auxId = { digest = 'digest' }
					}
				}
			},
			revision_id = revId,
			type = 'asset',
		}}, pack( pusher:DEBUG_convertOneDocToDelta( makeAssetApplyingOverlay( docId, revId, { develop = developUpdate }, assetType), ozDocXmpCameraRawSha256 ) ) )

		-- changed doc, ozDoc, has same develop.xmpCameraRaw.sha256, but with develop.filters section update
		local developFiltersUpdate = {
			filters = {
				groupDigest = {
					auxId = { digest = 'digest' }
				}
			},
			xmpCameraRaw = { sha256 = 'sha256' }
		}
		rxTest:expectEqual( {{
			action = 'update',
			id = docId,
			if_match = ozRevId,
			payload = {
				develop = {
					filters = {
						groupDigest = {
							auxId = { digest = 'digest' }
						}
					}
				},
			},
			revision_id = revId,
			type = 'asset',
		}}, pack( pusher:DEBUG_convertOneDocToDelta( makeAssetApplyingOverlay( docId, revId, { develop = developFiltersUpdate }, assetType), ozDocXmpCameraRawSha256 ) ) )

		-- changed doc, ozDoc, has video internal develop settings
		if assetType == 'video' then

			local developAdjustments = {
				adjustments = {
					userUpdated = kUpdated
				}
			}

			-- changed doc, ozDoc, has develop.adjustments
			rxTest:expectEqual( {{
				action = 'update',
				id = docId,
				if_match = ozRevId,
				generate_renditions = 'all',
				payload = {
					develop = developAdjustments
				},
				revision_id = revId,
				type = 'asset',
			}}, pack( pusher:DEBUG_convertOneDocToDelta( makeAssetApplyingOverlay( docId, revId, { develop = developAdjustments }, assetType), ozDoc ) ) )

			local developTemporal = {
				temporal = {
					userUpdated = kUpdated
				}
			}

			-- changed doc, ozDoc, has internal develop.temperal
			rxTest:expectEqual( {{
				action = 'update',
				id = docId,
				if_match = ozRevId,
				generate_renditions = 'all',
				payload = {
					develop = developTemporal
				},
				revision_id = revId,
				type = 'asset',
			}}, pack( pusher:DEBUG_convertOneDocToDelta( makeAssetApplyingOverlay( docId, revId, { develop = developTemporal }, assetType), ozDoc ) ) )

		end

		-- changed doc, ozDoc, has external develop settings
		local extDevelopSettings = {
			xmpCameraRaw = {
				sha256 = '456'
			}
		}

		rxTest:expectEqual( {{
			action = 'update',
			id = docId,
			if_match = ozRevId,
			-- generate_renditions = 'all', -- make sure this isn't set
			payload = {
				develop = extDevelopSettings,
			},
			revision_id = revId,
			type = 'asset',
		}}, pack( pusher:DEBUG_convertOneDocToDelta( makeAssetApplyingOverlay( docId, revId, { develop = extDevelopSettings }, assetType ), ozDoc) ) )

		-- with external develop settings with AtomicDevelopSettings enabled should include generate_renditions=all

		pusher:TEST_getCapabilitiesCache():setCapabilitiesForTests( {
			[ 'WFCatBinaryUploader.atomicDevelopSettings' ] = true,
		})

		rxTest:expectEqual( {{
			action = 'update',
			id = docId,
			if_match = ozRevId,
			generate_renditions = 'all',
			payload = {
				develop = extDevelopSettings,
			},
			revision_id = revId,
			type = 'asset',
		}}, pack( pusher:DEBUG_convertOneDocToDelta( makeAssetApplyingOverlay( docId, revId, { develop = extDevelopSettings }, assetType ), ozDoc) ) )

		pusher:TEST_getCapabilitiesCache():setCapabilitiesForTests( {
			[ 'WFCatBinaryUploader.atomicDevelopSettings' ] = false,
		} )

		-- make sure purgeDate isn't sent
		local newTime =  ">>newTime"
		local purgeDate = ">>newPurgeDate"
		rxTest:expectEqual( {{
			action = 'update',
			id = docId,
			if_match = ozRevId,
			-- generate_renditions = 'all', -- make sure this isn't set
			payload = {
				userUpdated = newTime,
			},
			revision_id = revId,
			type = 'asset',
		}}, pack( pusher:DEBUG_convertOneDocToDelta( makeAssetApplyingOverlay( docId, revId, { userUpdated = newTime, purgeDate = purgeDate } ), ozDoc ) ) )

		-- Create Scenarios -- cases where deltaForCreateDoc is called

		-- Regular Asset --

		local regularCreate = {
			action = 'create',
			id = docId,
			payload = {
				userCreated = kUserCreated,
			},
			subtype = assetType,
			type = 'asset',
			revision_id = revId,
		}
		-- changed doc no createFrom, no ozDoc
		rxTest:expectEqual( { regularCreate }, pack( pusher:DEBUG_convertOneDocToDelta( makeAsset( docId, revId, assetType ), ozDocNil ) ) )

		-- changed doc no createFrom, no ozDoc, but overquota
		pusher:setUserIsOverQuota( true )
		rxTest:expectEqual( {}, pack( pusher:DEBUG_convertOneDocToDelta( makeAsset( docId, revId, assetType ), ozDocNil ) ) )
		pusher:setUserIsOverQuota( false )
		rxTest:expectEqual( { regularCreate }, pack( pusher:DEBUG_convertOneDocToDelta( makeAsset( docId, revId, assetType ), ozDocNil ) ) )

		-- changed doc no createFrom, no ozDoc, but account expired
		pusher:setUserIsExpired( true )
		rxTest:expectEqual( {}, pack( pusher:DEBUG_convertOneDocToDelta( makeAsset( docId, revId, assetType ), ozDocNil ) ) )
		pusher:setUserIsExpired( false )
		rxTest:expectEqual( { regularCreate }, pack( pusher:DEBUG_convertOneDocToDelta( makeAsset( docId, revId, assetType ), ozDocNil ) ) )

		-- changed doc no localDoc.type, no ozDoc
		rxTest:expectEqual( {}, -- Don't send anything as doc has a conflict to be resolved first
			pack( pusher:DEBUG_convertOneDocToDelta( makeAssetApplyingOverlay( docId, revId, { type = { _delete = true } } ), ozDocNil ) ) )

		-- changed doc no createFrom, no ozDoc with preset, profile, or camera_profile
		local profileSha256 = '123'
		local profileImpSrc = { sha256 = profileSha256 }

		for _, testSubtype in pairs( { 'preset', 'profile', 'camera_profile' } ) do

			rxTest:expectEqual( {{
				action = 'create',
				id = docId,
				payload = {
					importSource = profileImpSrc,
				},
				if_none_match = profileSha256,
				subtype = testSubtype,
				type = 'asset',
				revision_id = revId,
			}}, pack( pusher:DEBUG_convertOneDocToDelta( makeAssetApplyingOverlay( docId, revId, { subtype = testSubtype, importSource = profileImpSrc } ), ozDocNil ) ) )

		end

		-- Restored Asset --

		-- changed doc with createFrom with different subtype, no ozDoc - (Restore from trash asset)
		rxTest:expectEqual( {{
			action = 'create',
			id = docId,
			payload = {
			},
			restore_from = {
				id = createFromAssetId,
			},
			subtype = assetType,
			type = 'asset',
		}}, pack( pusher:DEBUG_convertOneDocToDelta( makeAsset( docId, revId, assetType, createFromAssetId, createFromRevId, 'deleted_' .. assetType ), ozDocNil ) ) )

		-- changed doc with createFrom with same subtype, no ozDoc (asset that was deleted and restored while offline)
		local restored_createFromInfo = {
			id = createFromAssetId,
			revision_id = createFromRevId,
			subtype = assetType,
			lastAssetId = intermediateTrashId,
			lastRevisionId = intermediateTrashRevId,
			lastTrashPurgeDate = kPurgeDate,
			lastTrashUserCreated = kUserCreated,
		}

		rxTest:expectEqual( {{
			action = 'create',
			id = docId,
			payload = {
			},
			restore_from = {
				id = intermediateTrashId,
			},
			subtype = assetType,
			type = 'asset',
		},
		{
			action = 'delete',
			id = createFromAssetId,
			revision_id = createFromRevId,
			type = 'asset',
			deleted_asset = {
				id = intermediateTrashId,
				revision_id = intermediateTrashRevId,
				payload = {
					purgeDate = kPurgeDate,
					userCreated = kUserCreated,
				}
			}
		}}, pack( pusher:DEBUG_convertOneDocToDelta( makeAssetWithCreateFrom( docId, revId, assetType, restored_createFromInfo ), ozDocNil ) ) )


		-- Trash Asset --

		-- changed trash doc no createFrom, no ozDoc (offline imported and deleted trash asset)
		rxTest:expectEqual( {{
			action = 'create',
			id = docId,
			type = 'asset',
			subtype = 'deleted_' .. assetType,
			payload = {
				purgeDate = kPurgeDate,
				userCreated = kUserCreated,
			},
			revision_id = revId,
		}}, pack( pusher:DEBUG_convertOneDocToDelta( makeAsset( docId, revId, 'deleted_' .. assetType ), ozDocNil ) ) )

		-- changed trash doc with createFrom with different subtype, no ozDoc (offline restored and deleted trash asset)
		rxTest:expectEqual( {{
			action = 'delete',
			deleted_asset = {
				id = docId,
				payload = {
					purgeDate = kPurgeDate,
					userCreated = kUserCreated,
				},
			},
			id = createFromAssetId,
			revision_id = createFromRevId,
			type = 'asset',
		}}, pack( pusher:DEBUG_convertOneDocToDelta( makeAsset( docId, revId, 'deleted_' .. assetType, createFromAssetId, createFromRevId, assetType ), ozDocNil ) ) )

		-- changed trash doc with createFrom with same subtype, no ozDoc (asset that was restored and deleted while offline)
		local trashed_createFromInfo = {
			id = createFromAssetId,
			revision_id = createFromRevId,
			subtype = 'deleted_' .. assetType,
			lastAssetId = intermediateRestoredId,
			lastRevisionId = intermediateRestoredRevId,
			lastTrashPurgeDate = kPurgeDate,
			lastTrashUserCreated = kUserCreated,
		}

		rxTest:expectEqual( {{
			action = 'delete',
			deleted_asset = {
				id = docId,
				payload = {
					purgeDate = kPurgeDate,
					userCreated = kUserCreated,
				}
			},
			id = intermediateRestoredId,
			revision_id = intermediateRestoredRevId,
			type = 'asset',
		},
		{
			action = 'create',
			id = intermediateRestoredId,
			payload = {
				userUpdated = kUserCreated,
			},
			restore_from = {
				id = createFromAssetId,
			},
			revision_id = intermediateRestoredRevId,
			subtype = assetType,
			type = 'asset',
		}}, pack( pusher:DEBUG_convertOneDocToDelta( makeAssetWithCreateFrom( docId, revId, 'deleted_' .. assetType, trashed_createFromInfo ), ozDocNil ) ) )


		-- Invalid Docs Scenarios--

		-- NMB-13862
		-- changed doc with createFrom with same subtype, no ozDoc (asset that was deleted and restored while offline)
		-- but in this invalid case that should never happen set the createFrom info to the docId itself.
		-- Oz has seen this happen but I haven't determined how it can yet so we are guarding against it
		-- in WFCatMetadataPusher's assetCreateRemapFn function.
		local invalid_restored_createFromInfo = {
			id = docId,
			revision_id = revId,
			subtype = assetType,
			lastAssetId = docId,
			lastRevisionId = revId,
			lastTrashPurgeDate = kPurgeDate,
			lastTrashUserCreated = kUserCreated,
		}

		rxTest:expectEqual( {{
			action = 'create',
			id = docId,
			payload = {
				userCreated = kUserCreated,
			},
			revision_id = revId,
			subtype = assetType,
			type = 'asset',
		}}, pack( pusher:DEBUG_convertOneDocToDelta( makeAssetWithCreateFrom( docId, revId, assetType, invalid_restored_createFromInfo ), ozDocNil ) ) )

		-- changed doc, deleted ozDoc (asset updated but was previously deleted, not sure if this can happen)
		-- Oz would likely respond with a resource not found error but verify what we do in this case.

		local ozDocDeleted = KSTableUtils.deepCopy( ozDoc )
		ozDocDeleted._deleted = true
		ozDocDeleted._revId = ozRevId

		rxTest:expectEqual( {{
			action = 'update',
			id = docId,
			if_match = ozRevId,
			payload = {
				userCreated = kUserCreated,
			},
			revision_id = revId,
			type = 'asset',
		}}, pack( pusher:DEBUG_convertOneDocToDelta( makeAsset( docId, revId, assetType), ozDocDeleted ) ) )

		-- get deltaForUpdateDoc code coverage to 100%

		-- changed doc, ozDoc with remap function (keep this last in case a call to addUpdateDocRemapFunction is called)
		pusher:addUpdateDocRemapFunction( 'asset', function( self, localDoc, delta, docToSend ) docToSend.testRemapFunctionCalled = true end )
		rxTest:expectEqual( {{
			action = 'update',
			id = docId,
			if_match = ozRevId,
			payload = {
				userCreated = kUserCreated,
			},
			testRemapFunctionCalled = true,
			revision_id = revId,
			type = 'asset',
		}}, pack( pusher:DEBUG_convertOneDocToDelta( makeAsset( docId, revId, assetType ), ozDoc ) ) )

	end

end

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

function WFCatMetadataPusherTest.rxTestRequestHeaders( rxTest )

	local catalog = WFCatMockCatalog.createFakeWFCatalog( rxTest )

	local pusher = WFCatMetadataPusher( catalog, options )

	rxTest:addCloseTarget( pusher )

	pusher:addRequestHeader( 'X-Quicker-Retries', 'true' )

	local o = Rx.Observable.from( pusher._requestHeaders )
		:where( function( request )

			return request.header == 'X-Quicker-Retries'

		end )

		:map( function( header )

			return header.value

		end )

	rxTest:expectSingleValue( o, 'true' )

	pusher:removeRequestHeader( 'X-Quicker-Retries' )

	o = Rx.Observable.from( pusher._requestHeaders )
		:where( function( request )

			return request.header == 'X-Quicker-Retries'

		end )

	rxTest:expectOnlyCompleted( o )

	rxTest:expectEqual( #pusher._requestHeaders > 0, true )

end

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

function WFCatMetadataPusherTest.rxTestSendUpdates_realOz_testNightly( rxTest )

	rxTest:setTimeLimit( 120 )

	local catalogTestSession = WFCatTestSession( rxTest )

	local catalog = rxTest:getFirstValue( catalogTestSession:catalog() )

	local puller = WFCatMetadataPuller( catalog, "Test Puller", options )
	rxTest:addCloseTarget( puller )

	-- do a manual pull first to ensure the catalog doc gets localId 1
	rxTest:expectOnlyCompleted( puller:getUpdatesOnce() )

	puller:getUpdatesContinuously()

	local pusher = WFCatMetadataPusher( catalog, options )
	rxTest:addCloseTarget( pusher )

	-- Very simple test: We push one asset to Oz and then we delete it.

	local catalogSpec = rxTest:getFirstValue( catalogTestSession:catalogSpec() )

	local assetId = KSIdGenerator.uuidInLowercaseHex()

	rxTest:waitFor(

		catalog:docStore():setDocValue {
			_id = assetId,
			_localOnly = {
				mumble = "stuff",
			},
			captureDate = "2004-11-02T11:10:09",
			importSource = {
				fileName = "fakeFile01.jpg",
				importedBy = catalogSpec:ozUserId(),
				importTimestamp = "2015-11-02T08:37:24.513Z",
				originalHeight = 16,
				originalWidth = 16,
			},
			subtype = "image",
			type = "asset",
		}

	)

	local sendNUpdates = rxTest:observe( pusher:_DEBUG_sendNUpdates( 1 ) )
	sendNUpdates:expectCompletedWithinTime( 30 )
	sendNUpdates:dispose()

	local observeAssetOnOz = observeHasServerProvidedMetadata( rxTest, catalog, assetId )
	observeAssetOnOz:expectDataWithinTime( 40, true )
	observeAssetOnOz:dispose()

	rxTest:expectSingleValue(

		catalog:docStore():changedDocs { fromChangeCounter = 0 }
			:map( remapChangedDocsFeed ),

		{
			{
				_labels = {
					oz = true,
				},
				_localOnly = {
					mumble = "stuff",
				},
				_serverProvidedMetadata = {
					base = ">>base",
					created = ">>created",
					links = ">>links",
					revision_ids = ">>revision_ids",
					updated = ">>updated",
				},
				captureDate = "2004-11-02T11:10:09",
				importSource = {
					fileName = "fakeFile01.jpg",
					importedBy = catalogSpec:ozUserId(),
					importTimestamp = "2015-11-02T08:37:24.513Z",
					originalHeight = 16,
					originalWidth = 16,
				},
				subtype = "image",
				type = "asset",
			},
			{
				_labels = {
					oz = true,
				},
				_serverProvidedMetadata = {
					base = ">>base",
					created = ">>created",
					links = ">>links",
					revision_ids = ">>revision_ids",
					updated = ">>updated",
				},
				name = "WFCatTestSession",
				subtype = "lightroom",
				type = "catalog",
			},
			continueFrom = 4,
		} )

	-- Now delete the doc locally and make sure that gets sent to Oz.

	rxTest:waitFor(
		catalog:docStore():deleteDoc( assetId )
	)

	sendNUpdates = rxTest:observe( pusher:_DEBUG_sendNUpdates( 1 ) )
	sendNUpdates:expectCompletedWithinTime( 30 )
	sendNUpdates:dispose()

	local observeAssetDeletedOnOz = observeIsDeletedOnOz( rxTest, catalog, assetId )
	observeAssetDeletedOnOz:expectDataWithinTime( 40, true )
	observeAssetDeletedOnOz:dispose()

	rxTest:expectSingleValue(

		catalog:docStore():changedDocs { fromChangeCounter = 0 }
			:map( remapChangedDocsFeed ),

		{
			{
				_deleted = true,
				_labels = {
					oz = true,
				},
				_localOnly = {
					mumble = "stuff",
				},
				_serverProvidedMetadata = {
					base = ">>base",
					deleted_by = ">>deleted_by",
					new_id = ">>new_id",
					updated = ">>updated",
				},
				type = 'asset',
			},
			{
				_labels = {
					oz = true,
				},
				_serverProvidedMetadata = {
					base = ">>base",
					created = ">>created",
					links = ">>links",
					revision_ids = ">>revision_ids",
					updated = ">>updated",
				},
				captureDate = "2004-11-02T11:10:09",
				importSource = {
					fileName = "fakeFile01.jpg",
					importTimestamp = "2015-11-02T08:37:24.513Z",
					importedBy = catalogSpec:ozUserId(),
					originalHeight = 16,
					originalWidth = 16,
				},
				original = {
					id = assetId,
					userCreated = "2015-11-02T08:37:24.513000Z",
				},
				purgeDate = ">>purgeDate",
				subtype = "deleted_image",
				type = "asset",
				userCreated = ">>userCreated",
				userUpdated = ">>userUpdated",
			},
			{
				_labels = {
					oz = true,
				},
				_serverProvidedMetadata = {
					base = ">>base",
					created = ">>created",
					links = ">>links",
					revision_ids = ">>revision_ids",
					updated = ">>updated",
				},
				name = "WFCatTestSession",
				subtype = "lightroom",
				type = "catalog",
			},
			continueFrom = 7,
		} )

	-- Read the update feed directly from Oz, too.

	rxTest:expectSingleValue(

		catalogSpec:get( string.format( '%s/updates?updated_since=2008-08-17T09:55:34Z', catalogSpec:catalogHref() ) )
			:map( KSJsonDecoder.decodeWebSafeJson )
			:map( stripOzChangesFeed ),

		{
			base = "https://photos-stage.adobe.io/v2/catalogs/"
						.. catalog:ozCatalogSpec():catalogId() .. "/",
			updates = {
				{
					created = ">>created",
					id = ">>id",
					links = ">>links",
					original = ">>original",
					payload = {
						captureDate = "2004-11-02T11:10:09",
						importSource = {
							fileName = "fakeFile01.jpg",
							importTimestamp = "2015-11-02T08:37:24.513Z",
							importedBy = catalogSpec:ozUserId(),
							originalHeight = 16,
							originalWidth = 16,
						},
						purgeDate = ">>purgeDate",
						userCreated = ">>userCreated",
						userUpdated = ">>userUpdated",
					},
					revision_ids = ">>revision_ids",
					subtype = "deleted_image",
					type = "asset",
					updated = ">>updated",
				},
				{
					deleted = true,
					deleted_by = ">>deleted_by",
					id = ">>id",
					new_id = ">>new_id",
					type = "asset",
					updated = ">>updated",
				}
			},
		} )

end

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

function WFCatMetadataPusherTest.rxTestSendUpdates_withDevelopAux_testNightly( rxTest )

	rxTest:setTimeLimit( 120 )

	local catalogTestSession = WFCatTestSession( rxTest )

	local catalog = rxTest:getFirstValue( catalogTestSession:catalog() )

	local puller = WFCatMetadataPuller( catalog, "Test Puller", options )
	rxTest:addCloseTarget( puller )

	-- do a manual pull first to ensure the catalog doc gets localId 1
	rxTest:expectOnlyCompleted( puller:getUpdatesOnce() )

	puller:getUpdatesContinuously()

	local pusher = WFCatMetadataPusher( catalog, options )
	rxTest:addCloseTarget( pusher )

	-- Very simple test: We push one asset to Oz and then we delete it.

	local catalogSpec = rxTest:getFirstValue( catalogTestSession:catalogSpec() )

	local assetId = KSIdGenerator.uuidInLowercaseHex()
	
	local maskAuxId = KSIdGenerator.uuidInLowercaseHex()
	local filterAuxId = KSIdGenerator.uuidInLowercaseHex()

	local maskDigest = "E01B487AA95FA9FAAC0BC8AFDEFB08C5"
	local filterDigest = "DCA6FDD08DA1D5D305490DE65DAE8F30"

	rxTest:waitFor(

		catalog:docStore():setDocValue {
			_id = assetId,
			_localOnly = {
				mumble = "stuff",
			},
			captureDate = "2004-11-02T11:10:09",
			importSource = {
				fileName = "fakeFile01.jpg",
				importedBy = catalogSpec:ozUserId(),
				importTimestamp = "2015-11-02T08:37:24.513Z",
				originalHeight = 16,
				originalWidth = 16,
			},
			subtype = "image",
			type = "asset",
			aux = {
				[ maskAuxId ] = {
					type = "mask",
					fileSize = 64157,
					digest = maskDigest,
					sha256 = "83b7295d955f5449a6001d7e78b947944c2cba4f97bfcac12cfb28d21664984a"
				},
				[ filterAuxId ] = {
					type = "filter",
					fileSize = 64157,
					digest = filterDigest,
					sha256 = "447446a72d9466febcff65c3b74aa6ebce7562cc39c2d6188e3cc12514c272a2"
				}
			},
			develop = {
				processingModel = "lightroom",
				xmpCameraRaw = {
					sha256 = "99b1dd7ad29e57b7c016aab7390997fb70c0b767112f9075aec7910facfc29de"
				},
				masks = {
					[ maskAuxId ] = {
						digest = maskDigest
					},
				},
				filters = {
					[ "7A993C9D22BD72BFB1A45E3DFC7D95BA" ] = {
						[ filterAuxId ] = {
							digest = filterDigest,
              				crsSizeX = 1024,
							crsSizeY = 768,
							sourceLongEdge = 6000
						}
					}
				}
			}
		}

	)

	local sendNUpdates = rxTest:observe( pusher:_DEBUG_sendNUpdates( 1 ) )
	sendNUpdates:expectCompletedWithinTime( 30 )
	sendNUpdates:dispose()

	local observeAssetOnOz = observeHasServerProvidedMetadata( rxTest, catalog, assetId )
	observeAssetOnOz:expectDataWithinTime( 40, true )
	observeAssetOnOz:dispose()

	rxTest:expectSingleValue(

		catalog:docStore():changedDocs { fromChangeCounter = 0 }
			:map( remapChangedDocsFeed ),

		{
			{
				_labels = {
					oz = true,
				},
				_localOnly = {
					mumble = "stuff",
				},
				_serverProvidedMetadata = {
					base = ">>base",
					created = ">>created",
					links = ">>links",
					revision_ids = ">>revision_ids",
					updated = ">>updated",
				},
				captureDate = "2004-11-02T11:10:09",
				importSource = {
					fileName = "fakeFile01.jpg",
					importedBy = catalogSpec:ozUserId(),
					importTimestamp = "2015-11-02T08:37:24.513Z",
					originalHeight = 16,
					originalWidth = 16,
				},
				subtype = "image",
				type = "asset",
				aux = {
					[ maskAuxId ] = {
						type = "mask",
						fileSize = 64157,
						digest = maskDigest,
						sha256 = "83b7295d955f5449a6001d7e78b947944c2cba4f97bfcac12cfb28d21664984a"
					},
					[ filterAuxId ] = {
						type = "filter",
						fileSize = 64157,
						digest = filterDigest,
						sha256 = "447446a72d9466febcff65c3b74aa6ebce7562cc39c2d6188e3cc12514c272a2"
					}
				},
				develop = {
					processingModel = "lightroom",
					xmpCameraRaw = {
						sha256 = "99b1dd7ad29e57b7c016aab7390997fb70c0b767112f9075aec7910facfc29de"
					},
					masks = {
						[ maskAuxId ] = {
							digest = maskDigest
						},
					},
					filters = {
						[ "7A993C9D22BD72BFB1A45E3DFC7D95BA" ] = {
							[ filterAuxId ] = {
								digest = filterDigest,
								crsSizeX = 1024,
								crsSizeY = 768,
								sourceLongEdge = 6000
							}
						}
					}
				}
			},
			{
				_labels = {
					oz = true,
				},
				_serverProvidedMetadata = {
					base = ">>base",
					created = ">>created",
					links = ">>links",
					revision_ids = ">>revision_ids",
					updated = ">>updated",
				},
				name = "WFCatTestSession",
				subtype = "lightroom",
				type = "catalog",
			},
			continueFrom = 4,
		} )
		
	-- Read the update feed directly from Oz, too.

	rxTest:expectSingleValue(

		catalogSpec:get( string.format( '%s/updates?updated_since=2008-08-17T09:55:34Z', catalogSpec:catalogHref() ) )
			:map( KSJsonDecoder.decodeWebSafeJson )
			:map( stripOzChangesFeed ),

		{
			base = "https://photos-stage.adobe.io/v2/catalogs/"
						.. catalog:ozCatalogSpec():catalogId() .. "/",
			updates = {
				{
					created = ">>created",
					id = ">>id",
					links = ">>links",
					payload = {
						captureDate = "2004-11-02T11:10:09",
						importSource = {
							fileName = "fakeFile01.jpg",
							importTimestamp = "2015-11-02T08:37:24.513Z",
							importedBy = catalogSpec:ozUserId(),
							originalHeight = 16,
							originalWidth = 16,
						},
						aux = {
							[ maskAuxId ] = {
								type = "mask",
								fileSize = 64157,
								digest = maskDigest,
								sha256 = "83b7295d955f5449a6001d7e78b947944c2cba4f97bfcac12cfb28d21664984a"
							},
							[ filterAuxId ] = {
								type = "filter",
								fileSize = 64157,
								digest = filterDigest,
								sha256 = "447446a72d9466febcff65c3b74aa6ebce7562cc39c2d6188e3cc12514c272a2"
							}
						},
						develop = {
							processingModel = "lightroom",
							xmpCameraRaw = {
								sha256 = "99b1dd7ad29e57b7c016aab7390997fb70c0b767112f9075aec7910facfc29de"
							},
							masks = {
								[ maskAuxId ] = {
									digest = maskDigest
								},
							},
							filters = {
								[ "7A993C9D22BD72BFB1A45E3DFC7D95BA" ] = {
									[ filterAuxId ] = {
										digest = filterDigest,
										crsSizeX = 1024,
										crsSizeY = 768,
										sourceLongEdge = 6000
									}
								}
							}
						}
					},
					revision_ids = ">>revision_ids",
					subtype = "image",
					type = "asset",
					updated = ">>updated",
				}
			}
		} )

end

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

function WFCatMetadataPusherTest.rxTestSendUpdates_VideoAsset_testNightly( rxTest )

	rxTest:setTimeLimit( 120 )

	local catalogTestSession = WFCatTestSession( rxTest )

	local catalog, catalogSpec, catalogId, docStore

	rxTest:waitFor(

		catalogTestSession:catalog()
			:map ( function( _catalog )
				check( 'WFCatalog' )
				catalog = _catalog
				catalogSpec = _catalog:ozCatalogSpec()
				catalogId = _catalog:ozCatalogSpec():catalogId()
				docStore = catalog:docStore()
				return _catalog
			end ) )

	local assetId = KSIdGenerator.uuidInLowercaseHex()
	local puller = WFCatMetadataPuller( catalog, "Test Puller", options )
	local pusher = WFCatMetadataPusher( catalog, options )

	puller:getUpdatesContinuously()

	rxTest:waitFor(
		catalog:docStore():setDocValue {
			_id = assetId,
			captureDate = "2004-11-02T11:10:09",
			importSource = {
				importTimestamp = '2016-01-29T15:53:04Z',
				importedBy = catalogSpec:ozUserId(),
				fileName = "IMG_7772.mp4",
				originalHeight = 480,
				originalWidth = 270,
			},
			video = {
				frameRate = { 93817 , 3215 },
				duration = { 416, 125 }
			},
			type = 'asset',
			subtype = 'video',
		}
	)

	local sendNUpdates = rxTest:observe( pusher:_DEBUG_sendNUpdates( 1 ) )
	sendNUpdates:expectCompletedWithinTime( 30 )
	sendNUpdates:dispose()

	local observeAssetOnOz = observeHasServerProvidedMetadata( rxTest, catalog, assetId )
	observeAssetOnOz:expectDataWithinTime( 40, true )
	observeAssetOnOz:dispose()

	rxTest:waitFor (
		catalog:docStore():changedDocs { fromChangeCounter = 0 }
			:map( remapChangedDocsFeed ),
		{
			{
				_labels = {
					oz = true,
				},
				_serverProvidedMetadata = {
					base = ">>base",
					created = ">>created",
					links = ">>links",
					revision_ids = ">>revision_ids",
					updated = ">>updated",
				},
				captureDate = "2004-11-02T11:10:09",
				importSource = {
					importTimestamp = '2016-01-29T15:53:04Z',
					importedBy = catalogSpec:ozUserId(),
					fileName = "IMG_7772.mp4",
					originalHeight = 480,
					originalWidth = 270,
				},
				video = {
					frameRate = { 93817 , 3215 },
					duration = { 416, 125 }
				},
				type = 'asset',
				subtype = 'video',
			},
			{
				_labels = {
					oz = true,
				},
				_serverProvidedMetadata = {
					base = ">>base",
					created = ">>created",
					links = ">>links",
					revision_ids = ">>revision_ids",
					updated = ">>updated",
				},
				name = "WFCatTestSession",
				subtype = "lightroom",
				type = "catalog",
			},
		})

	-- Now edit the video locally and pusher will sent those updates to Oz.
	local develop = {
		version = 1,
		processVersion = 1,
		adjustments = {
			userUpdated = "2020-09-15T12:18:38.23Z",
			adjustmentsFraction = 1000,
			exposure2012 = 0,
			contrast2012 = 0,
			highlights2012 = 0,
			shadows2012 = 0,
			whites2012 = 0,
			blacks2012 = 0,
			incrementalTemperature = 0,
			incrementalTint = 0,
			vibrance = 0,
			saturation = 0,
			hueAdjustmentRed = 0,
			hueAdjustmentOrange = 0,
			hueAdjustmentYellow = 0,
			hueAdjustmentGreen = 0,
			hueAdjustmentAqua = 0,
			hueAdjustmentBlue = 0,
			hueAdjustmentPurple = 0,
			hueAdjustmentMagenta = 0,
			saturationAdjustmentRed = 0,
			saturationAdjustmentOrange = 0,
			saturationAdjustmentYellow = 0,
			saturationAdjustmentGreen = 0,
			saturationAdjustmentAqua = 0,
			saturationAdjustmentBlue = 0,
			saturationAdjustmentPurple = 0,
			saturationAdjustmentMagenta = 0,
			luminanceAdjustmentRed = 0,
			luminanceAdjustmentOrange = 0,
			luminanceAdjustmentYellow = 0,
			luminanceAdjustmentGreen = 0,
			luminanceAdjustmentAqua = 0,
			luminanceAdjustmentBlue = 0,
			luminanceAdjustmentPurple = 0,
			luminanceAdjustmentMagenta = 0,
			splitToningHighlightHue = 0,
			splitToningHighlightSaturation = 0,
			splitToningBalance = 0,
			splitToningShadowHue = 0,
			splitToningShadowSaturation = 0,
			colorGradeMidtoneHue = 0,
			colorGradeMidtoneSat = 0,
			colorGradeShadowLum = 0,
			colorGradeMidtoneLum = 0,
			colorGradeHighlightLum = 0,
			colorGradeBlending = 0,
			colorGradeGlobalHue = 0,
			colorGradeGlobalSat = 0,
			colorGradeGlobalLum = 0,
			postCropVignetteAmount = 0,
			postCropVignetteRoundness = 0,
			postCropVignetteMidpoint = 50000,
			postCropVignetteFeather = 0,
			postCropVignetteHighlightContrast = 50000,
			grainAmount = 0,
			grainSize = 25000,
			grainFrequency = 50000,
			lutAmount = 0,
			profile = "neutral",
			geometric = {
				cropZoom = 100000,
				cropOffsetX = 0,
				cropOffsetY = 0,
				cropRectTop = 0,
				cropRectLeft = 0,
				cropRectBottom = 1000,
				cropRectRight = 1000,
				cropRotation = 0,
				cropFraction = 100000,
				orientation = "normal"
			}
		},
		temporal = {
			userUpdated = "2020-07-25T00:00:42.12Z",
			temporalFraction = 100000000,
			inPoint = 0,
			duration = 10000000
		}
	}

	rxTest:waitFor(
		catalog:docStore():applyOverlay {
			_id = assetId,
			captureDate = "2004-11-02T11:10:09",
			importSource = {
				importTimestamp = '2016-01-29T15:53:04Z',
				importedBy = catalogSpec:ozUserId(),
				fileName = "IMG_7772.mp4",
				originalHeight = 480,
				originalWidth = 270,
			},
			develop = develop,
			video = {
				frameRate = { 93817 , 3215 },
				duration = { 416, 125 }
			},
			type = 'asset',
			subtype = 'video',
		}
	)

	sendNUpdates = rxTest:observe( pusher:_DEBUG_sendNUpdates( 1 ) )
	sendNUpdates:expectCompletedWithinTime( 30 )
	sendNUpdates:dispose()

	rxTest:waitFor (
		catalog:docStore():changedDocs { fromChangeCounter = 0 }
			:map( remapChangedDocsFeed ),
		{
			{
				_labels = {
					oz = true,
				},
				_serverProvidedMetadata = {
					base = ">>base",
					created = ">>created",
					links = ">>links",
					revision_ids = ">>revision_ids",
					updated = ">>updated",
				},
				captureDate = "2004-11-02T11:10:09",
				importSource = {
					importTimestamp = '2016-01-29T15:53:04Z',
					importedBy = catalogSpec:ozUserId(),
					fileName = "IMG_7772.mp4",
					originalHeight = 480,
					originalWidth = 270,
				},
				video = {
					frameRate = { 93817 , 3215 },
					duration = { 416, 125 }
				},
				type = 'asset',
				subtype = 'video',
			},

			{
				_labels = {
					oz = true,
				},
				_serverProvidedMetadata = {
					base = ">>base",
					created = ">>created",
					links = ">>links",
					revision_ids = ">>revision_ids",
					updated = ">>updated",
				},
				captureDate = "2004-11-02T11:10:09",
				importSource = {
					importTimestamp = '2016-01-29T15:53:04Z',
					importedBy = catalogSpec:ozUserId(),
					fileName = "IMG_7772.mp4",
					originalHeight = 480,
					originalWidth = 270,
				},
				develop = develop,
				video = {
					frameRate = { 93817 , 3215 },
					duration = { 416, 125 }
				},
				type = 'asset',
				subtype = 'video',

			},

			{
				_labels = {
					oz = true,
				},
				_serverProvidedMetadata = {
					base = ">>base",
					created = ">>created",
					links = ">>links",
					revision_ids = ">>revision_ids",
					updated = ">>updated",
				},
				name = "WFCatTestSession",
				subtype = "lightroom",
				type = "catalog",
			},
		})


end

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

function WFCatMetadataPusherTest.reportAllChanges( scanState, docValue )
	return scanState, docValue
end

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

function WFCatMetadataPusherTest.rxTestSendUpdates_retryAfter_testNightly( rxTest )

	rxTest:setTimeLimit( 120 )

	local catalogTestSession = WFCatTestSession( rxTest )

	local catalog = rxTest:getFirstValue( catalogTestSession:catalog() )
	local catalogSpec = catalog:ozCatalogSpec()

	local retryAfterSecondsOnError = 10
	local retryAfterSecondsOnSuccess = 15

	local inject429 = false
	local injectRetryAfter = false

	local catalogSpec_put = catalogSpec.put
	rawset( catalogSpec, 'put', function( self, url, data, options )

		local modifiedOptions = KSTableUtils.deepCopy( options )

		if inject429 then

			inject429 = false

			table.insert( modifiedOptions.requestHeaders, {
				header = 'X-Return-429-Status',
				value = 1,
			} )

			table.insert( modifiedOptions.requestHeaders, {
				header = 'X-Return-Retry-After',
				value = retryAfterSecondsOnError,
			} )

		end

		if injectRetryAfter then

			injectRetryAfter = false

			table.insert( modifiedOptions.requestHeaders, {
				header = 'X-Return-Retry-After',
				value = retryAfterSecondsOnSuccess,
			} )

		end

		return catalogSpec_put( self, url, data, modifiedOptions )

	end )

	local puller = WFCatMetadataPuller( catalog, "Test Puller", options )
	rxTest:addCloseTarget( puller )

	local pusher = WFCatMetadataPusher( catalog, options )
	rxTest:addCloseTarget( pusher )

	puller:getUpdatesContinuously()

	-- Very simple test: We push one asset to Oz and then we delete it.

	local assetId = KSIdGenerator.uuidInLowercaseHex()

	rxTest:waitFor(

		catalog:docStore():setDocValue {
			_id = assetId,
			_localOnly = {
				mumble = "stuff",
			},
			captureDate = "2004-11-02T11:10:09",
			importSource = {
				fileName = "fakeFile01.jpg",
				importedBy = catalogSpec:ozUserId(),
				importTimestamp = "2015-11-02T08:37:24.513Z",
				originalHeight = 16,
				originalWidth = 16,
			},
			subtype = "image",
			type = "asset",
		}

	)

	-- Should take longer than the two retry delays added when retrying, also
	-- note that this is actually looking for an error.
	inject429 = true

	rxTest:expectSingleValueWithinTimeLimits( retryAfterSecondsOnError * 2, 60,
		pusher:_DEBUG_sendUpdatesOnce()
		:catch( function( responseData, responseHeader )
			return Rx.Observable.just {
				[ 'Retry-After' ] = responseHeader[ 'Retry-After' ],
				retryCount = responseHeader.retryCount,
				status_code = responseHeader.status_code,
			}
		end ),

		{
			retryCount = 3,
			status_code = '429',
			[ 'Retry-After' ] = tostring( retryAfterSecondsOnError ),
		} )

	-- Try again this time no error will be injected only a Retry-After.
	-- which will cause the follwing update to be delayed, but this
	-- put should complete quickly.
	injectRetryAfter = true

	local sendNUpdates = rxTest:observe( pusher:_DEBUG_sendNUpdates( 1 ) )
	sendNUpdates:expectCompletedWithinTime( 9 )
	sendNUpdates:dispose()

	-- Now add another document so we can send another update.
	local assetId2 = KSIdGenerator.uuidInLowercaseHex()

	rxTest:waitFor(

		catalog:docStore():setDocValue {
			_id = assetId2,
			_localOnly = {
				mumble = "morestuff",
			},
			captureDate = "2005-11-02T11:10:09",
			importSource = {
				fileName = "fakeFile02.jpg",
				importedBy = catalogSpec:ozUserId(),
				importTimestamp = "2015-11-02T08:37:24.513Z",
				originalHeight = 32,
				originalWidth = 32,
			},
			subtype = "image",
			type = "asset",
		}

	)

	-- Send the update, the prior response header sent by Oz should have
	-- requested a delay before we send again, so make sure it takes longer
	-- than the specified delay.
	rxTest:expectSingleValueWithinTimeLimits( retryAfterSecondsOnSuccess, 60, pusher:_DEBUG_sendUpdatesOnce(), 1 )

	-- Wait for the puller to receive all the updates.
	local assetsHaveSPM = rxTest:observe( Rx.Observable.combineLatest( catalog:docStore():ongoingDocValue( assetId ), catalog:docStore():ongoingDocValue( assetId2 ) )
		:map( function( docValue1, docValue2 )
			return not not ( docValue1._serverProvidedMetadata and docValue2._serverProvidedMetadata )
		end ) )

	assetsHaveSPM:expectDataWithinTime( 60, true )
	assetsHaveSPM:dispose()

	rxTest:expectSingleValue(

		catalog:docStore():changedDocs { fromChangeCounter = 3 }
			:map( remapChangedDocsFeedWithDocId ),
			sortByDocId( {
				{
					_id = assetId2,
					_labels = {
						oz = true,
					},
					_localOnly = {
						mumble = "morestuff",
					},
					_serverProvidedMetadata = {
						base = ">>base",
						created = ">>created",
						links = ">>links",
						revision_ids = ">>revision_ids",
						updated = ">>updated",
					},
					captureDate = "2005-11-02T11:10:09",
					importSource = {
						fileName = "fakeFile02.jpg",
						importedBy = catalogSpec:ozUserId(),
						importTimestamp = "2015-11-02T08:37:24.513Z",
						originalHeight = 32,
						originalWidth = 32,
					},
					subtype = "image",
					type = "asset",
				},
				{
					_id = assetId,
					_labels = {
						oz = true,
					},
					_localOnly = {
						mumble = "stuff",
					},
					_serverProvidedMetadata = {
						base = ">>base",
						created = ">>created",
						links = ">>links",
						revision_ids = ">>revision_ids",
						updated = ">>updated",
					},
					captureDate = "2004-11-02T11:10:09",
					importSource = {
						fileName = "fakeFile01.jpg",
						importedBy = catalogSpec:ozUserId(),
						importTimestamp = "2015-11-02T08:37:24.513Z",
						originalHeight = 16,
						originalWidth = 16,
					},
					subtype = "image",
					type = "asset",
				},
				continueFrom = 6,
			} ) )

end

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

function WFCatMetadataPusherTest.rxTestSendUpdates_retryAfterContinuous_testNightly( rxTest )

-- The purpose of this test is to make sure the WFCatMetadataPusher recovers from 429 errors
-- when run in continous mode.  So it doesn't check for any delays, delays are checked
-- in a different test.

	rxTest:setTimeLimit( 120 )

	local catalogTestSession = WFCatTestSession( rxTest )

	local catalog = rxTest:getFirstValue( catalogTestSession:catalog() )
	local catalogSpec = catalog:ozCatalogSpec()

	local inject429 = false
	local injectRetryAfter = false

	local retryAfterSecondsOnError = 1
	local retryAfterSecondsOnSuccess = 1

	local sawError429 = false
	local sawRetryAfterWithoutError = false

	local catalogSpec_put = catalogSpec.put
	rawset( catalogSpec, 'put', function( self, url, data, options )

		local modifiedOptions = KSTableUtils.deepCopy( options )

		if inject429 then

			inject429 = false
			injectRetryAfter = true

			table.insert( modifiedOptions.requestHeaders, {
				header = 'X-Return-429-Status',
				value = 1,
			} )

			table.insert( modifiedOptions.requestHeaders, {
				header = 'X-Return-Retry-After',
				value = retryAfterSecondsOnError,
			} )

		elseif injectRetryAfter then

			injectRetryAfter = false

			table.insert( modifiedOptions.requestHeaders, {
				header = 'X-Return-Retry-After',
				value = retryAfterSecondsOnSuccess,
			} )

		end

		return catalogSpec_put( self, url, data, modifiedOptions )
			:tap( function( responseData, responseHeaders )
				if responseHeaders[ 'Retry-After' ] then
					sawRetryAfterWithoutError = true
				end

			end, function( responseData, responseHeaders )
				if responseHeaders.status_code == '429' and responseHeaders[ 'Retry-After' ] then
					sawError429 = true
				end
			end )

	end )


	local puller = WFCatMetadataPuller( catalog, "Test Puller", options )
	rxTest:addCloseTarget( puller )

	local pusherOptions = KSTableUtils.deepCopy( options )
	pusherOptions.delayBeforeSendingUpdates = 0.01

	local pusher = WFCatMetadataPusher( catalog, pusherOptions )

	rxTest:addCloseTarget( pusher )

	-- Prime the updates feed pump.

	rxTest:expectOnlyCompleted( puller:getUpdatesOnce() )

	-- Very simple test: We push one asset to Oz and then we delete it.

	local assetId = KSIdGenerator.uuidInLowercaseHex()

	rxTest:waitFor(

		catalog:docStore():setDocValue {
			_id = assetId,
			_localOnly = {
				mumble = "stuff",
			},
			captureDate = "2004-11-02T11:10:09",
			importSource = {
				fileName = "fakeFile01.jpg",
				importedBy = catalogSpec:ozUserId(),
				importTimestamp = "2015-11-02T08:37:24.513Z",
				originalHeight = 16,
				originalWidth = 16,
			},
			subtype = "image",
			type = "asset",
		}

	)

	local updateCount = rxTest:observe( pusher:countUpdatesToSend() )

	-- The countUpdatesToSend checks only for the continueFrom
	-- and doesn't look at the oz label. Hence even though
	-- the first change happened due to fetching of catalog
	-- document, and it doesnt have to be sent up, the func
	-- still includes that count as well.
	updateCount:expectData( 2 )

	inject429 = true
	rxTest:waitFor( pusher:sendUpdatesContinuously() )
	rxTest:finally( function()
		puller:stopContinuousUpdates()
	end )

	-- wait for the pusher to finish pushing up the first document
	-- so there will be two updates sent.

	updateCount:expectDataWithinTime( 40, 0 )

	puller:getUpdatesContinuously()
		-- Start the puller

	-- Now add another document so we can send another update.
	local assetId2 = KSIdGenerator.uuidInLowercaseHex()
	rxTest:waitFor(

		catalog:docStore():setDocValue {
			_id = assetId2,
			_localOnly = {
				mumble = "morestuff",
			},
			captureDate = "2005-11-02T11:10:09",
			importSource = {
				fileName = "fakeFile02.jpg",
				importedBy = catalogSpec:ozUserId(),
				importTimestamp = "2015-11-02T08:37:24.513Z",
				originalHeight = 32,
				originalWidth = 32,
			},
			subtype = "image",
			type = "asset",
		}

	)

	-- Wait for the puller to receive all the updates.

	local numUpdatesReceived = 0
	local numUpdatesExpected = 5

	local numUpdatesObserver = rxTest:observe(
		catalog:docStore():scannedFilteredMappedChangesFeedSince( "WFCatMetadataPusherTest" , "reportAllChanges", {}, 0 )
			:map( function( doc )
				numUpdatesReceived = numUpdatesReceived + 1
				return numUpdatesReceived
			end ) )

	numUpdatesObserver:expectDataWithinTime( 40, numUpdatesExpected )

	numUpdatesObserver:dispose()

	rxTest:expectSingleValue(

		catalog:docStore():changedDocs { fromChangeCounter = 3 }
			:map( remapChangedDocsFeedWithDocId ),
			sortByDocId( {
				{
					_id = assetId2,
					_labels = {
						oz = true,
					},
					_localOnly = {
						mumble = "morestuff",
					},
					_serverProvidedMetadata = {
						base = ">>base",
						created = ">>created",
						links = ">>links",
						revision_ids = ">>revision_ids",
						updated = ">>updated",
					},
					captureDate = "2005-11-02T11:10:09",
					importSource = {
						fileName = "fakeFile02.jpg",
						importedBy = catalogSpec:ozUserId(),
						importTimestamp = "2015-11-02T08:37:24.513Z",
						originalHeight = 32,
						originalWidth = 32,
					},
					subtype = "image",
					type = "asset",
				},
				{
					_id = assetId,
					_labels = {
						oz = true,
					},
					_localOnly = {
						mumble = "stuff",
					},
					_serverProvidedMetadata = {
						base = ">>base",
						created = ">>created",
						links = ">>links",
						revision_ids = ">>revision_ids",
						updated = ">>updated",
					},
					captureDate = "2004-11-02T11:10:09",
					importSource = {
						fileName = "fakeFile01.jpg",
						importedBy = catalogSpec:ozUserId(),
						importTimestamp = "2015-11-02T08:37:24.513Z",
						originalHeight = 16,
						originalWidth = 16,
					},
					subtype = "image",
					type = "asset",
				},
				continueFrom = 6,
			} ) )

	rxTest:expectEqual( true, sawError429 )
	rxTest:expectEqual( true, sawRetryAfterWithoutError )

end

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

function WFCatMetadataPusherTest.rxTestSendSpaces_testNightly( rxTest )

	rxTest:setTimeLimit( 120 )

	local spaceId = KSIdGenerator.uuidInLowercaseHex()

	local catalogTestSession = WFCatTestSession( rxTest )

	local catalog = rxTest:getFirstValue( catalogTestSession:catalog() )
	local catalogSpec = catalog:ozCatalogSpec()

	local puller = WFCatMetadataPuller( catalog, "Test Puller", options )
	rxTest:addCloseTarget( puller )

	puller:getUpdatesContinuously()

	local pusher = WFCatMetadataPusher( catalog, options )
	rxTest:addCloseTarget( pusher )

	local ozUpdatedDate
	local isSpaceOnOz = rxTest:observe( catalog:docStore():ongoingDocValue( spaceId )
			:filter( function( docValue )
				return docValue._serverProvidedMetadata ~= nil
			end )
			:map( function( docValue )
				ozUpdatedDate = KSTableUtils.derefOpt( docValue, '_serverProvidedMetadata', 'updated' )
				return docValue._serverProvidedMetadata ~= nil
			end )
			:take(1) )

	-- Create space
	rxTest:waitFor( catalog:docStore():applyOverlay {
		_id = spaceId,
		type = 'space',
		subtype = 'shared',
		private = false,
		download = false,
		metadata = false,
		location = false,
	} )

	-- Send space to Oz

	local sendNUpdates = rxTest:observe( pusher:_DEBUG_sendNUpdates( 1 ) )
	sendNUpdates:expectCompletedWithinTime( 30 )
	sendNUpdates:dispose()

	isSpaceOnOz:expectDataWithinTime( 40, true )

	-- Update space
	rxTest:waitFor( catalog:docStore():applyOverlay {
		_id = spaceId,
		private = true,
	} )

	sendNUpdates = rxTest:observe( pusher:_DEBUG_sendNUpdates( 1 ) )
	sendNUpdates:expectCompletedWithinTime( 30 )
	sendNUpdates:dispose()

	local isSpacePrivateOnOz = rxTest:observe( catalog:docStore():ongoingDocValue( spaceId )
			:map( function( docValue )
				local newUpdatedDate = KSTableUtils.derefOpt( docValue, '_serverProvidedMetadata', 'updated' )
				local result = ozUpdatedDate and newUpdatedDate and ( ozUpdatedDate < newUpdatedDate ) and docValue.private
				ozUpdatedDate = newUpdatedDate
				return result
			end ) )

	isSpacePrivateOnOz:expectDataWithinTime( 40, true )

	-- Make sure space exists via REST API

	rxTest:expectSingleValue(
		catalogSpec:get(
			string.format( '%s/spaces/%s', catalogSpec:catalogHref(), spaceId ),
			{ returnHttpErrorsAsData = true, }
		)

		:map( function( data, headers )

			return headers.status_code == '200'

		end ),

		true )

	-- Delete space

	rxTest:waitFor( catalog:docStore():applyOverlay {
		_id = spaceId,
		_deleted = true,
	} )

	sendNUpdates = rxTest:observe( pusher:_DEBUG_sendNUpdates( 1 ) )
	sendNUpdates:expectCompletedWithinTime( 30 )
	sendNUpdates:dispose()

	local isSpaceDeleted = observeIsDeletedOnOz( rxTest, catalog, spaceId )
	isSpaceDeleted:expectDataWithinTime( 40, true )
	isSpaceDeleted:dispose()

	rxTest:expectSingleValue(

		catalogSpec:get(
			string.format( '%s/spaces/%s', catalogSpec:catalogHref(), spaceId ),
			{ returnHttpErrorsAsData = true, }
		)

		:map( function( data, headers )

			return headers.status_code == '404'

		end ), true )

end

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

function WFCatMetadataPusherTest.rxTestPutAlbumIntoSpace_testNightly( rxTest )

	rxTest:setTimeLimit( 120 )

	local spaceId = KSIdGenerator.uuidInLowercaseHex()
	local albumId = KSIdGenerator.uuidInLowercaseHex()
	local spaceAlbumId = KSIdGenerator.uuidInLowercaseHex()

	local catalogTestSession = WFCatTestSession( rxTest )

	local catalog = rxTest:getFirstValue( catalogTestSession:catalog() )
	local catalogSpec = catalog:ozCatalogSpec()

	local puller = WFCatMetadataPuller( catalog, "Test Puller", options )

	local pusher = WFCatMetadataPusher( catalog, options )

	puller:getUpdatesContinuously()

	rxTest:finally( function()

		puller:close()
		pusher:close()

	end )

	-- Create space & album

	rxTest:waitFor( catalog:docStore():applyOverlay {
		_id = spaceId,
		type = 'space',
		subtype = 'shared',
		private = false,
		download = false,
		metadata = false,
		location = false,
	} )

	local sendNUpdates = rxTest:observe( pusher:_DEBUG_sendNUpdates( 1 ) )
	sendNUpdates:expectCompletedWithinTime( 30 )
	sendNUpdates:dispose()

	local observeOnOz = observeIsOnOz( rxTest, catalog, spaceId )
	observeOnOz:expectDataWithinTime( 60, true )
	observeOnOz:dispose()

	rxTest:waitFor( catalog:docStore():applyOverlay {
		_id = albumId,
		type = 'album',
		subtype = 'collection',
		name = 'test album',
	} )

	sendNUpdates = rxTest:observe( pusher:_DEBUG_sendNUpdates( 1 ) )
	sendNUpdates:expectCompletedWithinTime( 30 )
	sendNUpdates:dispose()

	observeOnOz = observeIsOnOz( rxTest, catalog, albumId )
	observeOnOz:expectDataWithinTime( 60, true )
	observeOnOz:dispose()

	rxTest:waitFor( catalog:docStore():applyOverlay {
		_id = spaceAlbumId,
		type = 'space_album',
		album = {
			id = albumId,
		},
		space = {
			id = spaceId,
		},

	} )

	-- Send changes to Oz

	sendNUpdates = rxTest:observe( pusher:_DEBUG_sendNUpdates( 1 ) )
	sendNUpdates:expectCompletedWithinTime( 30 )
	sendNUpdates:dispose()

	observeOnOz = observeIsOnOz( rxTest, catalog, spaceAlbumId )
	observeOnOz:expectDataWithinTime( 60, true )
	observeOnOz:dispose()

	-- Verify album was added to space

	rxTest:expectSingleValue(

		catalogSpec:get(
			string.format( '%sspaces/%s/resources', catalogSpec:baseHref(), spaceId ),
			{ returnHttpErrorsAsData = true, }
		)

		:map( KSJsonDecoder.decodeWebSafeJson_stripNull_ignoreEmpty )

		:map( function( spaceDocValue )

			local albumId = KSTableUtils.derefOpt( spaceDocValue, 'resources', 1, 'id' )

			return albumId

		end ), albumId )

end

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

function WFCatMetadataPusherTest.rxTestCommentsAndFavorites_testNightly( rxTest )

	rxTest:setTimeLimit( 120 )

	local catalogTestSession = WFCatTestSession( rxTest )

	local catalog
	local spaceId = KSIdGenerator.uuidInLowercaseHex()
	local spaceAlbumId = KSIdGenerator.uuidInLowercaseHex()
	local albumId = KSIdGenerator.uuidInLowercaseHex()
	local albumAssetId = KSIdGenerator.uuidInLowercaseHex()
	local assetId = KSIdGenerator.uuidInLowercaseHex()
	local commentId = KSIdGenerator.uuidInLowercaseHex()
	local favoriteId = KSIdGenerator.uuidInLowercaseHex()

	rxTest:waitFor(

		catalogTestSession:catalog()
			:map( function( _catalog )
				check( 'WFCatalog' )
				catalog = _catalog
				return _catalog
			end )

	)

	local puller = WFCatMetadataPuller( catalog, "Test Puller", options )
	rxTest:addCloseTarget( puller )

	puller:getUpdatesContinuously()

	local pusher = WFCatMetadataPusher( catalog, options )
	rxTest:addCloseTarget( pusher )

	-- Very simple test: We push one asset to Oz and then we delete it.

	local catalogSpec

	rxTest:waitFor(

		catalogTestSession:catalogSpec()
			:map( function( _catalogSpec )
				check( 'WFCatSpec' )
				catalogSpec = _catalogSpec
				return _catalogSpec
			end )

	)

	rxTest:waitFor(

		catalog:docStore():setDocValue {
			_id = assetId,
			_localOnly = {
				mumble = "stuff",
			},
			captureDate = "2004-11-02T11:10:09",
			importSource = {
				fileName = "fakeFile01.jpg",
				importedBy = catalogSpec:ozUserId(),
				importTimestamp = "2015-11-02T08:37:24.513Z",
				originalHeight = 16,
				originalWidth = 16,
			},
			subtype = "image",
			type = "asset",
		}

	)

	local sendNUpdates = rxTest:observe( pusher:_DEBUG_sendNUpdates( 1 ) )
	sendNUpdates:expectCompletedWithinTime( 30 )
	sendNUpdates:dispose()

	puller:checkOzForUpdates()

	local observeUpdateFromOz = observeHasServerProvidedMetadata( rxTest, catalog, assetId )
	observeUpdateFromOz:expectDataWithinTime( 40, true )
	observeUpdateFromOz:dispose()

	-- Create space & album

	rxTest:waitFor( catalog:docStore():applyOverlay {
		_id = albumId,
		type = 'album',
		subtype = 'collection',
		name = 'test album',
	} )

	sendNUpdates = rxTest:observe( pusher:_DEBUG_sendNUpdates( 1 ) )
	sendNUpdates:expectCompletedWithinTime( 30 )
	sendNUpdates:dispose()

	local observeOnOz = observeIsOnOz( rxTest, catalog, albumId )
	observeOnOz:expectDataWithinTime( 40, true )
	observeOnOz:dispose()

	rxTest:waitFor( catalog:docStore():applyOverlay {
		_id = spaceId,
		type = 'space',
		subtype = 'shared',
		private = false,
		download = false,
		metadata = false,
		location = false,
	} )

	sendNUpdates = rxTest:observe( pusher:_DEBUG_sendNUpdates( 1 ) )
	sendNUpdates:expectCompletedWithinTime( 30 )
	sendNUpdates:dispose()

	puller:checkOzForUpdates()

	observeUpdateFromOz = observeIsOnOz( rxTest, catalog, spaceId )
	observeUpdateFromOz:expectDataWithinTime( 40, true )
	observeUpdateFromOz:dispose()

	rxTest:waitFor( catalog:docStore():applyOverlay {
		_id = spaceAlbumId,
		type = 'space_album',
		album = {
			id = albumId,
		},
		space = {
			id = spaceId,
		},
	} )

	sendNUpdates = rxTest:observe( pusher:_DEBUG_sendNUpdates( 1 ) )
	sendNUpdates:expectCompletedWithinTime( 30 )
	sendNUpdates:dispose()

	puller:checkOzForUpdates()

	observeUpdateFromOz = observeHasServerProvidedMetadata( rxTest, catalog, spaceAlbumId )
	observeUpdateFromOz:expectDataWithinTime( 40, true )
	observeUpdateFromOz:dispose()

	rxTest:waitFor( catalog:docStore():applyOverlay {
		_id = albumAssetId,
		type = 'album_asset',
		album = {
			id = albumId,
		},
		asset = {
			id = assetId,
		},

	} )

	sendNUpdates = rxTest:observe( pusher:_DEBUG_sendNUpdates( 1 ) )
	sendNUpdates:expectCompletedWithinTime( 30 )
	sendNUpdates:dispose()

	puller:checkOzForUpdates()

	observeUpdateFromOz = observeHasServerProvidedMetadata( rxTest, catalog, albumAssetId )
	observeUpdateFromOz:expectDataWithinTime( 40, true )
	observeUpdateFromOz:dispose()

	-- Add comment and favorite to asset in space

	rxTest:waitFor( catalog:docStore():bulkApplyOverlay {
		{
			_id = commentId,
			type = 'comment',
			subtype = 'user',
			body = 'test comment',
			space = {
				id = spaceId,
			},
			asset = {
				id = assetId,
			}
		},
		{
			_id = favoriteId,
			type = 'favorite',
			subtype = 'user',
			space = {
				id = spaceId,
			},
			asset = {
				id = assetId,
			}
		}
	} )

	sendNUpdates = rxTest:observe( pusher:_DEBUG_sendNUpdates( 2 ) )
	sendNUpdates:expectCompletedWithinTime( 30 )
	sendNUpdates:dispose()

	puller:checkOzForUpdates()

	-- Also verify comment and favorite have synced
	local commentDocOnOz = observeHasServerProvidedMetadata( rxTest, catalog, commentId )
	commentDocOnOz:expectDataWithinTime( 40, true )
	commentDocOnOz:dispose()

	local favoriteDocOnOz = observeHasServerProvidedMetadata( rxTest, catalog, favoriteId )
	favoriteDocOnOz:expectDataWithinTime( 40, true )
	favoriteDocOnOz:dispose()

	-- Verify favorite and comment made it to Oz via rest APIs
	rxTest:expectSingleValue(

		catalogSpec:get(
			string.format( '%sspaces/%s/assets/%s/favorites', catalogSpec:baseHref(), spaceId, assetId ),
			{ returnHttpErrorsAsData = true, }
		)

		:map( KSJsonDecoder.decodeWebSafeJson_stripNull_ignoreEmpty )

		:map( function( spaceDocValue )

			local resId = KSTableUtils.derefOpt( spaceDocValue, 'resources', 1, 'id' )

			return resId

		end ), favoriteId )

	rxTest:expectSingleValue(

		catalogSpec:get(
			string.format( '%sspaces/%s/assets/%s/comments', catalogSpec:baseHref(), spaceId, assetId ),
			{ returnHttpErrorsAsData = true, }
		)

		:map( KSJsonDecoder.decodeWebSafeJson_stripNull_ignoreEmpty )

		:map( function( spaceDocValue )

			local resId = KSTableUtils.derefOpt( spaceDocValue, 'resources', 1, 'id' )

			return resId

		end ), commentId )

	rxTest:waitFor( catalog:docStore():bulkApplyOverlay {
		{
			_id = commentId,
			_deleted = true,
		},
		{
			_id = favoriteId,
			_deleted = true,
		}
	} )

	sendNUpdates = rxTest:observe( pusher:_DEBUG_sendNUpdates( 2 ) )
	sendNUpdates:expectCompletedWithinTime( 30 )
	sendNUpdates:dispose()

	puller:checkOzForUpdates()

	local observeCommentDeletedOnOz = observeIsDeletedOnOz( rxTest, catalog, commentId )
	observeCommentDeletedOnOz:expectDataWithinTime( 40, true )
	observeCommentDeletedOnOz:dispose()

	local observeFavoriteDeletedOnOz = observeIsDeletedOnOz( rxTest, catalog, favoriteId, true )
	observeFavoriteDeletedOnOz:expectDataWithinTime( 40, true )
	observeFavoriteDeletedOnOz:dispose()

	rxTest:expectSingleValue(

		catalogSpec:get(
			string.format( '%sspaces/%s/assets/%s/favorites', catalogSpec:baseHref(), spaceId, assetId ),
			{ returnHttpErrorsAsData = true, }
		)

		:map( KSJsonDecoder.decodeWebSafeJson_stripNull_ignoreEmpty )

		:map( function( favoritesDocValue )

			local resources = KSTableUtils.derefOpt( favoritesDocValue, 'resources' )

			return #resources

		end ), 0 )

	rxTest:expectSingleValue(

		catalogSpec:get(
			string.format( '%sspaces/%s/assets/%s/comments', catalogSpec:baseHref(), spaceId, assetId ),
			{ returnHttpErrorsAsData = true, }
		)

		:map( KSJsonDecoder.decodeWebSafeJson_stripNull_ignoreEmpty )

		:map( function( commentsDocValue )

			local resources = KSTableUtils.derefOpt( commentsDocValue, 'resources' )

			return #resources

		end ), 0 )

end

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

function WFCatMetadataPusherTest.rxTestSendAlbumAssets_testNightly( rxTest )

	rxTest:setTimeLimit( 120 )

	local catalogTestSession = WFCatTestSession( rxTest )
	local catalog = rxTest:getFirstValue( catalogTestSession:catalog() )

	local albumId = KSIdGenerator.uuidInLowercaseHex()
	local albumAssetId = KSIdGenerator.uuidInLowercaseHex()
	local assetId = KSIdGenerator.uuidInLowercaseHex()

	local puller = WFCatMetadataPuller( catalog, "Test Puller", options )
	rxTest:addCloseTarget( puller )

	puller:getUpdatesContinuously()

	local pusher = WFCatMetadataPusher( catalog, options )
	rxTest:addCloseTarget( pusher )

	-- Create asset
	rxTest:waitFor(

		catalog:docStore():setDocValue {
			_id = assetId,
			_localOnly = {
				mumble = "stuff",
			},
			captureDate = "2004-11-02T11:10:09",
			importSource = {
				fileName = "fakeFile01.jpg",
				importedBy = catalog:ozCatalogSpec():ozUserId(),
				importTimestamp = "2015-11-02T08:37:24.513Z",
				originalHeight = 16,
				originalWidth = 16,
			},
			subtype = "image",
			type = "asset",
		}

	)

	local sendNUpdates = rxTest:observe( pusher:_DEBUG_sendNUpdates( 1 ) )
	sendNUpdates:expectCompletedWithinTime( 30 )
	sendNUpdates:dispose()

	local observeUpdateFromOz = observeHasServerProvidedMetadata( rxTest, catalog, assetId )
	observeUpdateFromOz:expectDataWithinTime( 40, true )
	observeUpdateFromOz:dispose()

	-- Create album and add asset to album
	rxTest:waitFor( catalog:docStore():applyOverlay {
		_id = albumId,
		type = 'album',
		subtype = 'collection',
		name = 'test album',
	} )

	sendNUpdates = rxTest:observe( pusher:_DEBUG_sendNUpdates( 1 ) )
	sendNUpdates:expectCompletedWithinTime( 30 )
	sendNUpdates:dispose()

	observeUpdateFromOz = observeHasServerProvidedMetadata( rxTest, catalog, albumId )
	observeUpdateFromOz:expectDataWithinTime( 40, true )
	observeUpdateFromOz:dispose()

	local localAlbumAssetRev = rxTest:getFirstValue(

		catalog:docStore():applyOverlay {
			_id = albumAssetId,
			type = 'album_asset',
			album = {
				id = albumId,
			},
			asset = {
				id = assetId,
			}
		}

	)

	sendNUpdates = rxTest:observe( pusher:_DEBUG_sendNUpdates( 1 ) )
	sendNUpdates:expectCompletedWithinTime( 30 )
	sendNUpdates:dispose()

	observeUpdateFromOz = observeHasServerProvidedMetadata( rxTest, catalog, albumAssetId )
	observeUpdateFromOz:expectDataWithinTime( 40, true )
	observeUpdateFromOz:dispose()

	-- Verify oz label is set to latest local revId
	rxTest:expectSingleValue(

		catalog:docStore():docValue( albumAssetId, 'oz' )
			:map( function( docValue )

				return docValue._revId

			end ),

		localAlbumAssetRev )


	-- Update album_asset with order and cover informations

	localAlbumAssetRev = rxTest:getFirstValue(
		catalog:docStore():applyOverlay {
			_id = albumAssetId,
			order = 'v',
			cover = true,
		}
	)

	sendNUpdates = rxTest:observe( pusher:_DEBUG_sendNUpdates( 1 ) )
	sendNUpdates:expectCompletedWithinTime( 30 )
	sendNUpdates:dispose()

	local observeAlbumCover = rxTest:observe( catalog:docStore():ongoingDocValue( albumId )
		:map( function( docValue )
			return docValue and docValue.cover and docValue.cover.id
		end ) )

	-- Verify asset became the album cover
	observeAlbumCover:expectDataWithinTime( 40, assetId )

	-- Verify oz label is set to latest local revId

	rxTest:expectSingleValue(

		catalog:docStore():docValue( albumAssetId, 'oz' )
			:map( function( docValue )

				return docValue._revId

			end ),

		localAlbumAssetRev )

	-- Delete album_asset locally and upload to Oz

	rxTest:waitFor( catalog:docStore():applyOverlay {
		_id = albumAssetId,
		_deleted = true,

	} )

	sendNUpdates = rxTest:observe( pusher:_DEBUG_sendNUpdates( 1 ) )
	sendNUpdates:expectCompletedWithinTime( 30 )
	sendNUpdates:dispose()

	-- Ensure album_asset got deleted on Oz
	local observeAlbumAssetIsDeletedOnOz = observeIsDeletedOnOz( rxTest, catalog, albumAssetId )
	observeAlbumAssetIsDeletedOnOz:expectDataWithinTime( 40, true )
	observeAlbumAssetIsDeletedOnOz:dispose()

	-- Delete asset and send to Oz

	rxTest:waitFor( catalog:docStore():applyOverlay {
		_id = assetId,
		_deleted = true,
	}
	 )

	sendNUpdates = rxTest:observe( pusher:_DEBUG_sendNUpdates( 1 ) )
	sendNUpdates:expectCompletedWithinTime( 30 )
	sendNUpdates:dispose()

	-- Ensure asset got deleted on Oz
	local observeAssetIsDeletedOnOz = observeIsDeletedOnOz( rxTest, catalog, assetId )
	observeAssetIsDeletedOnOz:expectDataWithinTime( 40, true )
	observeAssetIsDeletedOnOz:dispose()

	-- Verify album has no longer a cover as asset got deleted
	observeAlbumCover:expectDataWithinTime( 40, nil )

end

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

function WFCatMetadataPusherTest.rxTestSendChangesWhenEmpty_mock( rxTest )

	local catalogId = KSIdGenerator:uuidInLowercaseHex()
	local catalogSpec = WFCatSpecMock( catalogId, kUserId, rxTest,

		{
			-- should be no HTTP comm if no changes to send
		} )

	local catalog = WFCatMockCatalog.createWFCatalogForTestWithSpec( rxTest, catalogSpec )

	local pusher = WFCatMetadataPusher( catalog, options )
	rxTest:addCloseTarget( pusher )

	local updateCount = rxTest:observe( pusher:countUpdatesToSend() )
	updateCount:expectData( 0 )

	-- Add an asset that will generate no sync activity because of
	-- the `dontsendtooz` label.

	local assetId = KSIdGenerator.uuidInLowercaseHex()
	local assetRevId = KSIdGenerator.uuidInLowercaseHex()

	rxTest:waitFor( catalog:docStore():setDocValue( {
				_id = assetId,
				_newRevId = assetRevId,
				_labels = {
					[ '_dontsendtooz' ] = true
				},
				captureDate = "2015-05-17T15:50:12",
				importSource = {
					importTimestamp = "2015-05-26T17:34:23-07:00",
					originalHeight = 1600,
					originalWidth = 1068,
				},
				location = {
					altitude = 377.9,
					latitude = 45.951733,
					longitude = -93.600107,
				},
				subtype = "image",
				type = "asset",
				version = 1,
			} ) )

	updateCount:expectData( 1 )

	-- Attempt to send updates to Oz. There should be none.

	rxTest:waitFor( pusher:_DEBUG_sendNUpdates( 0 ) )

	updateCount:expectData( 0 )

	-- The continueFrom counter should now be at 2, since the chnage counter associated
	-- with the asset insertion is 1.

	rxTest:expectSingleValue( catalog:docStore():localDocValue( 'Oz-sendUpdatesContinueFrom' )
		:pluck( 'continueFrom' ),
		2 )

end

--------------------------------------------------------------------------------
-- Test that a local delete marked "don't send to Oz" will not be sent to Oz,
-- but if a conflict is created on that delete by an update from Oz, the conflict
-- will be resolved and a followup delete *will* be sent to Oz. See WF-311 and
-- linked bugs for more details.
function WFCatMetadataPusherTest.rxTestSendResolvedLocalDeleteToOz_mock( rxTest )

	rxTest:setTimeLimit( 5 )

	local catalogId = KSIdGenerator:uuidInLowercaseHex()
	local catalogUpdatesUrl = string.format( "/v2/catalogs/%s/updates", catalogId )
	local catalogSpec = WFCatSpecMock( catalogId, kUserId, rxTest )

	local catalog = WFCatMockCatalog.createWFCatalogForTestWithSpec( rxTest, catalogSpec )

	local docStore = catalog:docStore()

	local pusher = WFCatMetadataPusher( catalog, options )
	rxTest:addCloseTarget( pusher )

	local conflictResolver = require 'WFCatConflictResolver'( docStore, catalog:indexStore() )
	rxTest:addCloseTarget( conflictResolver )

	local assetId = KSIdGenerator.uuidInLowercaseHex()
	local assetRevId = KSIdGenerator.uuidInLowercaseHex()

	-- add a doc to the doc store, make sure it sends
	rxTest:waitFor( docStore:setDocValue( {
				_id = assetId,
				_newRevId = assetRevId,
				captureDate = "2015-05-17T15:50:12",
				importSource = {
					importTimestamp = "2015-05-26T17:34:23-07:00",
					originalHeight = 1600,
					originalWidth = 1068,
				},
				location = {
					altitude = 377.9,
					latitude = 45.951733,
					longitude = -93.600107,
				},
				subtype = "image",
				type = "asset",
				version = 1,
				xmp = {
					dc = {
						description = "Rum River State Forest, near Onamia, Minnesota",
						rights = "test copyright string",
					},
				},
			} ) )

	catalogSpec:expectHttpRequest(
		WFMockHttpRequest()
			:put( catalogUpdatesUrl )
			:expectRequestData( ":ignore:" )
			:respond( '200', '{}' ) )

	rxTest:waitFor( pusher:_DEBUG_sendNUpdates( 1 ) )

	-- make it look like Oz got the doc
	rxTest:waitFor( docStore:bulkSetDocValue( {
		label = 'oz',
		{
			_id = assetId,
			_revIds = { assetRevId },
			_serverProvidedMetadata = {
				revision_ids = { assetRevId },
				links = {
					self = 'href'
				}
			},
			captureDate = "2015-05-17T15:50:12",
			importSource = {
				importTimestamp = "2015-05-26T17:34:23-07:00",
				originalHeight = 1600,
				originalWidth = 1068,
			},
			location = {
				altitude = 377.9,
				latitude = 45.951733,
				longitude = -93.600107,
			},
			subtype = "image",
			type = "asset",
			version = 1,
			xmp = {
				dc = {
					description = "Rum River State Forest, near Onamia, Minnesota",
					rights = "test copyright string",
				},
			},
		}
	} ) )

	rxTest:waitFor( pusher:_DEBUG_sendNUpdates( 0 ) )

	-- update doc to the doc store with dont send to oz, make sure it does not
	docStore:bulkApplyOverlay( { label = '_dontsendtooz',
		{
			_id = assetId,
			_deleted = true
		}
	} )

	rxTest:waitFor( pusher:_DEBUG_sendNUpdates( 0 ) )

	local newRevId = KSIdGenerator.uuidInLowercaseHex()

	rxTest:waitFor( docStore:bulkSetDocValue( {
		label = 'oz',
		{
			_id = assetId,
			_revIds = { assetRevId, newRevId },
			_serverProvidedMetadata = {
				revision_ids = { assetRevId, newRevId },
				links = {
					self = 'href'
				}
			},
			captureDate = "2015-05-17T15:50:12",
			importSource = {
				importTimestamp = "2015-05-26T17:34:23-07:00",
				originalHeight = 1600,
				originalWidth = 1068,
			},
			location = {
				altitude = 377.9,
				latitude = 45.951733,
				longitude = -93.600107,
			},
			subtype = "image",
			type = "asset",
			version = 1,
			xmp = {
				dc = {
					description = "Rum River State Forest, near Onamia, Minnesota",
					rights = "test copyright string",
				},
			},
		}
	} ) )

	-- wait while we resolve the conflict here
	local assetConflictResolved = rxTest:observe( docStore:ongoingDocValue( assetId )
		:map( function( docValue )
			return not docValue._conflict
		end ) )
	assetConflictResolved:expectDataWithinTime( 2, true )

	catalogSpec:expectHttpRequest(
		WFMockHttpRequest()
			:put( catalogUpdatesUrl )
			:expectRequestData( ":ignore:" )
			:respond( '200', '{}' ) )

	rxTest:waitFor( pusher:_DEBUG_sendNUpdates( 1 ) )

end

--------------------------------------------------------------------------------
-- Test that a non-local conflicting non-local delete marked "don't send to Oz"
-- will not be sent to Oz but the resolved delete will be sent to Oz will be sent.
-- This case is similar to the previous test but the order of the change and delete
-- are reversed.  This is almost impossible to have happen in current client practice
-- because clients generally don't produce conflicting changes locally as
-- almost all change operations happen via applyOverlay. But if multiple sync endpoints
-- are used or more asynchronous changes are applied not via applyOverlay, this could
-- occur. See WF-311 and linked bugs for more details.
function WFCatMetadataPusherTest.rxTestSendResolvedNonLocalDeleteToOz_mock( rxTest )

	rxTest:setTimeLimit( 5 )

	local catalogId = KSIdGenerator:uuidInLowercaseHex()
	local catalogUpdatesUrl = string.format( "/v2/catalogs/%s/updates", catalogId )
	local catalogSpec = WFCatSpecMock( catalogId, kUserId, rxTest )

	local catalog = WFCatMockCatalog.createWFCatalogForTestWithSpec( rxTest, catalogSpec )

	local docStore = catalog:docStore()

	local pusher = WFCatMetadataPusher( catalog, options )
	rxTest:addCloseTarget( pusher )

	local conflictResolver = require 'WFCatConflictResolver'( docStore, catalog:indexStore() )
	rxTest:addCloseTarget( conflictResolver )

	local assetId2 = KSIdGenerator.uuidInLowercaseHex()
	local assetRevId2 = KSIdGenerator.uuidInLowercaseHex()

	-- add a doc to the doc store, make sure it sends
	rxTest:waitFor( docStore:bulkSetDocValue( { label = 'oz',
			{
				_serverProvidedMetadata = {
					revision_ids = { assetRevId2 },
					links = {
						self = 'href'
					}
				},
				_id = assetId2,
				_newRevId = assetRevId2,
				captureDate = "2015-05-17T15:50:12",
				importSource = {
					importTimestamp = "2015-05-26T17:34:23-07:00",
					originalHeight = 1600,
					originalWidth = 1068,
				},
				location = {
					altitude = 377.9,
					latitude = 45.951733,
					longitude = -93.600107,
				},
				subtype = "image",
				type = "asset",
				version = 1,
				xmp = {
					dc = {
						description = "Rum River State Forest, near Onamia, Minnesota",
						rights = "test copyright string",
					},
				},
			} } ) )

	rxTest:waitFor( pusher:_DEBUG_sendNUpdates( 0 ) )

	local newRevId = KSIdGenerator.uuidInLowercaseHex()

	rxTest:waitFor( docStore:bulkSetDocValue( {
		{
			_serverProvidedMetadata = {
				revision_ids = { assetRevId2 },
				links = {
					self = 'href'
				}
			},
			_id = assetId2,
			_revIds = { assetRevId2, newRevId },
			captureDate = "2015-05-17T15:50:12",
			importSource = {
				importTimestamp = "2015-05-26T17:34:23-07:00",
				originalHeight = 1600,
				originalWidth = 1068,
			},
			location = {
				altitude = 380.9,
				latitude = 45.951733,
				longitude = -93.600107,
			},
			subtype = "image",
			type = "asset",
			version = 1,
			xmp = {
				dc = {
					description = "Rum River State Forest, near Onamia, Minnesota",
					rights = "test copyright string",
				},
			},
		}
	} ) )

	catalogSpec:expectHttpRequest(
		WFMockHttpRequest()
			:put( catalogUpdatesUrl )
			:expectRequestData( ":ignore:" )
			:respond( '200', '{}' ) )

	rxTest:waitFor( pusher:_DEBUG_sendNUpdates( 1 ) )

	newRevId = KSIdGenerator.uuidInLowercaseHex()

	-- make produce a deleted conflict
	rxTest:waitFor( docStore:bulkSetDocValue( {
		label = '_dontsendtooz',
		{
			_serverProvidedMetadata = {
				revision_ids = { assetRevId2 },
				links = {
					self = 'href'
				}
			},
			_id = assetId2,
			_revIds = { assetRevId2, newRevId },
			_deleted = true,
			captureDate = "2015-05-17T15:50:12",
			importSource = {
				importTimestamp = "2015-05-26T17:34:23-07:00",
				originalHeight = 1600,
				originalWidth = 1068,
			},
			location = {
				altitude = 377.9,
				latitude = 45.951733,
				longitude = -93.600107,
			},
			subtype = "image",
			type = "asset",
			version = 1,
			xmp = {
				dc = {
					description = "Rum River State Forest, near Onamia, Minnesota",
					rights = "test copyright string",
				},
			},
		}
	} ) )


	-- wait while we resolve the conflict here
	local assetConflictResolved = rxTest:observe( docStore:ongoingDocValue( assetId2 )
		:map( function( docValue )
			return not docValue._conflict
		end ) )
	assetConflictResolved:expectDataWithinTime( 2, true )

	catalogSpec:expectHttpRequest(
		WFMockHttpRequest()
			:put( catalogUpdatesUrl )
			:expectRequestData( ":ignore:" )
			:respond( '200', '{}' ) )

	rxTest:waitFor( pusher:_DEBUG_sendNUpdates( 1 ) )

end

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

function WFCatMetadataPusherTest.rxTestSendNewDocsOnlyOnce_mock( rxTest )

	-- Send new documents to Oz, read them back, make sure we don't
	-- then send the same (unmodified) documents back to Oz again.

	rxTest:setTimeLimit( 5 )

	local catalogId = KSIdGenerator:uuidInLowercaseHex()

	local docStoreMocks = generateDocStoreMocks( 'catalog', catalogId )

	local reqMocks = {

		pusherMockReqResp( 'catalog', catalogId, docStoreMocks.pusherUpdates ),
		prefetchReqResp( catalogId ),
		catalogPrefetchReqResp( catalogId ),
		pullerMockReqResp( 'catalog', catalogId, docStoreMocks.pullerUpdates ),
		pullerMockReqResp( 'catalog', catalogId, {} ),

	}

	local catalogSpec = WFCatSpecMock( catalogId, kUserId, rxTest, reqMocks )
	local catalog = WFCatMockCatalog.createWFCatalogForTestWithSpec( rxTest, catalogSpec )

	local docStore = catalog:docStore()

	local pusher = WFCatMetadataPusher( catalog, options )
	rxTest:addCloseTarget( pusher )

	local updateCount = rxTest:observe( pusher:countUpdatesToSend() )
	updateCount:expectData( 0 )

	local puller = WFCatMetadataPuller( catalog, "Test Puller", options )
	rxTest:addCloseTarget( puller )

	-- Create one document of each type currently supported on Oz.

	rxTest:waitFor( docStore:bulkSetDocValue( docStoreMocks.docStoreValues ) )

	updateCount:expectDataWithinTime( 1, 3 )

	-- Send these docs to Oz.

	rxTest:waitFor( pusher:_DEBUG_sendNUpdates( 3 ) )

	updateCount:expectDataWithinTime( 1, 0 )

	-- Now pull the changes back from Oz.

	rxTest:expectOnlyCompleted( puller:getUpdatesOnce() )

	rxTest:expectSingleValue(
		docStore:changedDocs { fromChangeCounter = 0 }
			:map( remapChangedDocsFeedForMockTests ),
		docStoreMocks.expectedDocStoreValues
	)

	-- Should be no updates to send.

	rxTest:waitFor( pusher:_DEBUG_sendNUpdates( 0 ) )

	updateCount:expectDataWithinTime( 1, 0 )

	-- And also, no updates to receive.

	rxTest:expectOnlyCompleted( puller:getUpdatesOnce() )

end

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

function WFCatMetadataPusherTest.rxTestSendNewRevisionNoDelta_testSerially( rxTest )

	-- WF-1757; If the pusher detects the local revision is the same as the revision on Oz,
	-- then update the local revision Id and label so that we know we do not need to sync it.

	rxTest:setTimeLimit( 5 )

	local catalogId = KSIdGenerator:uuidInLowercaseHex()

	local docStoreMocks = generateDocStoreMocks( 'catalog', catalogId, { excludeAlbum = true } )
	local assetDocId = docStoreMocks.assetDocId
	local ozRevision = docStoreMocks.assetRevisionId

	local reqMocks = {

		pusherMockReqResp( 'catalog', catalogId, docStoreMocks.pusherUpdates ),
		prefetchReqResp( catalogId ),
		catalogPrefetchReqResp( catalogId ),
		pullerMockReqResp( 'catalog', catalogId, docStoreMocks.pullerUpdates ),
		pullerMockReqResp( 'catalog', catalogId, {} ),

	}

	local catalogSpec = WFCatSpecMock( catalogId, kUserId, rxTest, reqMocks )
	local catalog = WFCatMockCatalog.createWFCatalogForTestWithSpec( rxTest, catalogSpec )

	local docStore = catalog:docStore()

	local pusher = WFCatMetadataPusher( catalog, options )
	rxTest:addCloseTarget( pusher )

	local updateCount = rxTest:observe( pusher:countUpdatesToSend() )
	updateCount:expectData( 0 )

	local puller = WFCatMetadataPuller( catalog, "Test Puller", options )
	rxTest:addCloseTarget( puller )

	-- Create an asset document and sync it with Oz.

	rxTest:waitFor( docStore:bulkSetDocValue( docStoreMocks.docStoreValues ) )

	updateCount:expectDataWithinTime( 1, 1 )

	-- Send the document to Oz.

	rxTest:waitFor( pusher:_DEBUG_sendNUpdates( 1 ) )

	updateCount:expectDataWithinTime( 1, 0 )

	-- Now pull the changes back from Oz.

	rxTest:expectOnlyCompleted( puller:getUpdatesOnce() )

	-- Make sure there's no additional updates to send.

	rxTest:waitFor( pusher:_DEBUG_sendNUpdates( 0 ) )

	updateCount:expectDataWithinTime( 1, 0 )

	-- And also, no updates to receive.

	rxTest:expectOnlyCompleted( puller:getUpdatesOnce() )

	-- Now create a new asset revision with a new capture date.

	local newRevisionWithDelta = rxTest:getFirstValue( docStore:applyOverlay( {
		_id = assetDocId,
		captureDate = "2015-06-17T15:50:12",
	} ) )

	-- Now revert the capture date change, creating another new revision
	-- that is identical to the initial revision that is on Oz.

	local newRevisionNoDelta = rxTest:getFirstValue( docStore:applyOverlay( {
		_id = assetDocId,
		captureDate = "2015-05-17T15:50:12",
	} ) )

	-- Verify the asset now has 3 revisions.

	rxTest:expectSingleValue(

		docStore:docRevisionHistory( assetDocId ),
		{
			{
				revId = newRevisionNoDelta,
				status = 'available',
			},
			{
				revId = newRevisionWithDelta,
				status = 'available',
			},
			{
				revId = ozRevision,
				status = 'available',
			},
		}
	)

	local function stripFussyBits( docValue )

		docValue._changeCounter = nil
		docValue._labels = nil
		docValue._revId = nil
		return docValue

	end

	-- Fetch the doc values for each specific revision

	local ozDocValue = rxTest:getFirstValue( docStore:docValue( assetDocId, ozRevision ):map( stripFussyBits ) )
	local updatedDocValue = rxTest:getFirstValue( docStore:docValue( assetDocId, newRevisionWithDelta ):map( stripFussyBits ) )
	local revertedDocValue = rxTest:getFirstValue( docStore:docValue( assetDocId, newRevisionNoDelta ):map( stripFussyBits ) )

	-- The first revision and the second revision should not match.

	rxTest:expectNotEqual( ozDocValue, updatedDocValue )

	-- The second and third revision also should not match.

	rxTest:expectNotEqual( updatedDocValue, revertedDocValue )

	-- But the first and third revisions should be identical.

	rxTest:expectEqual( ozDocValue, revertedDocValue )

	-- Confirm the pusher's change counter (sequence number).

	rxTest:expectSingleValue( catalog:docStore():localDocValue( 'Oz-sendUpdatesContinueFrom' )
			:pluck( 'continueFrom' ), 4 )

	-- Now activate the pusher. The pusher should not send any updates to Oz, but should
	-- create a new document revision containing a _dontsendtooz label.

	rxTest:waitFor( pusher:_DEBUG_sendNUpdates( 0 ) )

	-- Get the new revision Id:

	local newNewRevisionIdWithLabel = rxTest:getFirstValue( docStore:docValue( assetDocId )
			:pluck( '_revId' ) )

	-- Verify the new revision and the label

	rxTest:expectSingleValue(

		docStore:docRevisionHistory( assetDocId ),
		{
			{
				revId = newNewRevisionIdWithLabel,
				status = 'available',
			},
			{
				revId = newRevisionNoDelta,
				status = 'available',
			},
			{
				revId = newRevisionWithDelta,
				status = 'available',
			},
			{
				revId = ozRevision,
				status = 'available',
			},
		}
	)

	local newDocValue = rxTest:getFirstValue( docStore:docValue( assetDocId )
		:tap( function( docValue )
			rxTest:expectEqual( true, docValue._labels[ '_dontsendtooz' ] )
		end )
		:map( stripFussyBits ) )

	rxTest:expectEqual( ozDocValue, newDocValue )

	-- Confirm the pusher's change counter (sequence number) has increased.

	rxTest:expectSingleValue( catalog:docStore():localDocValue( 'Oz-sendUpdatesContinueFrom' )
			:pluck( 'continueFrom' ), 6 )

	-- Now activate the pusher one last time. The pusher should process the updated document once
	-- more, which is a no-op.

	rxTest:waitFor( pusher:_DEBUG_sendNUpdates( 0 ) )

	rxTest:expectSingleValue( catalog:docStore():localDocValue( 'Oz-sendUpdatesContinueFrom' )
			:pluck( 'continueFrom' ), 7 )

end

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

function WFCatMetadataPusherTest.rxTestSendNewRevisionNoDelta_WF_1853_testNightly( rxTest )

	--[[

		With the initial fix for WF-1757, there was a race condition that caused
		bug WF-1853:

		- User has a document at revision A, which is synced to Oz
		- User updates the document to revision B
		- Revision B is sent to Oz
		- User quickly creates a new Revision C that is identical to Revision A
			before the Revision B update response is synced down from Oz.

		The handling for WF-1757 would cause us to lose Revision C in the above scenario.

		The following test verifies we no longer lose Revision C.

	]]--

	rxTest:setTimeLimit( 120 )

	local catalogTestSession = WFCatTestSession( rxTest )

	local catalog = rxTest:getFirstValue( catalogTestSession:catalog() )

	local docStore = catalog:docStore()
	local indexStore = catalog:indexStore()

	local puller = WFCatMetadataPuller( catalog, "Test Puller", options )
	rxTest:addCloseTarget( puller )

	rxTest:expectOnlyCompleted( puller:getUpdatesOnce() )

	puller:getUpdatesContinuously()

	local pusher = WFCatMetadataPusher( catalog, options )
	rxTest:addCloseTarget( pusher )

	local catalogSpec = rxTest:getFirstValue( catalogTestSession:catalogSpec() )

	local assetId = KSIdGenerator.uuidInLowercaseHex()

	rxTest:waitFor(
		catalog:docStore():setDocValue {
			_id = assetId,
			_localOnly = {
				mumble = "stuff",
			},
			captureDate = "2004-11-02T11:10:09",
			importSource = {
				fileName = "fakeFile01.jpg",
				importedBy = catalogSpec:ozUserId(),
				importTimestamp = "2015-11-02T08:37:24.513Z",
				originalHeight = 16,
				originalWidth = 16,
			},
			subtype = "image",
			type = "asset",
		} )

	-- Send the document to Oz.

	local sendNUpdates = rxTest:observe( pusher:_DEBUG_sendNUpdates( 1 ) )
	sendNUpdates:expectCompletedWithinTime( 30 )
	sendNUpdates:dispose()

	-- Wait for the document response from Oz.

	local observeAssetOnOz = observeHasServerProvidedMetadata( rxTest, catalog, assetId )
	observeAssetOnOz:expectDataWithinTime( 40, true )
	observeAssetOnOz:dispose()

	rxTest:observe( docStore:ongoingDocValue( assetId ):pluck( '_labels' ) )
		:expectDataWithinTime( 30, { oz = true } )
		:dispose()

	-- Now create a new asset revision with a new capture date and send it to Oz.

	-- DATE UPDATE 1

	rxTest:waitFor( docStore:applyOverlay( {
		_id = assetId,
		captureDate = "2015-06-17T15:50:12",
	} ) )

	sendNUpdates = rxTest:observe( pusher:_DEBUG_sendNUpdates( 1 ) )
	sendNUpdates:expectCompletedWithinTime( 30 )
	sendNUpdates:dispose()

	-- Now quickly revert the change and try to send it:

	-- DATE UPDATE 2 (AKA Revision C)

	rxTest:waitFor( docStore:applyOverlay( {
		_id = assetId,
		captureDate = "2004-11-02T11:10:09",
	} ) )

	sendNUpdates = rxTest:observe( pusher:_DEBUG_sendNUpdates( 0 ) )
	sendNUpdates:expectCompletedWithinTime( 30 )
	sendNUpdates:dispose()

	-- Confirm the pusher's change counter (sequence number).

	rxTest:expectSingleValue( catalog:docStore():localDocValue( 'Oz-sendUpdatesContinueFrom' )
			:pluck( 'continueFrom' ), 6 )

	-- DATE UPDATE 1 should sync down from Oz, causing a conflict. This is because
	-- we haven't send DATE UPDATE 2 to Oz yet.

	rxTest:observe( docStore:ongoingDocValue( assetId ):pluck( '_conflict' ) )
		:expectDataWithinTime( 30, true )
		:dispose()

	-- Now wait for the conflict to be resolved

	local conflictResolver = WFCatConflictResolver( docStore, indexStore )
	rxTest:addCloseTarget( conflictResolver )

	rxTest:observe( docStore:ongoingDocValue( assetId ):pluck( '_conflict' ) )
		:expectDataWithinTime( 30, nil )
		:dispose()

	-- Confirm that a new revision was created. This revision should have no labels,
	-- the updated DATE UPDATE 2 capture date

	rxTest:expectSingleValue(
			catalog:docStore():docValue( assetId )
				:map( function( docValue )

					return not not docValue._labels, docValue.captureDate

				end ), false, "2004-11-02T11:10:09" )

	-- Now this should be sent to Oz:

	sendNUpdates = rxTest:observe( pusher:_DEBUG_sendNUpdates( 1 ) )
	sendNUpdates:expectCompletedWithinTime( 30 )
	sendNUpdates:dispose()

	-- Eventually the update should sync back down with the oz label:

	rxTest:observe( docStore:ongoingDocValue( assetId ):pluck( '_labels' ) )
		:expectDataWithinTime( 30, { oz = true } )
		:dispose()

end

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

function WFCatMetadataPusherTest.rxTestSendNewRevisionNoDelta_WF_1853_2_testNightly( rxTest )

	--[[

		See WFCatMetadataPusherTest.rxTestSendNewRevisionNoDelta_WF_1853_testNightly
		for additional information on the purpose of this test.

		This test verifies that we don't lose any updates in the following scenario:

		- User has a document at revision A, which is synced to Oz
		- User updates the document to revision B
		- Revision B is sent to Oz
		- User quickly creates a new Revision C that is identical to Revision A
			before the Revision B update response is synced down from Oz.
		- User quickly creates a new Revision D that is identical to Revision B
			before the Revision B update response is synced down from Oz.

	]]--

	rxTest:setTimeLimit( 120 )

	local catalogTestSession = WFCatTestSession( rxTest )

	local catalog = rxTest:getFirstValue( catalogTestSession:catalog() )

	local docStore = catalog:docStore()
	local indexStore = catalog:indexStore()

	local puller = WFCatMetadataPuller( catalog, "Test Puller", options )
	rxTest:addCloseTarget( puller )

	rxTest:expectOnlyCompleted( puller:getUpdatesOnce() )

	puller:getUpdatesContinuously()

	local pusher = WFCatMetadataPusher( catalog, options )
	rxTest:addCloseTarget( pusher )

	local catalogSpec = rxTest:getFirstValue( catalogTestSession:catalogSpec() )

	local assetId = KSIdGenerator.uuidInLowercaseHex()

	rxTest:waitFor(
		catalog:docStore():setDocValue {
			_id = assetId,
			_localOnly = {
				mumble = "stuff",
			},
			captureDate = "2004-11-02T11:10:09",
			importSource = {
				fileName = "fakeFile01.jpg",
				importedBy = catalogSpec:ozUserId(),
				importTimestamp = "2015-11-02T08:37:24.513Z",
				originalHeight = 16,
				originalWidth = 16,
			},
			subtype = "image",
			type = "asset",
		} )

	-- Send the document to Oz.

	local sendNUpdates = rxTest:observe( pusher:_DEBUG_sendNUpdates( 1 ) )
	sendNUpdates:expectCompletedWithinTime( 30 )
	sendNUpdates:dispose()

	-- Wait for the document response from Oz.

	local observeAssetOnOz = observeHasServerProvidedMetadata( rxTest, catalog, assetId )
	observeAssetOnOz:expectDataWithinTime( 40, true )
	observeAssetOnOz:dispose()

	rxTest:observe( docStore:ongoingDocValue( assetId ):pluck( '_labels' ) )
		:expectDataWithinTime( 30, { oz = true } )
		:dispose()

	-- Now create a new asset revision with a new capture date and send it to Oz.

	-- DATE UPDATE 1

	rxTest:waitFor( docStore:applyOverlay( {
		_id = assetId,
		captureDate = "2015-06-17T15:50:12",
	} ) )

	sendNUpdates = rxTest:observe( pusher:_DEBUG_sendNUpdates( 1 ) )
	sendNUpdates:expectCompletedWithinTime( 30 )
	sendNUpdates:dispose()

	-- Now quickly revert the change and try to send it:

	-- DATE UPDATE 2 (Revision C)

	rxTest:waitFor( docStore:applyOverlay( {
		_id = assetId,
		captureDate = "2004-11-02T11:10:09",
	} ) )

	sendNUpdates = rxTest:observe( pusher:_DEBUG_sendNUpdates( 0 ) )
	sendNUpdates:expectCompletedWithinTime( 30 )
	sendNUpdates:dispose()

	-- Now quickly revert the revert and try to send it:

	-- DATE UPDATE 3 (Revision D)

	rxTest:waitFor( docStore:applyOverlay( {
		_id = assetId,
		captureDate = "2015-06-17T15:50:12",
	} ) )

	sendNUpdates = rxTest:observe( pusher:_DEBUG_sendNUpdates( 1 ) )
	sendNUpdates:expectCompletedWithinTime( 30 )
	sendNUpdates:dispose()

	-- Confirm the pusher's change counter (sequence number).

	rxTest:expectSingleValue( catalog:docStore():localDocValue( 'Oz-sendUpdatesContinueFrom' )
			:pluck( 'continueFrom' ), 8 )

	-- DATE UPDATE 1 should sync down from Oz, causing a conflict with Revision D

	rxTest:observe( docStore:ongoingDocValue( assetId ):pluck( '_conflict' ) )
		:expectDataWithinTime( 30, true )
		:dispose()

	-- Now wait for the conflict to be resolved

	local conflictResolver = WFCatConflictResolver( docStore, indexStore )
	rxTest:addCloseTarget( conflictResolver )

	rxTest:observe( docStore:ongoingDocValue( assetId ):pluck( '_conflict' ) )
		:expectDataWithinTime( 30, nil )
		:dispose()

	-- Confirm that the conflict resolver chooses the revision from Oz
	-- since the local revision is identical.

	rxTest:expectSingleValue(
			catalog:docStore():docValue( assetId )
				:map( function( docValue )

					return docValue._labels, docValue.captureDate

				end ), { oz = true }, "2015-06-17T15:50:12" )

	-- And there should be nothing to send to Oz.

	sendNUpdates = rxTest:observe( pusher:_DEBUG_sendNUpdates( 0 ) )
	sendNUpdates:expectCompletedWithinTime( 30 )
	sendNUpdates:dispose()

end

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

function WFCatMetadataPusherTest.rxTestSendUpdatesWith429AndRetryDelay_mock( rxTest )

	-- Test to see if WFCatMetadataPuller handles error 429 properly and
	-- adds the requested Retry-After when added to successful response
	-- headers.

	rxTest:setTimeLimit( 5 )

	local kRetryDelay = 1 -- Amount of time used in Retry-After calls.
	local kNoDelay = .5 -- An amount of time any of the requests can easily be completed in if there are no Retry-Afters requested.

	local catalogId = KSIdGenerator:uuidInLowercaseHex()

	local docStoreMocks = generateDocStoreMocks( 'catalog', catalogId )

	local albumAssetPusherUpdate1 = docStoreMocks.pusherUpdates[2]
	local albumAssetDocId = albumAssetPusherUpdate1.id

	local newAlbumAssetRevisionId = KSIdGenerator.uuidInLowercaseHex()
	local albumAssetPusherUpdate2 = KSTableUtils.deepCopy( albumAssetPusherUpdate1 )
	albumAssetPusherUpdate2.revision_id = newAlbumAssetRevisionId
	albumAssetPusherUpdate2.payload.order = "v"

	local fail429 = {
		{},
		{
			status_code = '429',
			['Retry-After' ] = kRetryDelay, -- no delay expected, WFRxHttpClient would impose it
		},
	}

	local retryAfterHeader = {
		headers = {
			[ 'Retry-After'] = kRetryDelay
		}
	}

	local reqMocks = {

		pusherMockReqResp( 'catalog', catalogId, docStoreMocks.pusherUpdates ),
			-- initial create of asset, album, album_asset

		pusherMockReqResp( 'catalog', catalogId, { albumAssetPusherUpdate2 }, fail429 ),
			-- 429 error on attempted update of album_asset with Retry-After delay will be ignored becamse MFRxHttpClient would impose it

		KSTableUtils.merge( {}, pusherMockReqResp( 'catalog', catalogId, { albumAssetPusherUpdate2 } ), retryAfterHeader )
			-- successful update of album_asset and Retry-After which is applied

	}

	local albumPusherUpdate1 = docStoreMocks.pusherUpdates[3]
	local albumDocId = albumPusherUpdate1.id

	local newAlbumRevisionIds = {
		KSIdGenerator.uuidInLowercaseHex(),
		KSIdGenerator.uuidInLowercaseHex(),
		KSIdGenerator.uuidInLowercaseHex(),
		KSIdGenerator.uuidInLowercaseHex()
	}

	local newAlbumNames = {
		KSIdGenerator.uuidInLowercaseHex(),
		KSIdGenerator.uuidInLowercaseHex(),
		KSIdGenerator.uuidInLowercaseHex(),
		KSIdGenerator.uuidInLowercaseHex()
	}

	-- include 4 updates to album name, first and third with Retry-After delay
	for i, revisionId in ipairs(newAlbumRevisionIds) do

		local mock = KSTableUtils.deepCopy( albumPusherUpdate1 )
		mock.revision_id = revisionId
		mock.payload.name = newAlbumNames[i]

		if i % 2 == 1 then
			table.insert( reqMocks, KSTableUtils.merge( {}, pusherMockReqResp( 'catalog', catalogId, { mock } ), retryAfterHeader ) )
		else
			table.insert( reqMocks, pusherMockReqResp( 'catalog', catalogId, { mock } ) )
		end

	end

	local catalogSpec = WFCatSpecMock( catalogId, kUserId, rxTest, reqMocks )
	local catalog = WFCatMockCatalog.createWFCatalogForTestWithSpec( rxTest, catalogSpec )

	local docStore = catalog:docStore()

	local pusherOptions = KSTableUtils.deepCopy( options )
	pusherOptions.delayBeforeSendingUpdates = 0.01

	local pusher = WFCatMetadataPusher( catalog, pusherOptions )
	rxTest:addCloseTarget( pusher )

	local updateCount = rxTest:observe( pusher:countUpdatesToSend() )
	updateCount:expectData( 0 )

	-- Create one document of each type currently supported on Oz.

	rxTest:waitFor( docStore:bulkSetDocValue( docStoreMocks.docStoreValues ) )

	updateCount:expectDataWithinTime( 1, 3 )

	-- Send these docs to Oz.

	local sendNUpdates = rxTest:observe( pusher:_DEBUG_sendNUpdates( 3 ) )
	sendNUpdates:expectCompletedWithinTime( kNoDelay )
	sendNUpdates:dispose()

	updateCount:expectDataWithinTime( 1, 0 )

	rxTest:expectEqual( 1, catalogSpec._httpMock:requestCount() )

	-- Should be no updates to send, and no delay

	sendNUpdates = rxTest:observe( pusher:_DEBUG_sendNUpdates( 0 ) )
	sendNUpdates:expectCompletedWithinTime( kNoDelay )
	sendNUpdates:dispose()

	rxTest:expectEqual( 1, catalogSpec._httpMock:requestCount() )

	-- Update album_asset with order informations

	rxTest:waitFor( catalog:docStore():applyOverlay {
		_id = albumAssetDocId,
		_newRevId = newAlbumAssetRevisionId,
		order = 'v',
	} )

	updateCount:expectDataWithinTime( 1, 1 )

	-- Expect a 429 error in this case so no update should occur.
	-- Delay would be imposed by WFRxHttpClient in this case
	-- but since we are using a mocked http client expect no delay.

	rxTest:expectOnlyError( pusher:_DEBUG_sendUpdatesOnce(), {}, {
			["Retry-After"] = 1,
			status_code = "429",
		}
	)
	updateCount:expectDataWithinTime( 1, 1 )
	rxTest:expectEqual( 2, catalogSpec._httpMock:requestCount() )

	-- Try again, should succeed quickly, as above Retry-After would
	-- have been imposed on the above call by WFRxHttpClient.
	-- This response requests a delay on the following call.

	sendNUpdates = rxTest:observe( pusher:_DEBUG_sendNUpdates( 1 ) )
	sendNUpdates:expectCompletedWithinTime( kNoDelay )
	sendNUpdates:dispose()

	updateCount:expectDataWithinTime( 1, 0 )
	rxTest:expectEqual( 3, catalogSpec._httpMock:requestCount() )

	-- Rename album first time and upload to Oz, this time
	-- the response will indicate we should wait before
	-- trying to upload again on the following call.

	rxTest:waitFor( catalog:docStore():applyOverlay {
		_id = albumDocId,
		_newRevId = newAlbumRevisionIds[1],
		name = newAlbumNames[1],
	} )

	updateCount:expectDataWithinTime( 1, 1 )

	sendNUpdates = rxTest:observe( pusher:_DEBUG_sendNUpdates( 1 ) )
	sendNUpdates:expectCompletedWithinTime( 30 )
	sendNUpdates:dispose()

	-- retry delay applied from previous album_asset success response
	rxTest:expectEqual( true, catalogSpec._httpMock:timeBetweenRequests() >= kRetryDelay )

	updateCount:expectDataWithinTime( 1, 0 )
	rxTest:expectEqual( 4, catalogSpec._httpMock:requestCount() )

	-- Rename album a second time and upload to Oz, no Retry-After delay in this response
	-- but this request should be delayed due to Retry-After in previous response

	rxTest:waitFor( catalog:docStore():applyOverlay {
		_id = albumDocId,
		_newRevId = newAlbumRevisionIds[2],
		name = newAlbumNames[2],
	} )

	updateCount:expectDataWithinTime( 1, 1 )

	sendNUpdates = rxTest:observe( pusher:_DEBUG_sendNUpdates( 1 ) )
	sendNUpdates:expectCompletedWithinTime( 30 )
	sendNUpdates:dispose()

	-- retry delay applied from first album success response
	rxTest:expectEqual( true, catalogSpec._httpMock:timeBetweenRequests() >= kRetryDelay )

	updateCount:expectDataWithinTime( 1, 0 )
	rxTest:expectEqual( 5, catalogSpec._httpMock:requestCount() )

	-- Nothing to send and no retry delay.

	sendNUpdates = rxTest:observe( pusher:_DEBUG_sendNUpdates( 0 ) )
	sendNUpdates:expectCompletedWithinTime( kNoDelay )
	sendNUpdates:dispose()

	rxTest:expectEqual( 5, catalogSpec._httpMock:requestCount() )

	-- Rename album third time and upload to Oz, this time
	-- the response will indicate we should wait before
	-- trying to upload again on the following call.

	rxTest:waitFor( catalog:docStore():applyOverlay {
		_id = albumDocId,
		_newRevId = newAlbumRevisionIds[3],
		name = newAlbumNames[3],
	} )

	updateCount:expectDataWithinTime( 1, 1 )

	-- no delay this time since response from second album update had no Retry-After delay

	sendNUpdates = rxTest:observe( pusher:_DEBUG_sendNUpdates( 1 ) )
	sendNUpdates:expectCompletedWithinTime( kNoDelay )
	sendNUpdates:dispose()

	updateCount:expectDataWithinTime( 1, 0 )
	rxTest:expectEqual( 6, catalogSpec._httpMock:requestCount() )

	-- Rename album a fourth time and upload to Oz, no Retry-After delay in this response
	-- but this request should be delayed due to Retry-After in previous response

	rxTest:waitFor( catalog:docStore():applyOverlay {
		_id = albumDocId,
		_newRevId = newAlbumRevisionIds[4],
		name = newAlbumNames[4],
	} )

	updateCount:expectDataWithinTime( 1, 1 )

	sendNUpdates = rxTest:observe( pusher:_DEBUG_sendNUpdates( 1 ) )
	sendNUpdates:expectCompletedWithinTime( 30 )
	sendNUpdates:dispose()

	-- retry delay applied from third album success response
	rxTest:expectEqual( true, catalogSpec._httpMock:timeBetweenRequests() >= kRetryDelay )

	updateCount:expectDataWithinTime( 1, 0 )
	rxTest:expectEqual( 7, catalogSpec._httpMock:requestCount() )

	-- Nothing to send and no retry delay.

	sendNUpdates = rxTest:observe( pusher:_DEBUG_sendNUpdates( 0 ) )
	sendNUpdates:expectCompletedWithinTime( kNoDelay )
	sendNUpdates:dispose()

	rxTest:expectEqual( 7, catalogSpec._httpMock:requestCount() )

end

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

function WFCatMetadataPusherTest.rxTestLocalChangeBeforeReadbackFromOz_mock( rxTest )

	rxTest:setTimeLimit( 10 )
	
	-- Send new documents to Oz, change locally after that, simulate readback
	-- first change from Oz (after send).

	local catalogId = KSIdGenerator:uuidInLowercaseHex()

	local docStoreMocks = generateDocStoreMocks( 'catalog', catalogId, { includeAssetUpdate = true } )

	local assetDocId = docStoreMocks.assetDocId

	local reqMocks = {

		pusherMockReqResp( 'catalog', catalogId, docStoreMocks.pusherUpdates ),
		prefetchReqResp( catalogId ),
		catalogPrefetchReqResp( catalogId ),
		pullerMockReqResp( 'catalog', catalogId, docStoreMocks.pullerUpdates ),
		pullerMockReqResp( 'catalog', catalogId, {} ),
		pusherMockReqResp( 'catalog', catalogId, { docStoreMocks.assetUpdateMock } ),

	}

	local catalogSpec = WFCatSpecMock( catalogId, kUserId, rxTest, reqMocks )
	local catalog = WFCatMockCatalog.createWFCatalogForTestWithSpec( rxTest, catalogSpec )

	local docStore = catalog:docStore()

	local pusher = WFCatMetadataPusher( catalog, options )
	rxTest:addCloseTarget( pusher )

	local updateCount = rxTest:observe( pusher:countUpdatesToSend() )
	updateCount:expectData( 0 )

	local puller = WFCatMetadataPuller( catalog, "Test Puller", options )
	rxTest:addCloseTarget( puller )

	-- Create one document of each type currently supported on Oz.

	rxTest:waitFor( docStore:bulkSetDocValue( docStoreMocks.docStoreValues ) )

	-- Send these docs to Oz.

	updateCount:expectDataWithinTime( 1, 3 )

	rxTest:waitFor( pusher:_DEBUG_sendNUpdates( 3 ) )

	updateCount:expectDataWithinTime( 1, 0 )

	-- Make a local change to the asset.

	rxTest:waitFor( docStore:applyOverlay( docStoreMocks.assetUpdateOverlay ) )

	updateCount:expectDataWithinTime( 1, 1 )

	-- Now pull changes from Oz. This payload will contain a conflicting version
	-- for this asset document. (It might not *seem* like a conflict, but better
	-- to flag it this way and resolve later.)

	rxTest:expectOnlyCompleted( puller:getUpdatesOnce() )

	docStoreMocks.assetExpectedValue._conflict = true
	docStoreMocks.expectedDocStoreValues.continueFrom = 9
	docStoreMocks.expectedDocStoreValues[3] = docStoreMocks.assetExpectedValue

	rxTest:expectSingleValue(
		docStore:changedDocs { fromChangeCounter = 0 }
			:map( remapChangedDocsFeedForMockTests ),
		docStoreMocks.expectedDocStoreValues
	)

	-- Verify that asset document has two head revisions

	local localRevisionExpectedValue = docStoreMocks.revisionExpectedValue
	localRevisionExpectedValue._changeCounter = 4
	localRevisionExpectedValue._conflict = true

	local ozRevisionExpectedValue = KSTableUtils.deepCopy( localRevisionExpectedValue )
	ozRevisionExpectedValue._changeCounter = 6
	ozRevisionExpectedValue._revId = docStoreMocks.assetRevisionId
	ozRevisionExpectedValue.xmp.dc.description = "Rum River State Forest, near Onamia, Minnesota"

	rxTest:expectSingleValue( docStore:currentRevisionsForDoc( assetDocId ), { localRevisionExpectedValue, ozRevisionExpectedValue } )

	-- Should be no updates to send.

	rxTest:waitFor( pusher:_DEBUG_sendNUpdates( 0 ) )

	-- And also, no updates to receive.

	rxTest:expectOnlyCompleted( puller:getUpdatesOnce() )

	-- At this point pusher gets a little confused and its count considers
	-- the updates synced from Oz in its count of updates to send to Oz

	updateCount:expectDataWithinTime( 1, 3 )

	-- Now resolve the conflict by deleting the version from Oz.

	rxTest:waitFor(
		docStore:deleteDoc( assetDocId, docStoreMocks.assetRevisionId )
	)

	-- Should now send the locally-originated version of the asset document

	updateCount:expectDataWithinTime( 1, 4 )

	rxTest:waitFor( pusher:_DEBUG_sendNUpdates( 1 ) )

	updateCount:expectDataWithinTime( 1, 0 )

end

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

function WFCatMetadataPusherTest.rxTestLocalChangeAfterReadbackFromOz_mock( rxTest )

	-- Send new documents to Oz, read back from Oz, then change locally;
	-- make sure we send correct delta.

	rxTest:setTimeLimit( 5 )

	local catalogId = KSIdGenerator:uuidInLowercaseHex()

	local docStoreMocks = generateDocStoreMocks( 'catalog', catalogId, { includeAssetUpdate = true }  )

	local assetDocId = docStoreMocks.assetDocId

	local reqMocks = {

		pusherMockReqResp( 'catalog', catalogId, docStoreMocks.pusherUpdates ),
		prefetchReqResp( catalogId ),
		catalogPrefetchReqResp( catalogId ),
		pullerMockReqResp( 'catalog', catalogId, docStoreMocks.pullerUpdates ),
		pusherMockReqResp( 'catalog', catalogId, { docStoreMocks.assetUpdateMock } ),

	}

	local catalogSpec = WFCatSpecMock( catalogId, kUserId, rxTest, reqMocks )
	local catalog = WFCatMockCatalog.createWFCatalogForTestWithSpec( rxTest, catalogSpec )

	local docStore = catalog:docStore()

	local pusher = WFCatMetadataPusher( catalog, options )
	rxTest:addCloseTarget( pusher )

	local updateCount = rxTest:observe( pusher:countUpdatesToSend() )
	updateCount:expectData( 0 )

	local puller = WFCatMetadataPuller( catalog, "Test Puller", options )
	rxTest:addCloseTarget( puller )

	-- Create one document of each type currently supported on Oz.

	rxTest:waitFor( docStore:bulkSetDocValue( docStoreMocks.docStoreValues ) )

	-- Send these docs to Oz.

	updateCount:expectDataWithinTime( 1, 3 )

	rxTest:waitFor( pusher:_DEBUG_sendNUpdates( 3 ) )

	updateCount:expectDataWithinTime( 1, 0 )

	-- Now pull changes from Oz. There should be a single winning revision
	-- containing whatever we've received from Oz.

	rxTest:expectOnlyCompleted( puller:getUpdatesOnce() )

	rxTest:expectSingleValue(
		docStore:changedDocs { fromChangeCounter = 0 }
			:map( remapChangedDocsFeedForMockTests ),
		docStoreMocks.expectedDocStoreValues
	)

	-- Make a local change to the asset.

	rxTest:waitFor( docStore:applyOverlay( docStoreMocks.assetUpdateOverlay ) )

	-- Verify that asset document has one head revision.

	rxTest:expectSingleValue( docStore:currentRevisionsForDoc( assetDocId ), { docStoreMocks.revisionExpectedValue } )

	-- Should now send the locally-updated version of the asset document.

	updateCount:expectDataWithinTime( 1, 4 )

	rxTest:waitFor( pusher:_DEBUG_sendNUpdates( 1 ) )

	updateCount:expectDataWithinTime( 1, 0 )

end

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

function WFCatMetadataPusherTest.rxTestSendUpdatesContinuously_mock( rxTest )

	-- Send new documents to Oz, read back from Oz, then change locally;
	-- make sure we send correct delta.

	rxTest:setTimeLimit( 8 )

	local catalogId = KSIdGenerator:uuidInLowercaseHex()

	local docStoreMocks = generateDocStoreMocks( 'catalog', catalogId, { includeAssetUpdate = true }  )

	local assetDocId = docStoreMocks.assetDocId
	local albumDocId = docStoreMocks.albumDocId
	local albumAssetDocId = docStoreMocks.albumAssetDocId

	local reqMocks = {

		pusherMockReqResp( 'catalog', catalogId, docStoreMocks.pusherUpdates ),
		prefetchReqResp( catalogId ),
		catalogPrefetchReqResp( catalogId ),
		pullerMockReqResp( 'catalog', catalogId, docStoreMocks.pullerUpdates ),
		pusherMockReqResp( 'catalog', catalogId, { docStoreMocks.assetUpdateMock } ),

	}

	local indexSpecScript

	if WFPlatformInfo.isMobile() then
		indexSpecScript = 'LrmCatalogIndexDefinitions.lua'
	end

	local catalogSpec = WFCatSpecMock( catalogId, kUserId, rxTest, reqMocks )
	local catalog = WFCatMockCatalog.createWFCatalogForTestWithSpec( rxTest, catalogSpec, nil, indexSpecScript )

	local docStore = catalog:docStore()

	local pusher = WFCatMetadataPusher( catalog, options )
	rxTest:addCloseTarget( pusher )

	local syncDiagnosticsObserver = rxTest:observe( pusher:syncDiagnostics( 0.1, 10 )
		:map( function( diagnostics )
			diagnostics.numberOfDocsRemainingToSync = nil
			diagnostics.docsFilteredDueToErrorByType = nil
			return diagnostics
		end ) )

	rxTest:waitFor( pusher:sendUpdatesContinuously() )

	local puller = WFCatMetadataPuller( catalog, "Test Puller", options )
	rxTest:addCloseTarget( puller )

	syncDiagnosticsObserver:expectDataWithinTime( 5, {

		documentsSyncedByType = {},
		isCurrentlySyncing = false,
		pausedBySynchronizedOperation = 0,
		recentlySyncedDocIds = {},
		summaryStr = "Catalog Pusher synced up 0 new docs, 0 updates, 0 deletions to Oz.",
		totalDocumentsSynced = 0,

	} )

	-- Create one document of each type currently supported on Oz.

	rxTest:waitFor( docStore:bulkSetDocValue( docStoreMocks.docStoreValues ) )

	-- These documents should automatically be sent to Oz.

	rxTest:sleep( 1 )

	syncDiagnosticsObserver:expectDataWithinTime( 5, {

		documentsSyncedByType = {
			create = {
				album = {
					collection = 1,
				},
				album_asset = {
					NULL = 1,
				},
				asset = {
					image = 1,
				},
				total = 3,
			}
		},
		isCurrentlySyncing = false,
		pausedBySynchronizedOperation = 0,
		recentlySyncedDocIds = {
			assetDocId,
			albumAssetDocId,
			albumDocId,
		},
		summaryStr = "Catalog Pusher synced up 3 new docs, 0 updates, 0 deletions to Oz.",
		totalDocumentsSynced = 3,

	} )

	-- Now pull changes from Oz. There should be a single winning revision
	-- containing whatever we've received from Oz.

	rxTest:expectOnlyCompleted( puller:getUpdatesOnce() )

	rxTest:expectSingleValue(
		docStore:changedDocs { fromChangeCounter = 0 }
			:map( remapChangedDocsFeedForMockTests ),
		docStoreMocks.expectedDocStoreValues
	)

	-- Make a local change to the asset.

	rxTest:waitFor( docStore:applyOverlay( docStoreMocks.assetUpdateOverlay ) )

	-- Verify that asset document has one head revision.

	rxTest:expectSingleValue( docStore:currentRevisionsForDoc( assetDocId ), { docStoreMocks.revisionExpectedValue } )

	-- Should now send the locally-updated version of the asset document.

	rxTest:observe( catalogSpec._httpMock:requestsProcessed() )
	:expectDataWithinTime( 2, 5 )
	:dispose()

	syncDiagnosticsObserver:expectDataWithinTime( 5, {

		documentsSyncedByType = {
			create = {
				album = {
					collection = 1,
				},
				album_asset = {
					NULL = 1,
				},
				asset = {
					image = 1,
				},
				total = 3,
			},
			update = {
				asset = {
					image = 1,
				},
				total = 1,
			},
		},
		isCurrentlySyncing = false,
		pausedBySynchronizedOperation = 0,
		recentlySyncedDocIds = {
			assetDocId,
			albumAssetDocId,
			albumDocId,
			assetDocId,
		},
		summaryStr = "Catalog Pusher synced up 3 new docs, 1 updates, 0 deletions to Oz.",
		totalDocumentsSynced = 4,

	} )

end

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

function WFCatMetadataPusherTest.rxTestRecoverFromHTTPError_mock( rxTest )

	rxTest:setTimeLimit( 30 )

	local catalog = WFCatMockCatalog.createFakeWFCatalog( rxTest )

	local wasNetworkDown = Rx.Subject( 1 )

	rawset( catalog:ozCatalogSpec()._httpMock, 'put', function( user, url, params )

		wasNetworkDown:onNext( true )
		wasNetworkDown:onCompleted()

		return Rx.Observable.throw( 'no network connection' )

	end )

	local docStore = catalog:docStore()

	local pusher = WFCatMetadataPusher( catalog, options )
	rxTest:addCloseTarget( pusher )

	rxTest:waitFor( pusher:sendUpdatesContinuously() )

	rxTest:waitFor( docStore:applyOverlay {
		_id = string.rep( '1', 32 ),
		name = 'test',
		type = 'asset',
	} )

	rxTest:expectSingleValue( wasNetworkDown, true )

end

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

function WFCatMetadataPusherTest.rxTestSendDeletionAfterOverlay_NMB5690_mock( rxTest )

	-- Send new document to Oz, recieve confirmation. Then delete and apply a separate overlay
	-- after the deletion and ensure that the deletion is still sent to Oz. Before the fix
	-- for NMB-5690, the deletion would not be sent by WFCatMetadataPusher if an overlay was
	-- applied to the document after the deletion but before the WFCatMetadata collected the
	-- documents to push due to the lack of doc type info in the localDoc
	rxTest:setTimeLimit( 30 )

	local catalogId = KSIdGenerator:uuidInLowercaseHex()

	local docStoreMocks = generateDocStoreMocks( 'catalog', catalogId, { excludeAlbum = true, includeAssetDelete = true } )

	local assetDocId = docStoreMocks.assetDocId

	local newRevisionId = KSIdGenerator.uuidInLowercaseHex()
	docStoreMocks.assetDeleteMock.revision_id = newRevisionId
		-- the revisionId sent to Oz will come from the second overlay applied after the deletion

	local reqMocks = {

		pusherMockReqResp( 'catalog', catalogId, docStoreMocks.pusherUpdates ),
		prefetchReqResp( catalogId ),
		catalogPrefetchReqResp( catalogId ),
		pullerMockReqResp( 'catalog', catalogId, docStoreMocks.pullerUpdates ),
		pusherMockReqResp( 'catalog', catalogId, { docStoreMocks.assetDeleteMock } ),

	}

	local catalogSpec = WFCatSpecMock( catalogId, kUserId, rxTest, reqMocks )
	local catalog = WFCatMockCatalog.createWFCatalogForTestWithSpec( rxTest, catalogSpec )

	local docStore = catalog:docStore()

	local pusher = WFCatMetadataPusher( catalog, options )
	rxTest:addCloseTarget( pusher )

	local puller = WFCatMetadataPuller( catalog, "Test Puller", options )
	rxTest:addCloseTarget( puller )

	-- Create asset document

	rxTest:waitFor( docStore:bulkSetDocValue( docStoreMocks.docStoreValues ) )

	-- Send the doc to Oz.

	rxTest:waitFor( pusher:_DEBUG_sendNUpdates( 1 ) )

	-- Now pull changes from Oz.

	rxTest:expectOnlyCompleted( puller:getUpdatesOnce() )

	rxTest:expectSingleValue(
		docStore:changedDocs { fromChangeCounter = 0 }
			:map( remapChangedDocsFeedForMockTests ),
		docStoreMocks.expectedDocStoreValues
	)

	-- Should be no updates to send.

	rxTest:waitFor( pusher:_DEBUG_sendNUpdates( 0 ) )

	-- delete doc locally

	rxTest:waitFor( docStore:applyOverlay( docStoreMocks.assetDeleteOverlay ) )

	-- immediately apply a second overlay to the asset doc

	rxTest:waitFor(
		docStore:applyOverlay {
			_id = assetDocId,
			_newRevId = newRevisionId,
			_localOnly = {
				counter = 1
			}
		}
	)

	rxTest:waitFor( pusher:_DEBUG_sendNUpdates( 1 ) )

end

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

function WFCatMetadataPusherTest.rxTestPushChangesToOz_testNightly( rxTest )

	rxTest:setTimeLimit( 120 )

	local catalogTestSession = WFCatTestSession( rxTest )

	local catalog = rxTest:getFirstValue( catalogTestSession:catalog() )
	local catalogSpec = catalog:ozCatalogSpec()
	local docStore = catalog:docStore()

	local pusher = WFCatMetadataPusher( catalog, options )
	rxTest:addCloseTarget( pusher )

	local puller = WFCatMetadataPuller( catalog, "Test Puller", options )
	rxTest:addCloseTarget( puller )

	-- Pull changes feed once. By doing so, we'll get any error documents
	-- that might result from subsequent pushes.

	rxTest:expectOnlyCompleted( puller:getUpdatesOnce() )

	-- Create one document of each type currently supported on Oz.

	local assetDocId = KSIdGenerator.uuidInLowercaseHex()
	local albumDocId = KSIdGenerator.uuidInLowercaseHex()
	local albumAssetDocId = KSIdGenerator.uuidInLowercaseHex()

	rxTest:waitFor(

		docStore:bulkSetDocValue {
			{
				_id = assetDocId,
				captureDate = "2015-05-17T15:50:12",
				importSource = {
					fileName = 'test.jpg',
					importedBy = catalogSpec:ozUserId(),
					importTimestamp = "2015-05-26T17:34:23-07:00",
					originalHeight = 1600,
					originalWidth = 1068,
				},
				location = {
					altitude = 377.9,
					latitude = 45.951733,
					longitude = -93.600107,
				},
				subtype = "image",
				type = "asset",
				xmp = {
					dc = {
						description = "Rum River State Forest, near Onamia, Minnesota",
						rights = "test copyright string",
					},
				},
			},
			{
				_id = albumDocId,
				cover = {
					id = assetDocId,
				},
				name = "Sync me",
				subtype = "collection",
				type = "album",
			},
			{
				_id = albumAssetDocId,
				album = {
					id = albumDocId,
				},
				asset = {
					id = assetDocId,
				},
				cover = true,
				order = "V",
				type = "album_asset",
			},
		}

	)

	-- Send these docs to Oz.

	local sendNUpdates = rxTest:observe( pusher:_DEBUG_sendNUpdates( 3 ) )
	sendNUpdates:expectCompletedWithinTime( 30 )
	sendNUpdates:dispose()

	-- Make sure they landed successfully.

	local observeAlbumAssetUpdateFromOz = observeHasServerProvidedMetadata( rxTest, catalog, albumAssetDocId )
	local observeAlbumUpdateFromOz = observeHasServerProvidedMetadata( rxTest, catalog, albumDocId )
	local observeAssetUpdateFromOz = observeHasServerProvidedMetadata( rxTest, catalog, assetDocId )

	puller:getUpdatesContinuously()

	observeAlbumAssetUpdateFromOz:expectDataWithinTime( 40, true )
	observeAlbumAssetUpdateFromOz:dispose()

	observeAlbumUpdateFromOz:expectDataWithinTime( 40, true )
	observeAlbumUpdateFromOz:dispose()

	observeAssetUpdateFromOz:expectDataWithinTime( 40, true )
	observeAssetUpdateFromOz:dispose()

	puller:stopContinuousUpdates()

	rxTest:expectSingleValue(

		catalogSpec:get( catalogSpec:catalogHref() .. '/updates' )
			:map( KSJsonDecoder.decodeWebSafeJson_stripNull )
			:map( stripUpdateFeedLinkNoise )
			:map( function( changes, sortedDocs )

				for _, doc in ipairs( sortedDocs ) do

					if doc.type == 'asset' then
						doc.location = nil
					end

					doc.revision_ids = nil
				end

				return changes, sortedDocs

			end ),

		':ignore:',
		{
			{
				id = albumDocId,
				payload = {
					name = "Sync me",
					cover = {
						id = assetDocId,
					},
				},
				subtype = "collection",
				type = "album",
			},
			{
				id = albumAssetDocId,
				album = {
					id = albumDocId,
				},
				asset = {
					id = assetDocId,
				},
				payload = {
					cover = true,
					order = "V"
				},
				type = "album_asset",
			},
			{
				id = assetDocId,
				payload = {
					captureDate = "2015-05-17T15:50:12",
					importSource = {
						fileName = 'test.jpg',
						importedBy = catalogSpec:ozUserId(),
						importTimestamp = "2015-05-26T17:34:23-07:00",
						originalHeight = 1600,
						originalWidth = 1068,
					},
					xmp = {
						dc = {
							description = "Rum River State Forest, near Onamia, Minnesota",
							rights = "test copyright string",
						},
					},
				},
				subtype = "image",
				type = "asset",
			},
		} )

	-- Now examine the changes as pulled back from Oz.

	rxTest:expectSingleValue(

		docStore:changedDocs { fromChangeCounter = 0 }
			:map( remapChangedDocsFeed )
			:map( function( changes )

				-- Remove the location as Oz does geo-location decoding which we sometimes
				-- catch as part of the sync feed changes

				for _, doc in ipairs( changes ) do

					doc.location = nil

				end

				changes.continueFrom = nil

				return changes

			end ),

		{
			{
				_labels = {
					oz = true,
				},
				_serverProvidedMetadata = {
					base = ">>base",
					created = ">>created",
					links = ">>links",
					revision_ids = ">>revision_ids",
					updated = ">>updated",
				},
				cover = {
					id = assetDocId,
				},
				name = "Sync me",
				subtype = "collection",
				type = "album",
			},
			{
				_labels = {
					oz = true,
				},
				_serverProvidedMetadata = {
					base = ">>base",
					created = ">>created",
					links = ">>links",
					revision_ids = ">>revision_ids",
					updated = ">>updated",
				},
				album = {
					id = albumDocId,
				},
				asset = {
					id = assetDocId,
				},
				cover = true,
				order = "V",
				type = "album_asset",
			},
			{
				_labels = {
					oz = true,
				},
				_serverProvidedMetadata = {
					base = ">>base",
					created = ">>created",
					links = ">>links",
					revision_ids = ">>revision_ids",
					updated = ">>updated",
				},
				captureDate = "2015-05-17T15:50:12",
				importSource = {
					fileName = 'test.jpg',
					importedBy = catalogSpec:ozUserId(),
					importTimestamp = "2015-05-26T17:34:23-07:00",
					originalHeight = 1600,
					originalWidth = 1068,
				},
				subtype = "image",
				type = "asset",
				xmp = {
					dc = {
						description = "Rum River State Forest, near Onamia, Minnesota",
						rights = "test copyright string",
					},
				},
			},
			{
				_labels = {
					oz = true,
				},
				_serverProvidedMetadata = {
					base = ">>base",
					created = ">>created",
					links = ">>links",
					revision_ids = ">>revision_ids",
					updated = ">>updated",
				},
				name = "WFCatTestSession",
				subtype = "lightroom",
				type = "catalog",
			},
		} )

	-- Should be no updates to send.

	sendNUpdates = rxTest:observe( pusher:_DEBUG_sendNUpdates( 0 ) )
	sendNUpdates:expectCompletedWithinTime( 30 )
	sendNUpdates:dispose()

	-- And also, no updates to receive.

	rxTest:sleep( 6 )

	rxTest:expectOnlyCompleted( puller:getUpdatesOnce() )

end

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

function WFCatMetadataPusherTest.rxTestLocalDeleteBeforeReadbackFromOz_mock( rxTest )

	-- Send new documents to Oz, delete locally after that, simulate readback
	-- first change from Oz (after send).

	local catalogId = KSIdGenerator:uuidInLowercaseHex()

	local docStoreMocks = generateDocStoreMocks( 'catalog', catalogId, { includeAssetDelete = true } )

	local assetDocId = docStoreMocks.assetDocId

	local reqMocks = {

		pusherMockReqResp( 'catalog', catalogId, docStoreMocks.pusherUpdates ),
		prefetchReqResp( catalogId ),
		catalogPrefetchReqResp( catalogId ),
		pullerMockReqResp( 'catalog', catalogId, docStoreMocks.pullerUpdates ),
		pullerMockReqResp( 'catalog', catalogId, {} ),

	}

	local catalogSpec = WFCatSpecMock( catalogId, kUserId, rxTest, reqMocks )
	local catalog = WFCatMockCatalog.createWFCatalogForTestWithSpec( rxTest, catalogSpec )

	local docStore = catalog:docStore()

	local pusher = WFCatMetadataPusher( catalog, options )
	rxTest:addCloseTarget( pusher )

	local puller = WFCatMetadataPuller( catalog, "Test Puller", options )
	rxTest:addCloseTarget( puller )

	-- Create one document of each type currently supported on Oz.

	rxTest:waitFor( docStore:bulkSetDocValue( docStoreMocks.docStoreValues ) )

	-- Send these docs to Oz.

	rxTest:waitFor( pusher:_DEBUG_sendNUpdates( 3 ) )

	-- Delete the asset locally. Note that, unlike previous doc stores in WF,
	-- :deleteDoc doesn't clear out the body of the document.

	local localRevId = rxTest:getFirstValue( docStore:deleteDoc( assetDocId ) )

	-- Now pull changes from Oz. This payload will contain a conflicting version
	-- for this asset document. (It might not *seem* like a conflict, but better
	-- to flag it this way and resolve later.)

	rxTest:expectOnlyCompleted( puller:getUpdatesOnce() )

	docStoreMocks.deletedAssetExpectedValue._conflict = true
	docStoreMocks.deletedAssetExpectedValue._revId = localRevId
	docStoreMocks.expectedDocStoreValues.continueFrom = 9
	docStoreMocks.expectedDocStoreValues[3] = docStoreMocks.deletedAssetExpectedValue

	rxTest:expectSingleValue(
		docStore:changedDocs { fromChangeCounter = 0 }
			:map( remapChangedDocsFeedForMockTests ),
		docStoreMocks.expectedDocStoreValues
	)

	-- Verify that asset document has two head revisions,
	-- one of which is deleted.

	local localRevisionExpectedValue = docStoreMocks.deletedRevisionExpectedValue
	localRevisionExpectedValue._changeCounter = 4
	localRevisionExpectedValue._conflict = true
	localRevisionExpectedValue._revId = localRevId

	local ozRevisionExpectedValue = KSTableUtils.deepCopy( localRevisionExpectedValue )
	ozRevisionExpectedValue._changeCounter = 6
	ozRevisionExpectedValue._deleted = nil
	ozRevisionExpectedValue._revId = docStoreMocks.assetRevisionId

	local showDeletedRevisions = true
	rxTest:expectSingleValue( docStore:currentRevisionsForDoc( assetDocId, showDeletedRevisions ), { localRevisionExpectedValue, ozRevisionExpectedValue } )

	-- Should be no updates to send.

	rxTest:waitFor( pusher:_DEBUG_sendNUpdates( 0 ) )

	-- And also, no updates to receive.

	rxTest:expectOnlyCompleted( puller:getUpdatesOnce() )

	-- Now resolve the conflict by deleting the version from Oz.

	local deleteConflictRevId = rxTest:getFirstValue( docStore:deleteDoc( assetDocId, docStoreMocks.assetRevisionId ) )

	-- register the mock now that we have the revision id for the deletion

	docStoreMocks.assetDeleteMock.revision_id = localRevId

	local deleteMock = pusherMockReqResp( 'catalog', catalogId, { docStoreMocks.assetDeleteMock } )
	deleteMock.dataFilter = function( str )
		return str:gsub( deleteConflictRevId, localRevId )
			-- either of these revision ids might be used, not deterministic which
	end

	catalogSpec:expectHttpRequest( deleteMock )

	-- Should now send the locally-originated version of the asset document.

	rxTest:waitFor( pusher:_DEBUG_sendNUpdates( 1 ) )

end

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

function WFCatMetadataPusherTest.rxTestLocalDeleteAfterReadbackFromOz_mock( rxTest )

	-- Send new documents to Oz, read back from Oz, then delete locally;
	-- make sure we send correct delta.

	local catalogId = KSIdGenerator:uuidInLowercaseHex()

	local docStoreMocks = generateDocStoreMocks( 'catalog', catalogId, { includeAssetDelete = true } )

	local assetDocId = docStoreMocks.assetDocId

	local reqMocks = {

		pusherMockReqResp( 'catalog', catalogId, docStoreMocks.pusherUpdates ),
		prefetchReqResp( catalogId ),
		catalogPrefetchReqResp( catalogId ),
		pullerMockReqResp( 'catalog', catalogId, docStoreMocks.pullerUpdates ),

	}

	local catalogSpec = WFCatSpecMock( catalogId, kUserId, rxTest, reqMocks )
	local catalog = WFCatMockCatalog.createWFCatalogForTestWithSpec( rxTest, catalogSpec )

	local docStore = catalog:docStore()

	local pusher = WFCatMetadataPusher( catalog, options )
	rxTest:addCloseTarget( pusher )

	local puller = WFCatMetadataPuller( catalog, "Test Puller", options )
	rxTest:addCloseTarget( puller )

	-- Create one document of each type currently supported on Oz.

	rxTest:waitFor( docStore:bulkSetDocValue( docStoreMocks.docStoreValues ) )

	-- Send these docs to Oz.

	rxTest:waitFor( pusher:_DEBUG_sendNUpdates( 3 ) )

	-- Now pull changes from Oz. There should be a single winning revision
	-- containing whatever we've received from Oz.

	rxTest:expectOnlyCompleted( puller:getUpdatesOnce() )

	rxTest:expectSingleValue(
		docStore:changedDocs { fromChangeCounter = 0 }
			:map( remapChangedDocsFeedForMockTests ),
		docStoreMocks.expectedDocStoreValues
	)

	-- Delete the asset locally. Note that, unlike previous doc stores in WF,
	-- :deleteDoc doesn't clear out the body of the document.

	local localRevId = rxTest:getFirstValue( docStore:deleteDoc( assetDocId ) )

	-- register the mock now that we have the revision id for the deletion

	docStoreMocks.assetDeleteMock.revision_id = localRevId
	catalogSpec:expectHttpRequest( pusherMockReqResp( 'catalog', catalogId, { docStoreMocks.assetDeleteMock } ) )

	-- Verify that asset document has one head revision.

	docStoreMocks.deletedRevisionExpectedValue._revId = localRevId
	local showDeletedRevisions = true
	rxTest:expectSingleValue( docStore:currentRevisionsForDoc( assetDocId, showDeletedRevisions ), { docStoreMocks.deletedRevisionExpectedValue } )

	-- Should now send the locally-updated version of the asset document.

	rxTest:waitFor( pusher:_DEBUG_sendNUpdates( 1 ) )

end


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

function WFCatMetadataPusherTest.rxTestSendSpaceFeedUpdates_mock( rxTest )

	rxTest:setTimeLimit( 8 )

	local spaceId = KSIdGenerator.uuidInLowercaseHex()
	local catalogId = KSIdGenerator:uuidInLowercaseHex()

	local docStoreMocks = generateDocStoreMocks( 'space', spaceId, { includeAssetUpdate = true } )

	local assetDocId = docStoreMocks.assetDocId

	local reqMocks = {

		pusherMockReqResp( 'space', spaceId, docStoreMocks.pusherUpdates ),
		pullerMockReqResp( 'space', spaceId, docStoreMocks.pullerUpdates ),
		pusherMockReqResp( 'space', spaceId, { docStoreMocks.assetUpdateMock } ),

	}

	local catalogSpec = WFCatSpecMock( catalogId, kUserId, rxTest, reqMocks )
	local catalog = WFCatMockCatalog.createWFCatalogForTestWithSpec( rxTest, catalogSpec )

	local docStore = catalog:docStore()

	local spaceoptions = KSTableUtils.deepCopy( options )
	spaceoptions.isSpacePusher = true

	local pusher = WFCatMetadataPusher( catalog, spaceoptions )
	rxTest:addCloseTarget( pusher )

	local spaceUpdatesUrl = catalog:ozCatalogSpec():spaceUpdatesUrlForSpaceId( spaceId )

	local pullerOptions = KSTableUtils.deepCopy( options )
	pullerOptions.spaceId = spaceId
	pullerOptions.updatesFeedUrl = spaceUpdatesUrl
	pullerOptions.isCatalogPuller = false

	local puller = WFCatMetadataPuller( catalog, spaceId, pullerOptions )
	rxTest:addCloseTarget( puller )

	-- Create one document of each type currently supported on Oz.

	rxTest:waitFor( docStore:bulkSetDocValue( docStoreMocks.docStoreValues ) )

	-- Send these docs to Oz.

	rxTest:waitFor( pusher:_DEBUG_sendNUpdates( 3 ) )

	-- Now pull changes from Oz. There should be a single winning revision
	-- containing whatever we've received from Oz.

	rxTest:expectOnlyCompleted( puller:getUpdatesOnce() )

	rxTest:expectSingleValue(
		docStore:changedDocs { fromChangeCounter = 0 }
			:map( remapChangedDocsFeedForMockTests ),
		docStoreMocks.expectedDocStoreValues
	)

	-- Make a local change to the asset.

	rxTest:waitFor( docStore:applyOverlay( docStoreMocks.assetUpdateOverlay ) )

	-- Send the asset update doc to Oz.

	rxTest:waitFor( pusher:_DEBUG_sendNUpdates( 1 ) )

	-- Verify that asset document has one head revision.

	rxTest:expectSingleValue( docStore:currentRevisionsForDoc( assetDocId ), { docStoreMocks.revisionExpectedValue } )

	rxTest:observe( catalogSpec._httpMock:requestsProcessed() )
		:expectDataWithinTime( 2, 3 )
		:dispose()

end

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

function WFCatMetadataPusherTest.rxTestSendUpdatesForMultipleSpaces_mock( rxTest )

	rxTest:setTimeLimit( 8 )

	local spaceId1 = KSIdGenerator.uuidInLowercaseHex()
	local spaceId2 = KSIdGenerator.uuidInLowercaseHex()

	local catalogId = KSIdGenerator:uuidInLowercaseHex()

	local docStoreMocksSpace1 = generateDocStoreMocks( 'space', spaceId1,
			{ excludeAlbum = true, customDescription =  KSIdGenerator.uuidInLowercaseHex() } )
	local docStoreMocksSpace2 = generateDocStoreMocks( 'space', spaceId2,
			{ excludeAlbum = true, customDescription =  KSIdGenerator.uuidInLowercaseHex() } )
	local pusherUpdates = { docStoreMocksSpace1.pusherUpdates[1], docStoreMocksSpace2.pusherUpdates[1] }

	local reqMocks = {

		pusherMockReqResp( 'space', spaceId1, pusherUpdates ),
		pullerMockReqResp( 'space', spaceId1, docStoreMocksSpace1.pullerUpdates ),
		pullerMockReqResp( 'space', spaceId2, docStoreMocksSpace2.pullerUpdates ),

	}

	local catalogSpec = WFCatSpecMock( catalogId, kUserId, rxTest, reqMocks )
	local catalog = WFCatMockCatalog.createWFCatalogForTestWithSpec( rxTest, catalogSpec )

	local docStore = catalog:docStore()

	local spaceoptions = KSTableUtils.deepCopy( options )
	spaceoptions.isSpacePusher = true

	local pusher = WFCatMetadataPusher( catalog, spaceoptions )
	rxTest:addCloseTarget( pusher )

	local spaceUpdatesUrl = catalog:ozCatalogSpec():spaceUpdatesUrlForSpaceId( spaceId1 )

	local pullerOptions1 = KSTableUtils.deepCopy( options )
	pullerOptions1.spaceId = spaceId1
	pullerOptions1.updatesFeedUrl = spaceUpdatesUrl
	pullerOptions1.isCatalogPuller = false

	local puller1 = WFCatMetadataPuller( catalog, spaceId1, pullerOptions1 )
	rxTest:addCloseTarget( puller1 )

	local space2UpdatesUrl = catalog:ozCatalogSpec():spaceUpdatesUrlForSpaceId( spaceId2 )

	local pullerOptions2 = KSTableUtils.deepCopy( options )
	pullerOptions2.spaceId = spaceId2
	pullerOptions2.updatesFeedUrl = space2UpdatesUrl
	pullerOptions2.isCatalogPuller = false

	local puller2 = WFCatMetadataPuller( catalog, spaceId2, pullerOptions2 )
	rxTest:addCloseTarget( puller2 )

	-- Create two assets in  docStore, each in a separate space

	local docStoreValues = { docStoreMocksSpace1.docStoreValues[1], docStoreMocksSpace2.docStoreValues[1] }
	rxTest:waitFor( docStore:bulkSetDocValue( docStoreValues ) )

	-- sync updates to Oz

	rxTest:waitFor( pusher:_DEBUG_sendNUpdates( 2 ) )

	-- Now pull changes from Oz from space 1. should have sync metadata for that
	-- asset but none yet for the asset from space 2

	rxTest:expectOnlyCompleted( puller1:getUpdatesOnce() )

	local expectedDocStoreValues = {

		docStoreMocksSpace2.docStoreValues[1],
		docStoreMocksSpace1.expectedDocStoreValues[1],
		continueFrom = 4,

	}

	expectedDocStoreValues[1]._revId = expectedDocStoreValues[1]._newRevId
	expectedDocStoreValues[1]._newRevId = nil

	rxTest:expectSingleValue(
		docStore:changedDocs { fromChangeCounter = 0 }
			:map( remapChangedDocsFeedForMockTests ),
		expectedDocStoreValues
	)

	-- Now pull changes from Oz from space 2. should have sync metadata for both assets

	rxTest:expectOnlyCompleted( puller2:getUpdatesOnce() )

	expectedDocStoreValues = {

		docStoreMocksSpace1.expectedDocStoreValues[1],
		docStoreMocksSpace2.expectedDocStoreValues[1],
		continueFrom = 5,

	}

	rxTest:expectSingleValue(
		docStore:changedDocs { fromChangeCounter = 0 }
			:map( remapChangedDocsFeedForMockTests ),
		expectedDocStoreValues
	)

	rxTest:observe( catalogSpec._httpMock:requestsProcessed() )
		:expectDataWithinTime( 2, 3 )
		:dispose()

end

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

function WFCatMetadataPusherTest.rxTestSendUpdatesStripsChangesToOriginalHash_mock( rxTest )

	local catalogId = KSIdGenerator:uuidInLowercaseHex()
	local catalogSpec = WFCatSpecMock( catalogId, kUserId, rxTest,

		{
			-- should be no HTTP comm if no changes to send
		} )

	local catalog = WFCatMockCatalog.createWFCatalogForTestWithSpec( rxTest, catalogSpec )

	local pusher = WFCatMetadataPusher( catalog, options )
	rxTest:addCloseTarget( pusher )

	local docStoreMocks = generateDocStoreMocks( 'catalog', catalogId, { excludeAlbum = true } )

	local assetDocId = docStoreMocks.assetDocId

	docStoreMocks.docStoreValues.label = 'oz'
	rxTest:waitFor( catalog:docStore():bulkSetDocValue( docStoreMocks.docStoreValues ) )

	rxTest:waitFor( catalog:docStore():applyOverlay(
			{
				_id = assetDocId,
				original = {
					id = KSIdGenerator.uuidInLowercaseHex(),
					userCreated = '2018-12-04T07:35:22.111Z'
				},
			} )
	)

	local updateCount = rxTest:observe( pusher:countUpdatesToSend() )
	updateCount:expectData( 2 )

	rxTest:waitFor( pusher:_DEBUG_sendNUpdates( 0 ) )

	rxTest:expectSingleValue( catalog:docStore():localDocValue( 'Oz-sendUpdatesContinueFrom' )
		:pluck( 'continueFrom' ),
		3 )

end

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

function WFCatMetadataPusherTest.rxTestSendSpaceFeedUpdates_OverQuota_mock( rxTest )

	rxTest:setTimeLimit( 8 )

	local spaceId = KSIdGenerator.uuidInLowercaseHex()
	local catalogId = KSIdGenerator:uuidInLowercaseHex()

	local docStoreMocks = generateDocStoreMocks( 'space', spaceId )

	local reqMocks = {

		pusherMockReqResp( 'space', spaceId, { docStoreMocks.pusherUpdates[3] } ),
		pullerMockReqResp( 'space', spaceId, { docStoreMocks.pullerUpdates[3] } ),
			-- only include album document for pusher and puller mocks

	}

	local catalogSpec = WFCatSpecMock( catalogId, kUserId, rxTest, reqMocks )
	local catalog = WFCatMockCatalog.createWFCatalogForTestWithSpec( rxTest, catalogSpec )

	local docStore = catalog:docStore()

	local spaceoptions = KSTableUtils.deepCopy( options )
	spaceoptions.isSpacePusher = true
	spaceoptions.albumAssetsForFilteredAssetsQueryFn = albumAssetsForFilteredAssetsQueryFn

	local pusher = WFCatMetadataPusher( catalog, spaceoptions )
	rxTest:addCloseTarget( pusher )

	-- Set the space as over quota.

	pusher:setSpaceIsOverQuota( spaceId, true )

	local spaceUpdatesUrl = catalog:ozCatalogSpec():spaceUpdatesUrlForSpaceId( spaceId )

	local pullerOptions = KSTableUtils.deepCopy( options )
	pullerOptions.spaceId = spaceId
	pullerOptions.updatesFeedUrl = spaceUpdatesUrl
	pullerOptions.isCatalogPuller = false

	local puller = WFCatMetadataPuller( catalog, "Test Puller", pullerOptions )
	rxTest:addCloseTarget( puller )

	-- Create one document of each type currently supported on Oz.

	rxTest:waitFor( docStore:bulkSetDocValue( docStoreMocks.docStoreValues ) )

	-- This should only push the album asset since the owner is over quota.

	local sendNUpdates = rxTest:observe( pusher:_DEBUG_sendUpdatesOnce() )
	sendNUpdates:expectCompletedWithinTime( 5 )
	sendNUpdates:dispose()

	-- Now pull changes from Oz. Only album will have sync data from oz.

	rxTest:expectOnlyCompleted( puller:getUpdatesOnce() )

	local localError = {
		ozError = {
			reason = "prefiltered",
			subtype = "StorageExceededError",
		}
	}

	local expectedDocStoreValues = {

		docStoreMocks.expectedDocStoreValues[1], -- album with sync metadata
		docStoreMocks.docStoreValues[2], -- album_assset without sync metatdata
		docStoreMocks.docStoreValues[1], -- asset without sync metadata
		continueFrom = 7,

	}

	expectedDocStoreValues[2]._localOnly = localError
	expectedDocStoreValues[2]._revId = expectedDocStoreValues[2]._newRevId
	expectedDocStoreValues[2]._newRevId = nil

	expectedDocStoreValues[3]._localOnly = localError
	expectedDocStoreValues[3]._revId = expectedDocStoreValues[3]._newRevId
	expectedDocStoreValues[3]._newRevId = nil

	rxTest:expectSingleValue(
		docStore:changedDocs { fromChangeCounter = 0, includeDocumentsFromSource = WFRxIndexStoreSchema.sourceSpaceFeedDocuments, }
			:map( remapChangedDocsFeedForMockTests ),
		expectedDocStoreValues
	)

	rxTest:observe( catalogSpec._httpMock:requestsProcessed() )
		:expectDataWithinTime( 2, 2 )
		:dispose()

end


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

-- FIXME [jandrew 2025-03-11]: Disabled due to instability.

function WFCatMetadataPusherTest.DISABLED_rxTestSendSpaceFeedUpdates_OwnerExpired_CreateDoc_mock( rxTest )

	rxTest:setTimeLimit( 8 )

	local spaceId = KSIdGenerator.uuidInLowercaseHex()
	local catalogId = KSIdGenerator:uuidInLowercaseHex()

	local docStoreMocks = generateDocStoreMocks( 'space', spaceId )

	local catalogSpec = WFCatSpecMock( catalogId, kUserId, rxTest )
	local catalog = WFCatMockCatalog.createWFCatalogForTestWithSpec( rxTest, catalogSpec )

	local docStore = catalog:docStore()

	local spaceoptions = KSTableUtils.deepCopy( options )
	spaceoptions.isSpacePusher = true
	spaceoptions.albumAssetsForFilteredAssetsQueryFn = albumAssetsForFilteredAssetsQueryFn

	local pusher = WFCatMetadataPusher( catalog, spaceoptions )
	rxTest:addCloseTarget( pusher )

	-- Set the space owner as having an expired account.

	pusher:setSpaceIsExpired( spaceId, true )

	local spaceUpdatesUrl = catalog:ozCatalogSpec():spaceUpdatesUrlForSpaceId( spaceId )

	local pullerOptions = KSTableUtils.deepCopy( options )
	pullerOptions.spaceId = spaceId
	pullerOptions.updatesFeedUrl = spaceUpdatesUrl
	pullerOptions.isCatalogPuller = false

	local puller = WFCatMetadataPuller( catalog, "Test Puller", pullerOptions )
	rxTest:addCloseTarget( puller )

	-- Create one document of each type currently supported on Oz.

	rxTest:waitFor( docStore:bulkSetDocValue( docStoreMocks.docStoreValues ) )

	-- This should not push anything since the owner's account is expired.

	local sendNUpdates = rxTest:observe( pusher:_DEBUG_sendUpdatesOnce() )
	sendNUpdates:expectCompletedWithinTime( 5 )
	sendNUpdates:dispose()

	local localError = {
		ozError = {
			reason = "prefiltered",
			subtype = "TrialOrSubscriptionExpiredError",
		}
	}

	local expectedDocStoreValues = {
		docStoreMocks.docStoreValues[3],
		docStoreMocks.docStoreValues[2],
		docStoreMocks.docStoreValues[1],
		continueFrom = 7,
	}

	for _, v in ipairs( expectedDocStoreValues ) do
		v._localOnly = localError
		v._id = nil
		v._revId = nil
		v._newRevId = nil
	end

	rxTest:expectSingleValue(
		docStore:changedDocs { fromChangeCounter = 0, includeDocumentsFromSource = WFRxIndexStoreSchema.sourceSpaceFeedDocuments, }
			:map( remapChangedDocsFeed ),
		expectedDocStoreValues
	)

	rxTest:observe( catalogSpec._httpMock:requestsProcessed() )
		:expectDataWithinTime( 2, 0 )
		:dispose()

end

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

function WFCatMetadataPusherTest.rxTestSendSpaceFeedUpdates_OwnerExpired_UpdateDoc_mock( rxTest )

	rxTest:setTimeLimit( 8 )

	local spaceId = "17989b76100d40c49913beba0087d129"
	local catalogId = KSIdGenerator:uuidInLowercaseHex()

	local docStoreMocks = generateDocStoreMocks( 'space', spaceId, { includeAssetUpdate = true } )

	local assetDocId = docStoreMocks.assetDocId

	local reqMocks = {

		pusherMockReqResp( 'space', spaceId, docStoreMocks.pusherUpdates ),
		pullerMockReqResp( 'space', spaceId, docStoreMocks.pullerUpdates ),

	}

	local catalogSpec = WFCatSpecMock( catalogId, kUserId, rxTest, reqMocks )
	local catalog = WFCatMockCatalog.createWFCatalogForTestWithSpec( rxTest, catalogSpec )

	local docStore = catalog:docStore()

	local spaceoptions = KSTableUtils.deepCopy( options )
	spaceoptions.isSpacePusher = true

	local pusher = WFCatMetadataPusher( catalog, spaceoptions )
	rxTest:addCloseTarget( pusher )

	local spaceUpdatesUrl = catalog:ozCatalogSpec():spaceUpdatesUrlForSpaceId( spaceId )

	local pullerOptions = KSTableUtils.deepCopy( options )
	pullerOptions.spaceId = spaceId
	pullerOptions.updatesFeedUrl = spaceUpdatesUrl
	pullerOptions.isCatalogPuller = false

	local puller = WFCatMetadataPuller( catalog, spaceId, pullerOptions )
	rxTest:addCloseTarget( puller )

	-- Create one document of each type currently supported on Oz.

	rxTest:waitFor( docStore:bulkSetDocValue( docStoreMocks.docStoreValues ) )

	rxTest:waitFor( pusher:_DEBUG_sendNUpdates( 3 ) )

	-- Now pull changes from Oz. There should be a single winning revision
	-- containing whatever we've received from Oz.

	rxTest:expectOnlyCompleted( puller:getUpdatesOnce() )

	rxTest:expectSingleValue(
		docStore:changedDocs { fromChangeCounter = 0, includeDocumentsFromSource = WFRxIndexStoreSchema.sourceSpaceFeedDocuments, }
			:map( remapChangedDocsFeedForMockTests ),
		docStoreMocks.expectedDocStoreValues
	)

	-- Set the space owner as having an expired account.

	pusher:setSpaceIsExpired( spaceId, true )

	-- Make a local change to the asset.

	rxTest:waitFor( docStore:applyOverlay( docStoreMocks.assetUpdateOverlay ) )

	rxTest:waitFor( pusher:_DEBUG_sendNUpdates( 0 ) )

	-- Verify that asset document has marked with TrialOrSubscriptionExpiredError error

	local localError = {
		ozError = {
			reason = "prefiltered",
			subtype = "TrialOrSubscriptionExpiredError",
		}
	}

	docStoreMocks.revisionExpectedValue._localOnly = localError
	docStoreMocks.revisionExpectedValue._changeCounter = 8

	rxTest:expectSingleValue( docStore:currentRevisionsForDoc( assetDocId ), { docStoreMocks.revisionExpectedValue } )

	rxTest:observe( catalogSpec._httpMock:requestsProcessed() )
		:expectDataWithinTime( 2, 2 )
		:dispose()

end

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

function WFCatMetadataPusherTest.rxTestSendSpaceFeedUpdates_OwnerExpired_DeleteDoc_mock( rxTest )

	rxTest:setTimeLimit( 8 )

	local spaceId = "17989b76100d40c49913beba0087d129"
	local catalogId = KSIdGenerator:uuidInLowercaseHex()

	local docStoreMocks = generateDocStoreMocks( 'space', spaceId, { includeAssetDelete = true } )

	local reqMocks = {

		pusherMockReqResp( 'space', spaceId, docStoreMocks.pusherUpdates ),
		pullerMockReqResp( 'space', spaceId, docStoreMocks.pullerUpdates ),

	}

	local catalogSpec = WFCatSpecMock( catalogId, kUserId, rxTest, reqMocks )
	local catalog = WFCatMockCatalog.createWFCatalogForTestWithSpec( rxTest, catalogSpec )

	local docStore = catalog:docStore()

	local spaceoptions = KSTableUtils.deepCopy( options )
	spaceoptions.isSpacePusher = true

	local pusher = WFCatMetadataPusher( catalog, spaceoptions )
	rxTest:addCloseTarget( pusher )

	local spaceUpdatesUrl = catalog:ozCatalogSpec():spaceUpdatesUrlForSpaceId( spaceId )

	local pullerOptions = KSTableUtils.deepCopy( options )
	pullerOptions.spaceId = spaceId
	pullerOptions.updatesFeedUrl = spaceUpdatesUrl
	pullerOptions.isCatalogPuller = false

	local puller = WFCatMetadataPuller( catalog, spaceId, pullerOptions )
	rxTest:addCloseTarget( puller )

	-- Create one document of each type currently supported on Oz.

	rxTest:waitFor( docStore:bulkSetDocValue( docStoreMocks.docStoreValues ) )

	rxTest:waitFor( pusher:_DEBUG_sendNUpdates( 3 ) )

	-- Now pull changes from Oz. There should be a single winning revision
	-- containing whatever we've received from Oz.

	rxTest:expectOnlyCompleted( puller:getUpdatesOnce() )

	rxTest:expectSingleValue(
		docStore:changedDocs { fromChangeCounter = 0, includeDocumentsFromSource = WFRxIndexStoreSchema.sourceSpaceFeedDocuments, }
			:map( remapChangedDocsFeedForMockTests ),
		docStoreMocks.expectedDocStoreValues
	)

	-- Set the space owner as having an expired account.

	pusher:setSpaceIsExpired( spaceId, true )

	-- Delete asset locally

	rxTest:waitFor( docStore:applyOverlay( docStoreMocks.assetDeleteOverlay ) )

	rxTest:waitFor( pusher:_DEBUG_sendNUpdates( 0 ) )

	-- Verify that deleted asset document has marked with TrialOrSubscriptionExpiredError error

	local localError = {
		ozError = {
			reason = "prefiltered",
			subtype = "TrialOrSubscriptionExpiredError",
		}
	}

	docStoreMocks.deletedRevisionExpectedValue._localOnly = localError
	docStoreMocks.deletedRevisionExpectedValue._changeCounter = 8

	rxTest:observe( catalogSpec._httpMock:requestsProcessed() )
		:expectDataWithinTime( 2, 2 )
		:dispose()

end

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

function WFCatMetadataPusherTest.rxTestSendSpaceFeedUpdates_Editor_mock( rxTest )

	rxTest:setTimeLimit( 10 )

	local spaceoptions = KSTableUtils.deepCopy( options )
	spaceoptions.isSpacePusher = true
	spaceoptions.useSpaceFeedDocTypeAllowList = true

	local catalog = WFCatMockCatalog.createFakeWFCatalog( rxTest )
	local pusher = WFCatMetadataPusher( catalog, spaceoptions )
	rxTest:addCloseTarget( pusher )

	local spaceId = KSIdGenerator.uuidInLowercaseHex()

	local ozDoc = {
		_id = KSIdGenerator.uuidInLowercaseHex(),
		_spaceId = spaceId,
		type = 'asset',
		subtype = 'image',
		_revId = KSIdGenerator.uuidInLowercaseHex(),
		_serverProvidedMetadata = {
			base = 'base_url',
			created = '2019-09-05T09:07:01.002Z',
			links = { ['self'] = 'self_url_info' },
			revision_ids = { KSIdGenerator.uuidInLowercaseHex() },
			updated = '2019-09-05T09:07:02.002Z',
			editable = true,
		},
		develop = {
			xmpCameraRaw = "edit 1",
		},
	}

	local function extractPayload( delta )
		if delta then
			return delta.payload
		end
	end


	local ozDocTest, localDocTest

	-- 1. Update is sent for image assets if Oz confirms it is editable
	ozDocTest = KSTableUtils.deepCopy( ozDoc )
	localDocTest = KSTableUtils.deepCopy( ozDocTest )

	localDocTest.develop.xmpCameraRaw = "edit 2"
	localDocTest._revId = KSIdGenerator.uuidInLowercaseHex()

	rxTest:expectEqual( { develop = localDocTest.develop }, extractPayload( pusher:DEBUG_convertOneDocToDelta( localDocTest, ozDocTest ) ) )


	-- 2. Update is sent for video assets if Oz confirms it is editable
	ozDocTest = KSTableUtils.deepCopy( ozDoc )
	ozDocTest.subtype = 'video'

	localDocTest = KSTableUtils.deepCopy( ozDocTest )

	localDocTest.develop.xmpCameraRaw = "edit 3"
	localDocTest._revId = KSIdGenerator.uuidInLowercaseHex()

	rxTest:expectEqual( { develop = localDocTest.develop }, extractPayload( pusher:DEBUG_convertOneDocToDelta( localDocTest, ozDocTest ) ) )

end

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

function WFCatMetadataPusherTest.rxTestWorkHasCompletedStream( rxTest )

	local catalog = WFCatMockCatalog.createFakeWFCatalog( rxTest )

	local pusher = WFCatMetadataPusher( catalog, options )
	rxTest:addCloseTarget( pusher )

	local workCompletionCountStream = Rx.ReplaySubject( 1 )
	local workCompletionCountStreamObserver = rxTest:observe( workCompletionCountStream )

	local workHasCompletedCount = 0

	local workHasCompletedStream = pusher:workHasCompletedStream()

		:tap( function()

			workHasCompletedCount = workHasCompletedCount + 1

			workCompletionCountStream:onNext( workHasCompletedCount )

		end )

		:subscribe()

	pusher._isDoingWork:onNext( true )
	pusher._isDoingWork:onNext( true )
	pusher._isDoingWork:onNext( false ) -- work completed 1
	pusher._isDoingWork:onNext( false )
	pusher._isDoingWork:onNext( true )
	pusher._isDoingWork:onNext( false ) -- work completed 2
	pusher._isDoingWork:onNext( true )
	pusher._isDoingWork:onNext( false ) -- work completed 3
	pusher._isDoingWork:onNext( false )
	pusher._isDoingWork:onNext( false )
	pusher._isDoingWork:onNext( true )
	pusher._isDoingWork:onNext( true )
	pusher._isDoingWork:onNext( false ) -- work completed 4
	pusher._isDoingWork:onNext( true )

	workCompletionCountStreamObserver:expectDataWithinTime( 1, 4 )

	workCompletionCountStreamObserver:dispose()
	workHasCompletedStream:dispose()

end

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

function WFCatMetadataPusherTest.rxTestMakeDocValueAwareBufferWithTimeOrCountStream( rxTest )

	rxTest:setTimeLimit( 10 )

	local wfCatalog = WFCatMockCatalog.createFakeWFCatalog( rxTest )

	local docValueStream = Rx.ReplaySubject( 100 )

	local docValues = {}

	for i = 1, 14, 1 do
		local assetId = KSIdGenerator.uuidInLowercaseHex()
		local docValue = WFCatAssetTestUtils.createAssetDocument( rxTest, wfCatalog, assetId )
		docValue._changeCounter = i
		docValues[ i ] = docValue
		docValueStream:onNext( docValue )
	end

	local delay = 0.1

	local maxBufferSize = 5

	local bufferedDocValueStream = WFCatMetadataPusher.makeDocValueAwareBufferWithTimeOrCountStream(
			docValueStream, delay, maxBufferSize )

			:tap( function( bufferedDocs )

				if bufferedDocs.continueFrom == 15 then

					-- After we receive the first 14 documents, schedule more on the stream.
					-- This mimics real world behavior.

					for j = 15, 20, 1 do
						local assetId = KSIdGenerator.uuidInLowercaseHex()
						local docValue = WFCatAssetTestUtils.createAssetDocument( rxTest, wfCatalog, assetId )
						docValue._changeCounter = j
						docValues[ j ] = docValue

						WFRxScheduler.sharedScheduler():schedule( function()
							docValueStream:onNext( docValue )
						end )

					end

				end

			end )

	local observer = rxTest:observe( bufferedDocValueStream )

	observer:expectDataWithinTime( 0.15, {
		docValues[ 1 ],
		docValues[ 2 ],
		docValues[ 3 ],
		docValues[ 4 ],
		docValues[ 5 ],
		continueFrom = 6
	} )

	observer:expectDataWithinTime( 0.15, {
		docValues[ 6 ],
		docValues[ 7 ],
		docValues[ 8 ],
		docValues[ 9 ],
		docValues[ 10 ],
		continueFrom = 11
	} )

	observer:expectDataWithinTime( 0.15, {
		docValues[ 11 ],
		docValues[ 12 ],
		docValues[ 13 ],
		docValues[ 14 ],
		continueFrom = 15
	} )

	observer:expectDataWithinTime( 0.15, {
		docValues[ 15 ],
		docValues[ 16 ],
		docValues[ 17 ],
		docValues[ 18 ],
		docValues[ 19 ],
		continueFrom = 20
	} )

	observer:expectDataWithinTime( 0.15, {
		docValues[ 20 ],
		continueFrom = 21
	} )

	observer:dispose()

end

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

function WFCatMetadataPusherTest.rxTestMakeDocValueAwareBufferWithTimeOrCountStream_DropStaleDocs( rxTest )

	local wfCatalog = WFCatMockCatalog.createFakeWFCatalog( rxTest )

	local docValueStream = Rx.ReplaySubject( 100 )

	local docValues = {}

	for i = 1, 5, 1 do
		local assetId = KSIdGenerator.uuidInLowercaseHex()
		local docValue = WFCatAssetTestUtils.createAssetDocument( rxTest, wfCatalog, assetId )
		rxTest:waitFor( wfCatalog:docStore():setDocValue( docValue ) )
		docValue._changeCounter = i
		docValues[ i ] = docValue
		docValueStream:onNext( docValue )

		docValueStream:onNext( docValue ) -- Emit duplicate docValues on the stream.
	end

	local delay = 0.1

	local maxBufferSize = 100

	local bufferedDocValueStream = WFCatMetadataPusher.makeDocValueAwareBufferWithTimeOrCountStream(
			docValueStream, delay, maxBufferSize )

	local observer = rxTest:observe( bufferedDocValueStream )

	local expectedResults = {
		false,
		docValues[ 1 ],
		false,
		docValues[ 2 ],
		false,
		docValues[ 3 ],
		false,
		docValues[ 4 ],
		false,
		docValues[ 5 ],
		continueFrom = 6
	}

	observer:expectDataWithinTime( 1, expectedResults )

	observer:dispose()

	-- Compare the expected results against the results from :changeDocs for consistency.

	local changedDocsExpectedResults = {}
	changedDocsExpectedResults.continueFrom = expectedResults.continueFrom

	for _, value in ipairs( expectedResults ) do

		if value then
			value._newRevId = nil
			changedDocsExpectedResults[ #changedDocsExpectedResults + 1 ] = value
		end

	end

	local function simpleRemap( docValues )

		for _, docValue in ipairs( docValues ) do

			docValue._localId = nil
			docValue._revId = nil

		end

		return docValues

	end

	rxTest:expectSingleValue( wfCatalog:docStore():changedDocs {
			fromChangeCounter = 0,
			includeDocumentsFromSource = WFRxIndexStoreSchema.sourceAllDocuments
		}
		:map( simpleRemap ), changedDocsExpectedResults )

end

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

function WFCatMetadataPusherTest.rxTestMakeDocValueAwareBufferWithTimeOrCountStream_WF1424( rxTest )

	-- This test verifies that the race condition is fixed within
	-- MakeDocValueAwareBufferWithTimeOrCountStream. The timer could
	-- incorreclty fire when the buffer was empty, causing an error.

	rxTest:setTimeLimit( 10 )

	local wfCatalog = WFCatMockCatalog.createFakeWFCatalog( rxTest )

	local docValueStream = Rx.ReplaySubject( 100 )

	local docValues = {}

	-- Create a stream that fires with 5 doc values.

	for i = 1, 5, 1 do
		local assetId = KSIdGenerator.uuidInLowercaseHex()
		local docValue = WFCatAssetTestUtils.createAssetDocument( rxTest, wfCatalog, assetId )
		docValue._changeCounter = i
		docValues[ i ] = docValue
		docValueStream:onNext( docValue )
	end

	local delay = 0.1

	local maxBufferSize = 5

	local hitError = false

	-- Create a buffered stream with a max size of 5.

	local bufferedDocValueStream = WFCatMetadataPusher.makeDocValueAwareBufferWithTimeOrCountStream(
			docValueStream, delay, maxBufferSize )

			:tapOnError( function()

				hitError = true

			end )

	-- When the first doc value is added to the buffer, a timer is created.
	-- The buffer should fill up and fire right away. When the buffer is
	-- full, the timer should be disposed; nothing should fire and no
	-- error should be thrown.

	local observer = rxTest:observe( bufferedDocValueStream )

	observer:expectDataWithinTime( 0.1, {
		docValues[ 1 ],
		docValues[ 2 ],
		docValues[ 3 ],
		docValues[ 4 ],
		docValues[ 5 ],
		continueFrom = 6
	} )

	-- Now sleep for the duration of the timer and make sure
	-- the error was not hit.

	rxTest:sleep( delay )

	rxTest:expectEqual( false, hitError )

	observer:dispose()

end

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

function WFCatMetadataPusherTest.rxTestSpacePusherIgnoresCatalogUpdates_mock( rxTest )

	rxTest:setTimeLimit( 5 )

	local spaceId = KSIdGenerator:uuidInLowercaseHex()
	local catalogId = KSIdGenerator:uuidInLowercaseHex()

	local spaceDocStoreMocks = generateDocStoreMocks( 'space', spaceId, { excludeAlbum = true } )
	local catalogDocStoreMocks = generateDocStoreMocks( 'catalog', catalogId, { excludeAlbum = true } )

	local reqMocks = {

		pusherMockReqResp( 'space', spaceId, spaceDocStoreMocks.pusherUpdates ),

	}

	local catalogSpec = WFCatSpecMock( catalogId, kUserId, rxTest, reqMocks )
	local catalog = WFCatMockCatalog.createWFCatalogForTestWithSpec( rxTest, catalogSpec )

	local docStore = catalog:docStore()

	local spaceoptions = KSTableUtils.deepCopy( options )
	spaceoptions.isSpacePusher = true

	local pusher = WFCatMetadataPusher( catalog, spaceoptions )
	rxTest:addCloseTarget( pusher )

	local updateCount = rxTest:observe( pusher:countUpdatesToSend() )
	updateCount:expectData( 0 )

	-- Create an asset document. This is a SPACE document.

	rxTest:waitFor( docStore:bulkSetDocValue( spaceDocStoreMocks.docStoreValues ) )

	-- Create an asset document. This is a CATALOG document.

	rxTest:waitFor( docStore:bulkSetDocValue( catalogDocStoreMocks.docStoreValues ) )

	-- There should only be one asset to push. The space asset.

	updateCount:expectDataWithinTime( 1, 1 )

	rxTest:waitFor( pusher:_DEBUG_sendNUpdates( 1 ) )

end

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

function WFCatMetadataPusherTest.rxTestCatalogPusherIgnoresSpaceUpdates_mock( rxTest )

	rxTest:setTimeLimit( 5 )

	local spaceId = KSIdGenerator.uuidInLowercaseHex()
	local catalogId = KSIdGenerator:uuidInLowercaseHex()

	local spaceDocStoreMocks = generateDocStoreMocks( 'space', spaceId, { excludeAlbum = true } )
	local catalogDocStoreMocks = generateDocStoreMocks( 'catalog', catalogId, { excludeAlbum = true } )

	local reqMocks = {

		pusherMockReqResp( 'catalog', catalogId, catalogDocStoreMocks.pusherUpdates ),

	}

	local catalogSpec = WFCatSpecMock( catalogId, kUserId, rxTest, reqMocks )
	local catalog = WFCatMockCatalog.createWFCatalogForTestWithSpec( rxTest, catalogSpec )

	local docStore = catalog:docStore()

	local pusher = WFCatMetadataPusher( catalog, options )
	rxTest:addCloseTarget( pusher )

	local updateCount = rxTest:observe( pusher:countUpdatesToSend() )
	updateCount:expectData( 0 )

	-- Create an asset document. This is a SPACE document.

	rxTest:waitFor( docStore:bulkSetDocValue( spaceDocStoreMocks.docStoreValues ) )

	-- Create an asset document. This is a CATALOG document.

	rxTest:waitFor( docStore:bulkSetDocValue( catalogDocStoreMocks.docStoreValues ) )

	-- There should only be one asset to push. The CATALOG asset.

	updateCount:expectDataWithinTime( 1, 1 )

	rxTest:waitFor( pusher:_DEBUG_sendNUpdates( 1 ) )

end

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

function WFCatMetadataPusherTest.rxTestFreemiumPusherBehavior_mock( rxTest )

    rxTest:setTimeLimit( 5 )
	
    local catalogId = KSIdGenerator:uuidInLowercaseHex()
	
    local catalogDocStoreMocks = generateDocStoreMocks( 'catalog', catalogId, {  } )

    local reqMocks = {
        pusherMockReqResp( 'catalog', catalogId, catalogDocStoreMocks.pusherUpdates ),
    }

    local catalogSpec = WFCatSpecMock( catalogId, kUserId, rxTest, reqMocks )
    local catalog = WFCatMockCatalog.createWFCatalogForTestWithSpec( rxTest, catalogSpec )

    local docStore = catalog:docStore()

    -- Create freemium pusher
    local freemiumOptions = KSTableUtils.deepCopy( options )
    freemiumOptions.isFreemiumPusher = true

    local pusher = WFCatMetadataPusher( catalog, freemiumOptions )
    rxTest:addCloseTarget( pusher )

    local updateCount = rxTest:observe( pusher:countUpdatesToSend() )
    updateCount:expectData( 0 )

    rxTest:waitFor( docStore:bulkSetDocValue( catalogDocStoreMocks.docStoreValues ) )

	-- 1 asset, 1 album, 1 album_asset
	updateCount:expectDataWithinTime( 1, 3 )

	rxTest:expectSingleValue( catalog:docStore():localDocValue( pusher._continueFromLocalDocValueKey ),
		{
			_id = "Oz-freemiumSendUpdatesContinueFrom",
			_empty = true,
		} )

	rxTest:waitFor( pusher:_DEBUG_sendNUpdates( 3 ) )
	
	updateCount:expectDataWithinTime( 1, 0 )

	rxTest:expectSingleValue( catalog:docStore():localDocValue( pusher._continueFromLocalDocValueKey ),
		{
			_id = "Oz-freemiumSendUpdatesContinueFrom",
			continueFrom = 4,
		} )
end

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

function WFCatMetadataPusherTest.rxTestCatalogPusherIgnoresSpaceUpdatesAndUpdatesChangeCounter_mock( rxTest )

	rxTest:setTimeLimit( 5 )

	local spaceId = KSIdGenerator.uuidInLowercaseHex()
	local catalogId = KSIdGenerator:uuidInLowercaseHex()

	local docStoreMocks = generateDocStoreMocks( 'space', spaceId, { excludeAlbum = true } )

	local catalogSpec = WFCatSpecMock( catalogId, kUserId, rxTest )
	local catalog = WFCatMockCatalog.createWFCatalogForTestWithSpec( rxTest, catalogSpec )

	local docStore = catalog:docStore()

	local pusher = WFCatMetadataPusher( catalog, options )
	rxTest:addCloseTarget( pusher )

	local updateCount = rxTest:observe( pusher:countUpdatesToSend() )
	updateCount:expectData( 0 )

	-- Create an asset document. This is a SPACE document.

	rxTest:waitFor( docStore:bulkSetDocValue( docStoreMocks.docStoreValues ) )

	-- There should only be no assets to push.

	updateCount:expectDataWithinTime( 1, 1 )

	rxTest:expectSingleValue( catalog:docStore():localDocValue( pusher._continueFromLocalDocValueKey ),
		{
			_id = "Oz-sendUpdatesContinueFrom",
			_empty = true,
		} )

	rxTest:waitFor( pusher:_DEBUG_sendNUpdates( 0 ) )

	updateCount:expectDataWithinTime( 1, 0 )

	rxTest:expectSingleValue( catalog:docStore():localDocValue( pusher._continueFromLocalDocValueKey ),
		{
			_id = "Oz-sendUpdatesContinueFrom",
			continueFrom = 2,
		} )

end

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

function WFCatMetadataPusherTest.rxTestSpacePusherIgnoresCatalogUpdatesAndUpdatesChangeCounter_mock( rxTest )

	rxTest:setTimeLimit( 5 )

	local catalogId = KSIdGenerator:uuidInLowercaseHex()

	local docStoreMocks = generateDocStoreMocks( 'catalog', catalogId, { excludeAlbum = true } )

	local catalogSpec = WFCatSpecMock( catalogId, kUserId, rxTest )
	local catalog = WFCatMockCatalog.createWFCatalogForTestWithSpec( rxTest, catalogSpec )

	local docStore = catalog:docStore()

	local spaceoptions = KSTableUtils.deepCopy( options )
	spaceoptions.isSpacePusher = true

	local pusher = WFCatMetadataPusher( catalog, spaceoptions )
	rxTest:addCloseTarget( pusher )

	local updateCount = rxTest:observe( pusher:countUpdatesToSend() )
	updateCount:expectData( 0 )

	-- Create an asset document. This is a CATALOG document.

	rxTest:waitFor( docStore:bulkSetDocValue( docStoreMocks.docStoreValues ) )

	-- There should only be no assets to push.

	updateCount:expectDataWithinTime( 1, 1 )

	rxTest:expectSingleValue( catalog:docStore():localDocValue( pusher._continueFromLocalDocValueKey ),
		{
			_id = "Oz-spaceSendUpdatesContinueFrom",
			_empty = true,
		} )

	rxTest:waitFor( pusher:_DEBUG_sendNUpdates( 0 ) )

	updateCount:expectDataWithinTime( 1, 0 )

	rxTest:expectSingleValue( catalog:docStore():localDocValue( pusher._continueFromLocalDocValueKey ),
		{
			_id = "Oz-spaceSendUpdatesContinueFrom",
			continueFrom = 2,
		} )

end

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

function WFCatMetadataPusherTest.rxTestSendUpdatesTaskCompleteOnConflict_mock( rxTest )

	-- The following test verifies that upon seing a conflict, the sendUpdatesTask
	-- will throw an exception so that we don't skip updates that need to
	-- be sent to Oz.

	rxTest:setTimeLimit( 5 )

	local catalogId = KSIdGenerator:uuidInLowercaseHex()

	local docStoreMocks = generateDocStoreMocks( 'catalog', catalogId, { excludeAlbum = true } )
	local docStoreMocks2 = generateDocStoreMocks( 'catalog', catalogId, { excludeAlbum = true, customDescription = KSIdGenerator.uuidInLowercaseHex() } )
	local docStoreMocks3 = generateDocStoreMocks( 'catalog', catalogId, { excludeAlbum = true, customDescription = KSIdGenerator.uuidInLowercaseHex() } )

	local reqMocks = {

		pusherMockReqResp( 'catalog', catalogId, docStoreMocks.pusherUpdates ),

	}

	local catalogSpec = WFCatSpecMock( catalogId, kUserId, rxTest, reqMocks )
	local catalog = WFCatMockCatalog.createWFCatalogForTestWithSpec( rxTest, catalogSpec )

	local docStore = catalog:docStore()

	local pusher = WFCatMetadataPusher( catalog, options )
	rxTest:addCloseTarget( pusher )

	-- Create first asset document.

	rxTest:waitFor( docStore:setDocValue( docStoreMocks.docStoreValues[1] ) )

	-- Create second asset document.

	rxTest:waitFor( docStore:setDocValue( docStoreMocks2.docStoreValues[1] ) )

	-- Create a conflicting revision.

	local conflictingRevId = KSIdGenerator.uuidInLowercaseHex()
	docStoreMocks2.docStoreValues[1]._revIds = { conflictingRevId }

	local asset2RevId2 = rxTest:getFirstValue( docStore:setDocValue( docStoreMocks2.docStoreValues[1] ) )

	docStoreMocks2.pusherUpdates[1].revision_id = asset2RevId2

	catalogSpec:expectHttpRequest( pusherMockReqResp( 'catalog', catalogId, { docStoreMocks3.pusherUpdates[1], docStoreMocks2.pusherUpdates[1] } ) )

	-- Verify that asset document has two head revisions in conflict.

	rxTest:expectSingleValue(
		catalog:docStore():currentRevisionsForDoc( docStoreMocks2.assetDocId )
			:map( function( revisions )
				return #revisions == 2
					and revisions[ 1 ]._conflict == true
					and revisions[ 2 ]._conflict == true
			 end ),
		true
	)

	-- Create third asset document.

	rxTest:waitFor( docStore:setDocValue( docStoreMocks3.docStoreValues[1] ) )

	-- Attempt to send updates to Oz.

	rxTest:waitFor( pusher:_DEBUG_sendUpdatesOnce() )

	-- Verify the pusher's change counter is set to the document's sequence
	-- number containing conflict.

	local expectedContinueFrom = rxTest:getFirstValue(

		docStore:changedDocs { fromChangeCounter = 0 }
			:map( function( changedDocs )

				for _, docValue in ipairs( changedDocs ) do

					if docValue._id == docStoreMocks2.assetDocId then
						return docValue._changeCounter
					end

				end

			end ) )

	rxTest:expectSingleValue( docStore:localDocValue( 'Oz-sendUpdatesContinueFrom' ):pluck( 'continueFrom' ), expectedContinueFrom )

	-- Now attach conflict resolver and wait for the conflict to resolve.

	local conflictResolver = WFCatConflictResolver( docStore, catalog:indexStore() )
	rxTest:addCloseTarget( conflictResolver )

	rxTest:waitFor( docStore:ongoingFieldValue( docStoreMocks2.assetDocId, '_conflict' )
		:filter( function( isInConflict )
			return not isInConflict
		end )
		:take( 1 ) )

	-- We should see the update conflict resolve, and the updates flow to Oz.

	rxTest:waitFor( pusher:_DEBUG_sendNUpdates( 2 ) )

	-- Verify the new change counter.

	expectedContinueFrom = rxTest:getFirstValue(

		docStore:changedDocs { fromChangeCounter = 0 }
			:map( function( changedDocs )

				local largestChangeCounter = 0

				for _, docValue in ipairs( changedDocs ) do

					if largestChangeCounter < docValue._changeCounter then
						largestChangeCounter = docValue._changeCounter
					end

				end

				return largestChangeCounter + 1

			end ) )

	rxTest:expectSingleValue( docStore:localDocValue( 'Oz-sendUpdatesContinueFrom' ):pluck( 'continueFrom' ), expectedContinueFrom )

end

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

function WFCatMetadataPusherTest.rxTestSendUpdatesTaskCompleteOnConflict_testNightly( rxTest )

	rxTest:setTimeLimit( 60 )

	local catalogTestSession = WFCatTestSession( rxTest )

	local catalog

	rxTest:expectSingleValue(
		catalogTestSession:catalog()
			:tap( function( _catalog )
				catalog = _catalog
			end ),
		':ignore:' )

	local continuousTaskDelays = {
		progressiveBackoff = false,
		noWorkDelay = 1,
		maxNoWorkDelay = 1,
		errorDelay = 5,
		maxDelay = 1,
	}

	local options = {
		continuousTaskDelays = continuousTaskDelays,
		isCatalogPuller = true
	}

	local puller = WFCatMetadataPuller( catalog, "Test Puller", options )
	local pusher = WFCatMetadataPusher( catalog, options )

	rxTest:addCloseTarget( puller )
	rxTest:addCloseTarget( pusher )

	-- Create an asset

	local assetId = KSIdGenerator.uuidInLowercaseHex()
	local docValue = WFCatAssetTestUtils.createAssetDocument( rxTest, catalog, assetId )
	docValue.importSource = {
		fileName = '/path/',
		importTimestamp = KSDate.timeInIso8601Format(),
		importedBy = KSIdGenerator.uuidInLowercaseHex(),
	}
	docValue.develop = {
		userUpdated = KSDate.timeInIso8601Format(),
		processingModel = 'lightroom',

	}
	docValue._serverProvidedMetadata = nil
	rxTest:waitFor( catalog:docStore():setDocValue( docValue ) )

	-- Create a second asset

	local assetId2 = KSIdGenerator.uuidInLowercaseHex()
	local docValue2 = WFCatAssetTestUtils.createAssetDocument( rxTest, catalog, assetId2 )
	docValue2._serverProvidedMetadata = nil
	docValue2.importSource = {
		fileName = '/path/',
		importTimestamp = KSDate.timeInIso8601Format(),
		importedBy = KSIdGenerator.uuidInLowercaseHex(),
	}
	docValue2.develop = {
		userUpdated = KSDate.timeInIso8601Format(),
		processingModel = 'lightroom',

	}
	rxTest:waitFor( catalog:docStore():setDocValue( docValue2 ) )

	-- Create a conflicting revision of asset 2

	local conflictingRevId = KSIdGenerator.uuidInLowercaseHex()

	local conflicintgDocValue2 = WFCatAssetTestUtils.createAssetDocument( rxTest, catalog, assetId2 )
	conflicintgDocValue2._revIds = { conflictingRevId }
	conflicintgDocValue2._serverProvidedMetadata = nil
	conflicintgDocValue2.importSource = {
		fileName = '/path/',
		importTimestamp = KSDate.timeInIso8601Format(),
		importedBy = KSIdGenerator.uuidInLowercaseHex(),
	}
	conflicintgDocValue2.develop = {
		userUpdated = KSDate.timeInIso8601Format(),
		processingModel = 'lightroom',

	}
	rxTest:waitFor( catalog:docStore():setDocValue( conflicintgDocValue2 ) )

	-- Verify that asset document has two head revisions in conflict.

	rxTest:expectSingleValue(
		 catalog:docStore():currentRevisionsForDoc( assetId2 )
		 :map( function( revisions )
		 	return #revisions == 2
				and revisions[ 1 ]._conflict == true
				and revisions[ 2 ]._conflict == true
		 end ), true )

	-- Create a third asset

	local assetId3 = KSIdGenerator.uuidInLowercaseHex()
	local docValue3 = WFCatAssetTestUtils.createAssetDocument( rxTest, catalog, assetId3 )
	docValue3._serverProvidedMetadata = nil
	docValue3.importSource = {
		fileName = '/path/',
		importTimestamp = KSDate.timeInIso8601Format(),
		importedBy = KSIdGenerator.uuidInLowercaseHex(),
	}
	docValue3.develop = {
		userUpdated = KSDate.timeInIso8601Format(),
		processingModel = 'lightroom',

	}
	rxTest:waitFor( catalog:docStore():setDocValue( docValue3 ) )

	-- Figure out the sequence number of assetId 2. We use this later.

	local asset2ChangeCounter = rxTest:getFirstValue(

		catalog:docStore():changedDocs { fromChangeCounter = 0 }
			:map( function( changedDocs )


				for _, docValue in ipairs( changedDocs ) do

					if docValue._id == assetId2 then
						return docValue._changeCounter
					end

				end

			end ) )

	puller:getUpdatesContinuously()
	rxTest:waitFor( pusher:sendUpdatesContinuously() )
	
	rxTest:sleep( 5 ) -- Wait a few seconds for us to attempt to send all assets to Oz.

	-- Verify asset 1 is on Oz.

	local asset1OnOz = rxTest:observe(
							catalog:docStore():ongoingDocValue( assetId )
								:map( function( docValue )
									return docValue._serverProvidedMetadata ~= nil
								end ) )
	asset1OnOz:expectDataWithinTime( 30, true )
	asset1OnOz:dispose()

	-- Asset 2 and Asset 3 Should *not* be on Oz.

	rxTest:expectSingleValue( catalog:docStore():docValue( assetId2 )
								:map( function( docValue )
									return docValue._serverProvidedMetadata ~= nil
								end ), false )

	rxTest:expectSingleValue( catalog:docStore():docValue( assetId3 )
								:map( function( docValue )
									return docValue._serverProvidedMetadata ~= nil
								end ), false )

	-- Verify the pusher's change counter

	rxTest:expectSingleValue( catalog:docStore():localDocValue( 'Oz-sendUpdatesContinueFrom' ):pluck( 'continueFrom' ), asset2ChangeCounter )

	-- Now attach conflict resolver.

	local conflictResolver = WFCatConflictResolver( catalog:docStore(), catalog:indexStore() )
	rxTest:addCloseTarget( conflictResolver )

	-- Verify that all assets end up on Oz.

	local asset2OnOz = rxTest:observe(
							catalog:docStore():ongoingDocValue( assetId2 )
								:map( function( docValue )
									return docValue._serverProvidedMetadata ~= nil
								end ) )
	asset2OnOz:expectDataWithinTime( 30, true )
	asset2OnOz:dispose()

	local asset3OnOz = rxTest:observe(
							catalog:docStore():ongoingDocValue( assetId3 )
								:map( function( docValue )
									return docValue._serverProvidedMetadata ~= nil
								end ) )
	asset3OnOz:expectDataWithinTime( 30, true )
	asset3OnOz:dispose()

	puller:stopContinuousUpdates()
	pusher:stopContinuousUpdates()

end

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

function WFCatMetadataPusherTest.rxTestWF2107BugDemonstration( rxTest )

	-- The following PR verifies the fix that went in in commit 4ef1bf17e.
	-- Before the fix, this test would fail on the call to:
	--
	-- rxTest:expectEqual( 0, pusher:DEBUG_getPausedBySynchronizedOperation() )
	--
	-- Even though the observable was not subscribed to, the pusher was being paused.
	-- This is the same basic pattern in the CT functions, such as WFCatTrash.restoreTrashAssets.
	-- Because of this, it's theoretically possible that when some CT functions are called
	-- but the return observables are not subscribed to, the pusher would end up
	-- permanently paused, as the finally function that unpauses the pusher
	-- is never registered.

	local catalogId = KSIdGenerator:uuidInLowercaseHex()
	local catalogSpec = WFCatSpecMock( catalogId, kUserId, rxTest,
		{
			-- should be no HTTP comm if no changes to send
		} )

	local catalog = WFCatMockCatalog.createWFCatalogForTestWithSpec( rxTest, catalogSpec )

	local pusher = WFCatMetadataPusher( catalog, options )
	rxTest:addCloseTarget( pusher )

	local pusherControl = {
		pauseReturningCounter = function( ... )
			return pusher:pauseReturningCounter( ... )
		end,
		resume = function()
			pusher:resume()
		end,
	}

	local obs = Rx.Observable.combineLatest(
		pusherControl.pauseReturningCounter(),
		Rx.Observable.just( true )
	)
	:takeLast( 0 )
	:finally( pusherControl.resume )

	rxTest:expectEqual( 0, pusher:DEBUG_getPausedBySynchronizedOperation() )

	rxTest:waitFor( obs )

end

--==============================================================================
-- @mark Do not send update revisions multiple times
--==============================================================================

local function testDoNotPushSameRevisionIdTwice( rxTest, type, catalogOrSpaceId, catalogId )

	-- WF-2295 After pushing an update to Oz, getting an update for the given
	-- asset that does NOT contain the revisionId should NOT trigger sending
	-- the exact same payload with the same revisionId a second time. This
	-- can happen with atomic develop settings, where a /rels/temp/... link
	-- could come back while the pushed update is still pending the upload
	-- of other binaries before it can be accepted by Oz

	rxTest:setTimeLimit( 5 )

	local docStoreMocks = generateDocStoreMocks( type, catalogOrSpaceId, { excludeAlbum = true, includeAssetUpdate = true } )

	local assetDocId = docStoreMocks.assetDocId

	local reqMocks = {

		pusherMockReqResp( type, catalogOrSpaceId, docStoreMocks.pusherUpdates ),
		prefetchReqResp( catalogId ),
		catalogPrefetchReqResp( catalogId ),
		pullerMockReqResp( type, catalogOrSpaceId, docStoreMocks.pullerUpdates ),
		pusherMockReqResp( type, catalogOrSpaceId, { docStoreMocks.assetUpdateMock } ),

	}

	-- remove the prefetch calls for albums and catalog for space tests

	if type == 'space' then

		table.remove( reqMocks, 3 )
		table.remove( reqMocks, 2 )

	end

	local catalogSpec = WFCatSpecMock( catalogId, kUserId, rxTest, reqMocks )
	local catalog = WFCatMockCatalog.createWFCatalogForTestWithSpec( rxTest, catalogSpec )

	local docStore = catalog:docStore()

	local pusherPullerOptions = options

	if type == 'space' then

		pusherPullerOptions = KSTableUtils.deepCopy( options )
		pusherPullerOptions.spaceId = catalogOrSpaceId
		pusherPullerOptions.updatesFeedUrl = catalog:ozCatalogSpec():spaceUpdatesUrlForSpaceId( catalogOrSpaceId )
		pusherPullerOptions.isCatalogPuller = false
		pusherPullerOptions.isSpacePusher = true

	end

	local pusher = WFCatMetadataPusher( catalog, pusherPullerOptions )
	rxTest:addCloseTarget( pusher )

	local revisionsAlreadySent = pusher:DEBUG_revisionsAlreadySent()
		-- this is the in memory table used to track what revisions have been sent
		-- to Oz to avoid sending multiple times unnecessarily.

	local puller = WFCatMetadataPuller( catalog, "Test Puller", pusherPullerOptions )
	rxTest:addCloseTarget( puller )

	-- Create an asset, push it to Oz and pull server metadata from Oz

	rxTest:waitFor( docStore:bulkSetDocValue( docStoreMocks.docStoreValues ) )

	rxTest:waitFor( pusher:_DEBUG_sendNUpdates( 1 ) )

	rxTest:expectEqual( nil, revisionsAlreadySent[ assetDocId ] )
		-- this was a create request sent to Oz, so nothing should be stored to table yet

	rxTest:expectOnlyCompleted( puller:getUpdatesOnce() )

	rxTest:expectSingleValue(
		docStore:changedDocs { fromChangeCounter = 0 }
			:map( remapChangedDocsFeedForMockTests ),
		docStoreMocks.expectedDocStoreValues
	)

	-- now that asset has synced back from oz, puller should send no updates and
	-- there should still be no entry in table

	rxTest:waitFor( pusher:_DEBUG_sendNUpdates( 0 ) )

	rxTest:expectEqual( nil, revisionsAlreadySent[ assetDocId ] )

	-- Make a local change to the asset, push it to Oz and pull update from Oz with temp link

	rxTest:waitFor( docStore:applyOverlay( docStoreMocks.assetUpdateOverlay ) )

	rxTest:waitFor( pusher:_DEBUG_sendNUpdates( 1 ) )

	local entryForRevision = revisionsAlreadySent[ assetDocId ]
	rxTest:expectEqual( docStoreMocks.assetExpectedValue._revId, entryForRevision.revisionId )
	rxTest:expectEqual( docStoreMocks.assetRevisionId, entryForRevision.parentRevisionId )
	rxTest:expectEqual( nil, entryForRevision.localOnly )

	local tempLinkKey = '/rels/temp/xmp/develop/65a5c86c4aab2bd9fc60b0c336aeb68e664f1a084f392644cf3eef58de15d80d'

	docStoreMocks.pullerUpdates[1].links[tempLinkKey] = { exists = true }

	local includeUpdatedSinceParam = true
	catalogSpec:expectHttpRequest( pullerMockReqResp( type, catalogOrSpaceId, docStoreMocks.pullerUpdates, includeUpdatedSinceParam ) )

	rxTest:expectOnlyCompleted( puller:getUpdatesOnce() )

	local tempLinkUpdates = { docStoreMocks.assetExpectedValue }
	tempLinkUpdates[1]._conflict = true
	tempLinkUpdates[1]._serverProvidedMetadata.links[tempLinkKey] = { exists = true }
	tempLinkUpdates.continueFrom = 6

	if type == 'space' then
		tempLinkUpdates.continueFrom = tempLinkUpdates.continueFrom - 1
	end

	rxTest:expectSingleValue(
		docStore:changedDocs { fromChangeCounter = 4 }
			:map( remapChangedDocsFeedForMockTests ),
		tempLinkUpdates
	)

	-- no update will be sent because of the conflict

	rxTest:waitFor( pusher:_DEBUG_sendNUpdates( 0 ) )

	rxTest:expectEqual( entryForRevision, revisionsAlreadySent[ assetDocId ] )
		-- entry tracking sent revision should be unchanged

	-- Now resolve the conflict by deleting the version with the older revision id synced from Oz.

	rxTest:waitFor( docStore:deleteDoc( assetDocId, docStoreMocks.assetRevisionId ) )

	-- now STILL no updates should be sent because we will not send same payload with same revision id a second time

	rxTest:waitFor( pusher:_DEBUG_sendNUpdates( 0 ) )

	rxTest:expectEqual( entryForRevision, revisionsAlreadySent[ assetDocId ] )
		-- entry tracking sent revision should be unchanged

	-- add a _localOnly section to  document without changing revision id
	-- this SHOULD push a revision to Oz

	rxTest:waitFor( docStore:applyOverlay { _id = assetDocId, _localOnly = { foo = 'bar' } } )

	catalogSpec:expectHttpRequest( pusherMockReqResp( type, catalogOrSpaceId, { docStoreMocks.assetUpdateMock } ) )

	rxTest:waitFor( pusher:_DEBUG_sendNUpdates( 1 ) )

	entryForRevision = revisionsAlreadySent[ assetDocId ]
	rxTest:expectEqual( docStoreMocks.assetExpectedValue._revId, entryForRevision.revisionId )
	rxTest:expectEqual( docStoreMocks.assetRevisionId, entryForRevision.parentRevisionId )
	rxTest:expectEqual( 'bar', entryForRevision.localOnly.foo )

	-- modify the _localOnly section to document without changing revision id
	-- this SHOULD push a revision to Oz

	rxTest:waitFor( docStore:applyOverlay { _id = assetDocId, _localOnly = { bar = 'foo' } } )

	catalogSpec:expectHttpRequest( pusherMockReqResp( type, catalogOrSpaceId, { docStoreMocks.assetUpdateMock } ) )

	rxTest:waitFor( pusher:_DEBUG_sendNUpdates( 1 ) )

	entryForRevision = revisionsAlreadySent[ assetDocId ]
	rxTest:expectEqual( docStoreMocks.assetExpectedValue._revId, entryForRevision.revisionId )
	rxTest:expectEqual( docStoreMocks.assetRevisionId, entryForRevision.parentRevisionId )
	rxTest:expectEqual( 'bar', entryForRevision.localOnly.foo )
	rxTest:expectEqual( 'foo', entryForRevision.localOnly.bar )

	-- remove the _localOnly section from document without changing revision id
	-- this SHOULD push a revision to Oz

	rxTest:waitFor( docStore:applyOverlay { _id = assetDocId, _localOnly = { _delete = true } } )

	catalogSpec:expectHttpRequest( pusherMockReqResp( type, catalogOrSpaceId, { docStoreMocks.assetUpdateMock } ) )

	rxTest:waitFor( pusher:_DEBUG_sendNUpdates( 1 ) )

	entryForRevision = revisionsAlreadySent[ assetDocId ]
	rxTest:expectEqual( docStoreMocks.assetExpectedValue._revId, entryForRevision.revisionId )
	rxTest:expectEqual( docStoreMocks.assetRevisionId, entryForRevision.parentRevisionId )
	rxTest:expectEqual( nil, entryForRevision.localOnly )

	-- edge case 1: revision id has not changed but payload has changed (should be
	-- impossible, we simulate by hacking the revisionsAlreadySent table)

	local newRevisionId = KSIdGenerator.uuidInLowercaseHex()

	rxTest:waitFor( catalog:docStore():applyOverlay {
		_id = assetDocId,
		_newRevId = newRevisionId,
		xmp = { dc = { description = "The new description" } },
	} )

	revisionsAlreadySent[ assetDocId ].revisionId = newRevisionId

	local parentRevisionId = docStoreMocks.assetUpdateMock.revision_id
	docStoreMocks.assetUpdateMock.revision_id = newRevisionId
	docStoreMocks.assetUpdateMock.payload.xmp.dc.description = "The new description"

	catalogSpec:expectHttpRequest( pusherMockReqResp( type, catalogOrSpaceId, { docStoreMocks.assetUpdateMock } ) )

	rxTest:waitFor( pusher:_DEBUG_sendNUpdates( 1 ) )

	entryForRevision = revisionsAlreadySent[ assetDocId ]
	rxTest:expectEqual( newRevisionId, entryForRevision.revisionId )
	rxTest:expectEqual( docStoreMocks.assetRevisionId, entryForRevision.parentRevisionId )
	rxTest:expectEqual( "The new description", entryForRevision.payload.xmp.dc.description )

	-- edge case 2: Most recent revision id on Oz does not match the expected value
	-- again this should trigger conflict resolution which would chagne the revision
	-- id so have to hack it

	newRevisionId = KSIdGenerator.uuidInLowercaseHex()
	parentRevisionId = KSIdGenerator.uuidInLowercaseHex()

	rxTest:waitFor( catalog:docStore():applyOverlay {
		_id = assetDocId,
		_newRevId = newRevisionId,
		xmp = { dc = { description = "The new description" } },
	} )

	revisionsAlreadySent[ assetDocId ].revisionId = newRevisionId
	revisionsAlreadySent[ assetDocId ].parentRevisionId = parentRevisionId

	docStoreMocks.assetUpdateMock.revision_id = newRevisionId

	catalogSpec:expectHttpRequest( pusherMockReqResp( type, catalogOrSpaceId, { docStoreMocks.assetUpdateMock } ) )

	rxTest:waitFor( pusher:_DEBUG_sendNUpdates( 1 ) )

	entryForRevision = revisionsAlreadySent[ assetDocId ]
	rxTest:expectEqual( newRevisionId, entryForRevision.revisionId )
	rxTest:expectEqual( docStoreMocks.assetRevisionId, entryForRevision.parentRevisionId )

	-- edge case 3: Cannot match expected parent revision id because Oz did not
	-- return revision_ids array... this will never happen!

	-- pull change with no revisions from Oz

	docStoreMocks.pullerUpdates[1].revision_ids = nil

	catalogSpec:expectHttpRequest( pullerMockReqResp( type, catalogOrSpaceId, docStoreMocks.pullerUpdates, includeUpdatedSinceParam ) )

	rxTest:expectOnlyCompleted( puller:getUpdatesOnce() )

	-- apply revision

	local prevRevisionId = newRevisionId
	newRevisionId = KSIdGenerator.uuidInLowercaseHex()

	rxTest:waitFor( catalog:docStore():applyOverlay {
		_id = assetDocId,
		_newRevId = newRevisionId,
		xmp = { dc = { description = "The new description" } },
	} )

	revisionsAlreadySent[ assetDocId ].revisionId = newRevisionId

	docStoreMocks.assetUpdateMock.revision_id = newRevisionId
	docStoreMocks.assetUpdateMock.if_match = nil

	-- resolve the conflict by deleting the version with the older revision id synced from Oz.

	rxTest:waitFor( docStore:deleteDoc( assetDocId, prevRevisionId ) )

	-- expect update to be pushed to Oz

	catalogSpec:expectHttpRequest( pusherMockReqResp( type, catalogOrSpaceId, { docStoreMocks.assetUpdateMock } ) )

	rxTest:waitFor( pusher:_DEBUG_sendNUpdates( 1 ) )

	entryForRevision = revisionsAlreadySent[ assetDocId ]
	rxTest:expectEqual( newRevisionId, entryForRevision.revisionId )
	rxTest:expectEqual( nil, entryForRevision.parentRevisionId )

	-- edge case 4: Cannot match expected parent revision id because Oz returned
	-- empty revision_ids array... this will never happen!

	-- pull change with no revisions from Oz

	docStoreMocks.pullerUpdates[1].revision_ids = { }

	catalogSpec:expectHttpRequest( pullerMockReqResp( type, catalogOrSpaceId, docStoreMocks.pullerUpdates, includeUpdatedSinceParam ) )

	rxTest:expectOnlyCompleted( puller:getUpdatesOnce() )

	-- apply revision

	prevRevisionId = newRevisionId
	newRevisionId = KSIdGenerator.uuidInLowercaseHex()

	rxTest:waitFor( catalog:docStore():applyOverlay {
		_id = assetDocId,
		_newRevId = newRevisionId,
		xmp = { dc = { description = "The new description" } },
	} )

	revisionsAlreadySent[ assetDocId ].revisionId = newRevisionId
	revisionsAlreadySent[ assetDocId ].parentRevisionId = parentRevisionId

	docStoreMocks.assetUpdateMock.revision_id = newRevisionId

	-- resolve the conflict by deleting the version with the older revision id synced from Oz.

	rxTest:waitFor( docStore:deleteDoc( assetDocId, prevRevisionId ) )

	-- expect update to be pushed to Oz

	catalogSpec:expectHttpRequest( pusherMockReqResp( type, catalogOrSpaceId, { docStoreMocks.assetUpdateMock } ) )

	rxTest:waitFor( pusher:_DEBUG_sendNUpdates( 1 ) )

	entryForRevision = revisionsAlreadySent[ assetDocId ]
	rxTest:expectEqual( newRevisionId, entryForRevision.revisionId )
	rxTest:expectEqual( nil, entryForRevision.parentRevisionId )

	-- after syncing the update back from Oz with the current revision stored locally
	-- in Oz's list of revision_ids, the puller should clear the entry from the table
	-- next time it checks the updated document and decides NOT to send it.

	docStoreMocks.pullerUpdates[1].revision_ids = { docStoreMocks.assetRevisionId, newRevisionId, }

	catalogSpec:expectHttpRequest( pullerMockReqResp( type, catalogOrSpaceId, docStoreMocks.pullerUpdates, includeUpdatedSinceParam ) )

	rxTest:expectOnlyCompleted( puller:getUpdatesOnce() )

	rxTest:waitFor( pusher:_DEBUG_sendNUpdates( 0 ) )
		-- one for conflict resolution

	rxTest:expectEqual( nil, revisionsAlreadySent[ assetDocId ] )

end

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

function WFCatMetadataPusherTest.rxTestDoNotPushSameRevisionIdTwiceForCatalogs( rxTest )

	local catalogId = KSIdGenerator:uuidInLowercaseHex()

	testDoNotPushSameRevisionIdTwice( rxTest, 'catalog', catalogId, catalogId )

end

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

function WFCatMetadataPusherTest.rxTestDoNotPushSameRevisionIdTwiceForSpaces( rxTest )

	local spaceId = KSIdGenerator.uuidInLowercaseHex()
	local catalogId = KSIdGenerator:uuidInLowercaseHex()

	testDoNotPushSameRevisionIdTwice( rxTest, 'space', spaceId, catalogId )

end

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

-- FIXME [jandrew 2025-03-11]: Disabled due to instability.

function WFCatMetadataPusherTest.DISABLED_rxTestDoNotPushSameRevisionIdTwice_BinaryMissingError_testNightly( rxTest )

	-- WF-2295 see comment in rxTestDoNotPushSameRevisionIdTwice

	-- WF-2228 also add test for handling of BinaryMissingError to this test even
	-- though it is technically testing code in WFCatErrorHandler, this test sets
	-- up everything to test for retrying on that perfectly so putting it here.

	rxTest:setTimeLimit( 360 )

	local catalogTestSession = WFCatTestSession( rxTest )
	local catalog = rxTest:getFirstValue( catalogTestSession:catalog() )

	local puller = WFCatMetadataPuller( catalog, "Test Puller", options )
	rxTest:addCloseTarget( puller )

	-- do a manual pull first to ensure the catalog doc gets localId 1
	rxTest:expectOnlyCompleted( puller:getUpdatesOnce() )

	puller:getUpdatesContinuously()

	local pusherOptions = KSTableUtils.merge( {}, options, {
			capabilities = {
				[ 'WFCatBinaryUploader.atomicDevelopSettings' ] = true,
			}
		} )

	local pusher = WFCatMetadataPusher( catalog, pusherOptions )
	rxTest:addCloseTarget( pusher )

	pusher:addRequestHeader( 'X-Quicker-Retries', 'true' )
		-- need this to be able to see expected error documents sync

	-- need a binary uploader since easiest way to repro this today is with atomic develop settings upload

	local uploaderTaskDelays = {

		progressiveBackoff = false,
		noWorkDelay = 1,
		maxNoWorkDelay = 1,
		errorDelay = 1,
		maxDelay = 1,

	}

	local uploaderOpts = {

		clearFromRecentlyUploadedSeconds = 1,
		waitForOtherUploadSeconds = .5,
		waitForCopyFailedSeconds = .5,
		timer = function( seconds )
			return Rx.Observable.timer( seconds, WFRxScheduler.sharedScheduler() )
		end,
		capabilities = {
			[ 'WFCatBinaryUploader.atomicDevelopSettings' ] = true,
		},

	}

	local uploader = WFCatBinaryUploader( catalog, nil, uploaderTaskDelays, nil, uploaderOpts )
	rxTest:addCloseTarget( uploader )

	uploader:startContinuousUploads()

	-- need a conflict resolver too as when the temp link downloads there will be a conflict
	-- and second update with same revision id would not be attempted until after conflict
	-- is resolved in any case.

	local conflictResolver = WFCatConflictResolver( catalog:docStore(), catalog:indexStore() )
	rxTest:addCloseTarget( conflictResolver )

	-- and need an error handler to test BinaryMissingError handling

	local errorHandler = WFCatErrorHandler( catalog, { }, { debugMode = true }  )
	rxTest:addCloseTarget( errorHandler )

	-- create an asset with an original

	local assetId = KSIdGenerator.uuidInLowercaseHex()
	local revisionId = KSIdGenerator.uuidInLowercaseHex()

	local testDataPath = KSPathUtils.child( KSPathUtils.getStandardFilePath( 'testdata' ), 'import-samples' )
	local testDataFolder1 = KSPathUtils.child( testDataPath, "folder-1" )
	local originalFilePath = KSPathUtils.child( testDataFolder1, "small.JPG" )
	local sha256 = WFCrypto.digestFile( originalFilePath, 'SHA256' )

	local assetDoc = WFCatAssetTestUtils.createBaseAssetDocument( catalog:ozCatalogId(), assetId, revisionId, sha256, originalFilePath )
	assetDoc = WFCatAssetTestUtils.addOriginalPathToDocument( assetDoc, sha256, originalFilePath )

	rxTest:waitFor( catalog:docStore():setDocValue( assetDoc ) )

	-- initial asset upload to Oz

	local sendNUpdates = rxTest:observe( pusher:_DEBUG_sendNUpdates( 1 ) )
	sendNUpdates:expectCompletedWithinTime( 30 )
	sendNUpdates:dispose()

	-- wait for asset initial sync to Oz and back to WF

	puller:getUpdatesContinuously()

	local observeAsset = observeHasServerProvidedMetadata( rxTest, catalog, assetId )
	observeAsset:expectDataWithinTime( 30, true )
	observeAsset:dispose()

	-- wait for original to upload to oz

	observeAsset = rxTest:observe(
		catalog:docStore():ongoingDocValue( assetId )
			:map( WFOzDocUtils.isValidOriginalAvailableOnOz )
		)
	observeAsset:expectDataWithinTime(360, true)
	observeAsset:dispose()

	-- add a mask with local file missing so will not upload

	local auxId = KSIdGenerator.uuidInLowercaseHex()
	local auxDigest = string.upper( WFCrypto.MD5.digest( KSIdGenerator.uuidInLowercaseHex() ) )
	local auxType = "mask"

	-- This should be close to random
	local auxBinary = KSData.make( auxDigest )

	local baseAuxPath = WFCatAssetAux.getBasePath( catalog )
	local auxFilePath, auxRelativeFilePath = WFCatAssetAux.calculatePathUsingDigest( baseAuxPath, auxDigest )

	local tempAuxFilePath = auxFilePath .. '_temp'
		-- we write aux file initially to a different location so that it will fail to be uploaded

	local auxParentPath = KSPathUtils.parent( auxFilePath )
	WFFileUtils.mkdir( auxParentPath )
	KSData.saveToFile( auxBinary, tempAuxFilePath )

	local auxFileSize = WFFileUtils.fileAttributes( tempAuxFilePath ).fileSize
	local auxSha256 = WFCrypto.digestFile( tempAuxFilePath, 'SHA256' )

	local auxParamsList = {{ id = auxId, sha256 = auxSha256, filePath = auxRelativeFilePath, fileSize = auxFileSize, digest = auxDigest, type = auxType }}
	local auxOverlay = { _id = assetId }
	auxOverlay = WFCatAssetTestUtils.addAuxiliaryMetadataReference( auxOverlay, assetId, auxParamsList )

	rxTest:waitFor( catalog:docStore():applyOverlay( auxOverlay ) )

	-- add external xmp develop

	local carouselDocuments = KSPathUtils.getStandardFilePath( 'carouselDocuments' )
	local xmpDevelopFilePath = KSPathUtils.child( carouselDocuments, 'small.XMP' )
	local _, xmpSha256 =  WFCatAssetTestUtils.addXmpDevelopToDocument( rxTest, catalog, assetId, xmpDevelopFilePath )

	-- update should be sent to Oz just one time

	sendNUpdates = rxTest:observe( pusher:_DEBUG_sendNUpdates( 1 ) )
	sendNUpdates:expectCompletedWithinTime( 30 )
	sendNUpdates:dispose()

	-- wait for /rels/temp/xmp/develop/{sha256} link to sync back from Oz

	local tempLinkKey = '/rels/temp/xmp/develop/' .. xmpSha256

	observeAsset = rxTest:observe(
		catalog:docStore():ongoingDocValue( assetId )
			:map( function( docValue )
				revisionId = docValue._revId
				return KSTableUtils.derefOpt( docValue, '_serverProvidedMetadata', 'links', tempLinkKey, 'exists' )
			end )
		)
	observeAsset:expectDataWithinTime( 30, true )
	observeAsset:dispose()

	-- wait for conflict resolution to be complete. revision id should not change.

	local latestRevisionId

	observeAsset = rxTest:observe (
		Rx.Observable.timer( 5, 3, WFRxScheduler.sharedScheduler() )
			:flatMap( function()
				return catalog:docStore():currentRevisionsForDoc( assetId )
			end )
			:map( function( revisions )
				latestRevisionId = revisions[ 1 ]._revId
				return #revisions == 1
			end )
		)
	observeAsset:expectDataWithinTime( 30, true )
	observeAsset:dispose()

	rxTest:expectEqual( latestRevisionId, revisionId )

	-- should not send a second update to Oz

	sendNUpdates = rxTest:observe( pusher:_DEBUG_sendNUpdates( 0 ) )
	sendNUpdates:expectCompletedWithinTime( 30 )
	sendNUpdates:dispose()

	-- move the aux file to make it now uploadable so when BinaryMissingError
	-- is received it should trigger upload finally

	WFFileUtils.move( tempAuxFilePath, auxFilePath )

	-- should eventually sync a BinaryMissingError, which should trigger a retry

	local errorId = rxTest:getFirstValue(
		catalog:indexStore(): query( {
			sql = [[ SELECT docId FROM errors ]],
			singleColumnQuery = true }
		)
		:filter( function( errorIds )
			return #errorIds > 0
		end )
		:pluck( 1 )
		:take( 1 ) )

	local errorDoc = rxTest:getFirstValue( catalog:docStore():docValue( errorId ) )

	rxTest:expectEqual( 'BinaryMissingError', errorDoc.subtype )
	rxTest:expectEqual( assetId, errorDoc._serverProvidedMetadata.source.id )
	rxTest:expectEqual( revisionId, errorDoc._serverProvidedMetadata.source.revision_id )

	-- asset should still not have a new revision id but an update should be sent to Oz
	-- triggered by the _localOnly changes

	observeAsset = rxTest:observe (
		catalog:docStore():ongoingDocValue( assetId )
			:map( function( docValue )
				latestRevisionId = docValue._revId
				return latestRevisionId == revisionId
			end )
		)
	observeAsset:expectDataWithinTime( 30, true )
	observeAsset:dispose()

	sendNUpdates = rxTest:observe( pusher:_DEBUG_sendNUpdates( 1 ) )
	sendNUpdates:expectCompletedWithinTime( 30 )
	sendNUpdates:dispose()

	-- Once aux uploads the retried revision should eventually make it to Oz and finally
	-- sync back down without temp link and mark asset as in sync with Oz.

	observeAsset = observeIsOnOz( rxTest, catalog, assetId )
	observeAsset:expectDataWithinTime( 30, true )
	observeAsset:dispose()

	-- should no longer have temp link

	observeAsset = rxTest:observe (
		catalog:docStore():ongoingDocValue( assetId )
			:map( function( docValue )
				return KSTableUtils.derefOpt( docValue, '_serverProvidedMetadata', 'links', tempLinkKey ) == nil
			end )
		)
	observeAsset:expectDataWithinTime( 30, true )
	observeAsset:dispose()

end

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

local function testAllowPushOfMultipleCreatesAndDeletes( rxTest, type, catalogOrSpaceId, catalogId )

	rxTest:setTimeLimit( 5 )

	local docStoreMocks = generateDocStoreMocks( type, catalogOrSpaceId, { includeAssetUpdate = true, includeAssetDelete = true, excludeAlbumAsset = true } )

	local assetDocId = docStoreMocks.assetDocId
	local albumDocId = docStoreMocks.albumDocId

	local assetSecondCreateMock = KSTableUtils.deepCopy( docStoreMocks.pusherUpdates[ 1 ] )
	assetSecondCreateMock.revision_id = docStoreMocks.assetUpdateOverlay._newRevId
	assetSecondCreateMock.payload.xmp.dc.description = docStoreMocks.assetUpdateOverlay.xmp.dc.description

	docStoreMocks.pullerUpdates[ 1 ].revision_ids = { docStoreMocks.assetUpdateOverlay._newRevId }
	docStoreMocks.pullerUpdates[ 1 ].payload = KSTableUtils.deepCopy( docStoreMocks.pullerUpdates[ 1 ].payload )
	docStoreMocks.pullerUpdates[ 1 ].payload.xmp.dc.description = docStoreMocks.assetUpdateOverlay.xmp.dc.description

	local newAlbumName = KSIdGenerator.uuidInLowercaseHex()
	local newAlbumRevisionId = KSIdGenerator.uuidInLowercaseHex()
	local albumOverlay = {
		_id = albumDocId,
		_newRevId = newAlbumRevisionId,
		name = newAlbumName,
	}

	local albumSecondCreateMock = KSTableUtils.deepCopy( docStoreMocks.pusherUpdates[ 2 ] )
	albumSecondCreateMock.revision_id = newAlbumRevisionId
	albumSecondCreateMock.payload.name = newAlbumName

	docStoreMocks.pullerUpdates[ 2 ].revision_ids = { newAlbumRevisionId }
	docStoreMocks.pullerUpdates[ 2 ].payload = KSTableUtils.deepCopy( docStoreMocks.pullerUpdates[ 1 ].payload )
	docStoreMocks.pullerUpdates[ 2 ].payload.name = newAlbumName

	local albumDeleteOverlay1 = {
		_id = albumDocId,
		_newRevId = KSIdGenerator.uuidInLowercaseHex(),
		_deleted = true,
	}

	local albumDeleteOverlay2 = KSTableUtils.deepCopy( albumDeleteOverlay1 )
	albumDeleteOverlay2._newRevId = KSIdGenerator.uuidInLowercaseHex()

	local albumDeleteMock1 = {
		type = "album",
		action = "delete",
		revision_id = albumDeleteOverlay1._newRevId,
		id = albumDocId,
	}
	if type == 'space' then
		albumDeleteMock1.space_id = catalogOrSpaceId
	end

	local albumDeleteMock2 = KSTableUtils.deepCopy( albumDeleteMock1 )
	albumDeleteMock2.revision_id = albumDeleteOverlay2._newRevId

	local assetDeleteOverlay2 = KSTableUtils.deepCopy( docStoreMocks.assetDeleteOverlay )
	assetDeleteOverlay2._newRevId = KSIdGenerator.uuidInLowercaseHex()

	local assetDeleteMock2 = KSTableUtils.deepCopy( docStoreMocks.assetDeleteMock )
	assetDeleteMock2.revision_id = assetDeleteOverlay2._newRevId

	local reqMocks = {

		pusherMockReqResp( type, catalogOrSpaceId, docStoreMocks.pusherUpdates ),
		pusherMockReqResp( type, catalogOrSpaceId, { assetSecondCreateMock, albumSecondCreateMock } ),

		prefetchReqResp( catalogId ),
		catalogPrefetchReqResp( catalogId ),
		pullerMockReqResp( type, catalogOrSpaceId, docStoreMocks.pullerUpdates ),

		pusherMockReqResp( type, catalogOrSpaceId, { docStoreMocks.assetDeleteMock, albumDeleteMock1 } ),
		pusherMockReqResp( type, catalogOrSpaceId, { assetDeleteMock2, albumDeleteMock2 } ),

	}

	-- remove the prefetch calls for albums and catalog for space tests

	if type == 'space' then

		table.remove( reqMocks, 4 )
		table.remove( reqMocks, 3 )

	end

	local catalogSpec = WFCatSpecMock( catalogId, kUserId, rxTest, reqMocks )
	local catalog = WFCatMockCatalog.createWFCatalogForTestWithSpec( rxTest, catalogSpec )

	local docStore = catalog:docStore()

	local pusherPullerOptions = options

	if type == 'space' then

		pusherPullerOptions = KSTableUtils.deepCopy( options )
		pusherPullerOptions.spaceId = catalogOrSpaceId
		pusherPullerOptions.updatesFeedUrl = catalog:ozCatalogSpec():spaceUpdatesUrlForSpaceId( catalogOrSpaceId )
		pusherPullerOptions.isCatalogPuller = false
		pusherPullerOptions.isSpacePusher = true

	end

	local pusher = WFCatMetadataPusher( catalog, pusherPullerOptions )
	rxTest:addCloseTarget( pusher )

	local revisionsAlreadySent = pusher:DEBUG_revisionsAlreadySent()
		-- this is the in memory table used to track what revisions have been sent
		-- to Oz to avoid sending multiple times unnecessarily. Since these are
		-- all create and delete requests, we expect to see nothing saved to
		-- this table.

	local puller = WFCatMetadataPuller( catalog, "Test Puller", pusherPullerOptions )
	rxTest:addCloseTarget( puller )

	-- Create an album and asset, push them to Oz a first time

	rxTest:waitFor( docStore:bulkSetDocValue( docStoreMocks.docStoreValues ) )

	rxTest:waitFor( pusher:_DEBUG_sendNUpdates( 2 ) )

	rxTest:expectEqual( nil, revisionsAlreadySent[ assetDocId ] )
	rxTest:expectEqual( nil, revisionsAlreadySent[ albumDocId ] )

	-- Update album and asset, should send second create request to Oz

	rxTest:waitFor( docStore:applyOverlay( docStoreMocks.assetUpdateOverlay ) )
	rxTest:waitFor( docStore:applyOverlay( albumOverlay ) )

	rxTest:waitFor( pusher:_DEBUG_sendNUpdates( 2 ) )

	rxTest:expectEqual( nil, revisionsAlreadySent[ assetDocId ] )
	rxTest:expectEqual( nil, revisionsAlreadySent[ albumDocId ] )

	-- sync updates from Oz, WF will not try to delete something until it thinks Oz has it

	rxTest:expectOnlyCompleted( puller:getUpdatesOnce() )

	-- Delete album and asset, should send delete request to Oz

	rxTest:waitFor( docStore:applyOverlay( docStoreMocks.assetDeleteOverlay ) )
	rxTest:waitFor( docStore:applyOverlay( albumDeleteOverlay1 ) )

	rxTest:waitFor( pusher:_DEBUG_sendNUpdates( 2 ) )

	rxTest:expectEqual( nil, revisionsAlreadySent[ assetDocId ] )
	rxTest:expectEqual( nil, revisionsAlreadySent[ albumDocId ] )

	-- Delete album and asset again with new revision id, should send delete request to Oz
	-- (this is not something that should ever happen, very edge case, just ensuring
	-- code works as expected)

	rxTest:waitFor( docStore:applyOverlay( assetDeleteOverlay2 ) )
	rxTest:waitFor( docStore:applyOverlay( albumDeleteOverlay2 ) )

	rxTest:waitFor( pusher:_DEBUG_sendNUpdates( 2 ) )

	rxTest:expectEqual( nil, revisionsAlreadySent[ assetDocId ] )
	rxTest:expectEqual( nil, revisionsAlreadySent[ albumDocId ] )

end

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

function WFCatMetadataPusherTest.rxTestAllowPushOfMultipleCreatesAndDeletesForCatalog( rxTest )

	local catalogId = KSIdGenerator:uuidInLowercaseHex()

	testAllowPushOfMultipleCreatesAndDeletes( rxTest, 'catalog', catalogId, catalogId )

end

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

function WFCatMetadataPusherTest.rxTestAllowPushOfMultipleCreatesAndDeletesForSpaces( rxTest )

	local spaceId = KSIdGenerator.uuidInLowercaseHex()
	local catalogId = KSIdGenerator:uuidInLowercaseHex()

	testAllowPushOfMultipleCreatesAndDeletes( rxTest, 'space', spaceId, catalogId )

end

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

function WFCatMetadataPusherTest.rxTestAllowPushOfMultipleCreates_testNightly( rxTest )

	-- This test is to confirm that the fix for WF-2295 that disallows pushing
	-- same revision id and payload to Oz twice does not inadvertantly interfere
	-- with existing behavior that allows pushing creates multiple times. (if it
	-- ain't broke, don't fix it)

	rxTest:setTimeLimit( 120 )

	local catalogTestSession = WFCatTestSession( rxTest )
	local catalog = rxTest:getFirstValue( catalogTestSession:catalog() )

	local puller = WFCatMetadataPuller( catalog, "Test Puller", options )
	rxTest:addCloseTarget( puller )

	-- start pulling metadata and ensure catalog entry sync from Oz which should
	-- assure that when we resume pulling, the request will have an updated_since
	-- parameter which is required to sync error documents from Oz.

	puller:getUpdatesContinuously()

	local observeCatalogOnOz = observeHasServerProvidedMetadata( rxTest, catalog, catalog:ozCatalogId() )
	observeCatalogOnOz:expectDataWithinTime( 30, true )
	observeCatalogOnOz:dispose()

	-- pause metadata puller before creating asset to trigger scenario where create asset is sent to Oz twice

	puller:stopContinuousUpdates()

	local pusher = WFCatMetadataPusher( catalog, options )
	rxTest:addCloseTarget( pusher )

	pusher:addRequestHeader( 'X-Quicker-Retries', 'true' )
		-- need this to be able to see expected error documents sync

	-- create an asset

	local assetId = KSIdGenerator.uuidInLowercaseHex()
	local revisionId = KSIdGenerator.uuidInLowercaseHex()

	local testDataPath = KSPathUtils.child( KSPathUtils.getStandardFilePath( 'testdata' ), 'import-samples' )
	local testDataFolder1 = KSPathUtils.child( testDataPath, "folder-1" )
	local originalFilePath = KSPathUtils.child( testDataFolder1, "small.JPG" )
	local sha256 = WFCrypto.digestFile( originalFilePath, 'SHA256' )

	local assetDoc = WFCatAssetTestUtils.createBaseAssetDocument( catalog:ozCatalogId(), assetId, revisionId, sha256, originalFilePath )

	rxTest:waitFor( catalog:docStore():setDocValue( assetDoc ) )

	-- initial asset upload to Oz

	local sendNUpdates = rxTest:observe( pusher:_DEBUG_sendNUpdates( 1 ) )
	sendNUpdates:expectCompletedWithinTime( 30 )
	sendNUpdates:dispose()

	-- create a second revision before metadata syncs down to Oz.
	-- WF should allow a second create attempt to be made and
	-- Oz should generate an error for the second attempt.

	local secondCreateRevisionId = KSIdGenerator.uuidInLowercaseHex()
	local secondCreateOverlay = {
		_id = assetId,
		_newRevId = secondCreateRevisionId,
		xmp = {
			dc = {
				description = "Not Rum River State Forest, near Winona, Minnesota",
			},
		},
	}
	rxTest:waitFor( catalog:docStore():applyOverlay( secondCreateOverlay ) )

	sendNUpdates = rxTest:observe( pusher:_DEBUG_sendNUpdates( 1 ) )
	sendNUpdates:expectCompletedWithinTime( 30 )
	sendNUpdates:dispose()

	-- wait for the error for the duplicate create on Oz

	puller:getUpdatesContinuously()

	local observeAssetOnOz = observeHasServerProvidedMetadata( rxTest, catalog, assetId )
	observeAssetOnOz:expectDataWithinTime( 30, true )
	observeAssetOnOz:dispose()

	rxTest:sleep( 10 )

	local errorDoc
	local function grabErrorDoc( docs )
		for _, doc in ipairs( docs ) do
			if doc.type == 'error' then
				errorDoc = doc
				return true
			end
		end
	end

	local observeErrorOnOz = rxTest:observe(
		catalog:docStore():changedDocs { fromChangeCounter = 0 }
			:map( grabErrorDoc )
	)
	observeErrorOnOz:expectDataWithinTime( 30, true )
	observeErrorOnOz:dispose()

	rxTest:expectEqual( 'ResourceExistsError', errorDoc.subtype )
	rxTest:expectEqual( assetId, errorDoc._serverProvidedMetadata.source.id )

end

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

function WFCatMetadataPusherTest.rxTestAllowPushOfMultipleDeletes_testNightly( rxTest )

	-- This test is to confirm that the fix for WF-2295 that disallows pushing
	-- same revision id and payload to Oz twice does not inadvertantly interfere
	-- with existing behavior that allows pushing deletes multiple times. (if it
	-- ain't broke, don't fix it)

	rxTest:setTimeLimit( 120 )

	local catalogTestSession = WFCatTestSession( rxTest )
	local catalog = rxTest:getFirstValue( catalogTestSession:catalog() )

	local puller = WFCatMetadataPuller( catalog, "Test Puller", options )
	rxTest:addCloseTarget( puller )

	-- start pulling metadata and ensure catalog entry sync from Oz which should
	-- assure that when we resume pulling, the request will have an updated_since
	-- parameter which is required to sync error documents from Oz.

	puller:getUpdatesContinuously()

	local observeCatalogOnOz = observeHasServerProvidedMetadata( rxTest, catalog, catalog:ozCatalogId() )
	observeCatalogOnOz:expectDataWithinTime( 30, true )
	observeCatalogOnOz:dispose()

	local pusher = WFCatMetadataPusher( catalog, options )
	rxTest:addCloseTarget( pusher )

	pusher:addRequestHeader( 'X-Quicker-Retries', 'true' )
		-- need this to be able to see expected error documents sync

	-- create an asset

	local assetId = KSIdGenerator.uuidInLowercaseHex()
	local revisionId = KSIdGenerator.uuidInLowercaseHex()

	local testDataPath = KSPathUtils.child( KSPathUtils.getStandardFilePath( 'testdata' ), 'import-samples' )
	local testDataFolder1 = KSPathUtils.child( testDataPath, "folder-1" )
	local originalFilePath = KSPathUtils.child( testDataFolder1, "small.JPG" )
	local sha256 = WFCrypto.digestFile( originalFilePath, 'SHA256' )

	local assetDoc = WFCatAssetTestUtils.createBaseAssetDocument( catalog:ozCatalogId(), assetId, revisionId, sha256, originalFilePath )

	rxTest:waitFor( catalog:docStore():setDocValue( assetDoc ) )

	-- initial asset upload to Oz

	local sendNUpdates = rxTest:observe( pusher:_DEBUG_sendNUpdates( 1 ) )
	sendNUpdates:expectCompletedWithinTime( 30 )
	sendNUpdates:dispose()

	local observeAssetOnOz = observeHasServerProvidedMetadata( rxTest, catalog, assetId )
	observeAssetOnOz:expectDataWithinTime( 30, true )
	observeAssetOnOz:dispose()

	-- pause metadata puller before delete asset to trigger scenario where delete asset is sent to Oz twice

	puller:stopContinuousUpdates()

	-- delete asset twice

	rxTest:waitFor( catalog:docStore():applyOverlay( {
		_id = assetId,
		_newRevId = KSIdGenerator.uuidInLowercaseHex(),
		_deleted = true,
	} ) )

	sendNUpdates = rxTest:observe( pusher:_DEBUG_sendNUpdates( 1 ) )
	sendNUpdates:expectCompletedWithinTime( 30 )
	sendNUpdates:dispose()

	rxTest:waitFor( catalog:docStore():applyOverlay( {
		_id = assetId,
		_newRevId = KSIdGenerator.uuidInLowercaseHex(),
		_deleted = true,
	} ) )

	sendNUpdates = rxTest:observe( pusher:_DEBUG_sendNUpdates( 1 ) )
	sendNUpdates:expectCompletedWithinTime( 30 )
	sendNUpdates:dispose()

	-- wait for the error for the duplicate create on Oz

	puller:getUpdatesContinuously()

	local observeTombstone = observeIsDeletedOnOz( rxTest, catalog, assetId )
	observeTombstone:expectDataWithinTime( 30, true )
	observeTombstone:dispose()

	rxTest:sleep( 10 )

	local errorDoc
	local function grabErrorDoc( docs )
		for _, doc in ipairs( docs ) do
			if doc.type == 'error' then
				errorDoc = doc
				return true
			end
		end
	end

	local observeErrorOnOz = rxTest:observe(
		catalog:docStore():changedDocs { fromChangeCounter = 0 }
			:map( grabErrorDoc )
	)
	observeErrorOnOz:expectDataWithinTime( 30, true )
	observeErrorOnOz:dispose()

	rxTest:expectEqual( 'ResourceNotFoundError', errorDoc.subtype )
	rxTest:expectEqual( assetId, errorDoc._serverProvidedMetadata.source.id )

end

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

-- TODO [scouten 2015-07-21]: Error cases.

-- TODO [scouten 2015-08-05]: Failure within generate update logic.

-- TODO [scouten 2015-07-30]: Respond appropriately when error documents appear.

-- TODO [scouten 2015-08-05]: Send changes as appropriate when local changes occur.

-- TODO [scouten 2015-08-05]: Fix possible race condition when receiving Oz documents during send cycle.

-- TODO [scouten 2015-07-30]: Study WFSyncAgent unit tests for more test cases.

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

return WFRxTestCase.wrapSuite( 'WFCatMetadataPusherTest', WFCatMetadataPusherTest )
