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

WFCatWatermarkTestFactory.lua

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

ADOBE CONFIDENTIAL
------------------
Copyright 2020 Adobe.
All Rights Reserved.

NOTICE: All information contained herein is, and remains 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 are protected by all applicable
intellectual property laws, including trade secret and copyright laws.
Dissemination of this information or reproduction of this material is
strictly forbidden unless prior written permission is obtained from
Adobe.

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

	-- RxLua
local Rx = require 'Rx'
local Observable = Rx.Observable

	-- WF.core.self_test

local WFRxTestCase = require 'WFRxTestCase'

	-- WF.core.utils
local WFTableArchiving = require 'WFTableArchiving'
local WFModelsUtils = require 'WFModelsUtils'

	-- WF.oz_client.managed_catalog
local KSData = require 'KSData'
local WFCatWatermark = require 'WFCatWatermark'
local KSTableUtils = require 'KSTableUtils'
local WFCatMetadataPuller = require 'WFCatMetadataPuller'
local WFCatMetadataPusher = require 'WFCatMetadataPusher'
local WFArguments = require 'WFArguments'
local WFTableMerge = require 'WFTableMerge'
local KSPathUtils = require 'KSPathUtils'
local WFFileUtils = require 'WFFileUtils'
local WFCatErrorHandler = require 'WFCatErrorHandler'
local WFCrypto = require 'WFCrypto'
local KSIdGenerator = require 'KSIdGenerator'
local WFCatBinaryUploader = require 'WFCatBinaryUploader'
local WFCatMockCatalog = require 'WFCatMockCatalog'
local WFCatTestSession = require 'WFCatTestSession'
local WFTestFileUtils = require 'WFTestFileUtils'
local WFCatAssetUtils = require 'WFCatAssetUtils'
local WFAsyncFileUtils = require 'WFAsyncFileUtils'
	-- WF.core.rxlua
local WFRxScheduler = require 'WFRxScheduler'

--------------------------------------------------------------------------------
local derefOpt = KSTableUtils.derefOpt
--==============================================================================
local check = assert( WFArguments.check )


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

local function removeUserCreatedAndUpdatedField( watermarkInfo )

	watermarkInfo["userCreated"] = nil
	watermarkInfo["userUpdated"] = nil

	return watermarkInfo

end

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

local function getWatermarkTestSetttings()

	return {
		opacity = 73,
		rotation = 90,
		position = "center",
		xOffset = 0.23,
		yOffset = 0.40,
		size = 22,
		text = "@ saravana",
		font = {
			name = "SourceSansPro",
			bold = true,
			italic = true,
		},
		shadow = {
			enabled = "off"
		},
		rgbColor = { 0.4, 0.33, 0.0 },
		activeWatermarkType = "text",
	}

end

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

return function( optSubmodelsByImplementation, maybeExistingTestClass )

	----------------------------------------------------------------------------
	
	local function callFunctionOrSubmodel( rxTest, wfCatalog, fnNamespace, fnName, ... )
	
		return WFModelsUtils.callFunctionOrSubmodel( rxTest, optSubmodelsByImplementation, wfCatalog, fnNamespace, fnName, ... )
					
	end

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

	local function shouldIncludeNightlyTest()

		-- Don't run nightly tests for submodel tests.
		
		return not optSubmodelsByImplementation

	end

	----------------------------------------------------------------------------
	
	local WFCatWatermarkTest = maybeExistingTestClass or {}
	
	----------------------------------------------------------------------------
		
	function WFCatWatermarkTest.rxTestCreateTextWatermarkTest( rxTest )

		rxTest:setTimeLimit( 30 )

		local wfCatalog = WFCatMockCatalog.createFakeWFCatalog( rxTest )

		local name = "My First Watermark"

		local settings = {
			opacity = 73,
			rotation = 90,
			position = "center",
			xOffset = 0.23,
			yOffset = 0.40,
			size = 22,
			text = "@ saravana",
			font = {
				name = "SourceSansPro",
				bold = true,
				italic = true,
			},
			shadow = {
				enabled = "off"
			},
			rgbColor = { 0.4, 0.33, 0.0 },
			activeWatermarkType = "text",
		}

		local wmId

		rxTest:waitFor( callFunctionOrSubmodel(
					rxTest, wfCatalog,
					'WFCatWatermark',
					'createWatermark', wfCatalog, name, settings )
				:tap( function( watermarkId )
					-- capture the watermark id returned, it will be used for validation later.
					wmId = watermarkId
				end ) )

		-- wrapper for validating the received watermarkInfo
		local expectedSettings = settings

		expectedSettings.id = wmId
		expectedSettings.displayName = name
		expectedSettings.activeWatermarkType = "text"

		rxTest:expectSingleValue( callFunctionOrSubmodel(
					rxTest, wfCatalog,
					'WFCatWatermark',
					'getWatermarkInfo', wfCatalog, wmId )
			:map( function( watermarkInfo )

				watermarkInfo = removeUserCreatedAndUpdatedField( watermarkInfo )
				return KSTableUtils.deepEquals( expectedSettings, watermarkInfo )

			end ), true )

	end

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

	function WFCatWatermarkTest.rxTestUpdateTextWatermarkTest( rxTest )

		rxTest:setTimeLimit( 30 )

		local wfCatalog = WFCatMockCatalog.createFakeWFCatalog( rxTest )

		local name = "This watermark will be updated soon"
		local settings = {
			opacity = 73,
			rotation = 90,
			position = "center",
			xOffset = 0.23,
			yOffset = 0.40,
			size = 22,
			text = "@ saravana",
			font = {
				name = "SourceSansPro",
				bold = true,
				italic = true,
			},
			shadow = {
				enabled = "off" -- [ off, simple ]
			},
			rgbColor = { 0.5, 0.6, 0.7 },
			activeWatermarkType = "text",
		}

		local wmId

		rxTest:waitFor( callFunctionOrSubmodel(
					rxTest, wfCatalog,
					'WFCatWatermark',
					'createWatermark', wfCatalog, name, settings )
				:tap( function( watermarkId )
					-- capture the watermark id returned, it will be used for validation later.
					wmId = watermarkId
				end ) )

		-- update the watermark settings.
			local updatedName = "This watermark just got updated"
			local updatedSettings = {
				-- remove opacity
				--opacity = 73,		-- removed
				rotation = 90,
				position = "center",
				xOffset = 0.23,
				yOffset = 0.40,
				size = 33, 			-- modified
				text = "@ saravana",
				font = {
					name = "Arial", -- modified
					bold = true,
					italic = false, -- modified
				},
				shadow = {
					enabled = "simple" -- modified [ off, simple ]
				},
				rgbColor = { 0.1, 0.2, 0.3 },
				activeWatermarkType = "text",
			}

		rxTest:waitFor( callFunctionOrSubmodel(
					rxTest, wfCatalog,
					'WFCatWatermark',
					'updateWatermark', wfCatalog, wmId, updatedName, updatedSettings ) )

		updatedSettings = removeUserCreatedAndUpdatedField( updatedSettings )

		-- wrapper for validating the received watermarkInfo
		local expectedSettings = updatedSettings
		expectedSettings.displayName = updatedName
		expectedSettings.id = wmId


		rxTest:expectSingleValue( callFunctionOrSubmodel(
					rxTest, wfCatalog,
					'WFCatWatermark',
					'getWatermarkInfo', wfCatalog, wmId )
			:map( function( watermarkDoc )

				 watermarkDoc = removeUserCreatedAndUpdatedField(watermarkDoc)
				 return KSTableUtils.deepEquals( expectedSettings, watermarkDoc )

			end ), true )

	end

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

	function WFCatWatermarkTest.rxTestGetWatermarkInfoTest( rxTest )

		rxTest:setTimeLimit( 30 )

		local wfCatalog = WFCatMockCatalog.createFakeWFCatalog( rxTest )

		local name = "My First Watermark"
		local settings = {
			opacity = 73,
			rotation = 90,
			position = "center",
			xOffset = 0.23,
			yOffset = 0.40,
			size = 22,
			text = "@ saravana",
			font = {
				name = "SourceSansPro",
				bold = false,
				italic = true,
			},
			shadow = {
				enabled = "simple" -- [ off, simple ]
			},
			rgbColor = { 0.4, 0.33, 0.0 },
			activeWatermarkType = "text",
		}

		local wmId

		rxTest:waitFor( callFunctionOrSubmodel(
					rxTest, wfCatalog,
					'WFCatWatermark',
					'createWatermark', wfCatalog, name, settings )
				:tap( function( watermarkId )
					-- capture the watermark id returned, it will be used for validation later.
					wmId = watermarkId
				end ) )

		-- wrapper for validating the received watermarkInfo
		local expectedSettings = settings
		expectedSettings.id = wmId
		expectedSettings.displayName = name
		expectedSettings.activeWatermarkType = "text"

		rxTest:expectSingleValue( callFunctionOrSubmodel(
					rxTest, wfCatalog,
					'WFCatWatermark',
					'getWatermarkInfo', wfCatalog, wmId )
			:map( function( watermarkInfo )

				watermarkInfo = removeUserCreatedAndUpdatedField(watermarkInfo)
				return KSTableUtils.deepEquals( expectedSettings, watermarkInfo )

			end ), true )

	end

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

	function WFCatWatermarkTest.rxTestGetAllWatermarksInfoTest( rxTest )

		rxTest:setTimeLimit( 30 )

		local wfCatalog = WFCatMockCatalog.createFakeWFCatalog( rxTest )

		local name1 = "TEST-WATERMARK-1"
		local settings1 = {
			opacity = 19,
			rotation = 79,
			position = "center",
			xOffset = 0.23,
			yOffset = 0.40,
			size = 22,
			text = "@ saravana",
			font = {
				name = "SourceSansPro",
				bold = true,
				italic = true,
			},
			shadow = {
				enabled = "off" -- [ off, simple ]
			},
			rgbColor = { 0.4, 0.33, 0.0 },
			activeWatermarkType = "text",
		}

		local wmId1

		rxTest:waitFor( callFunctionOrSubmodel(
					rxTest, wfCatalog,
					'WFCatWatermark',
					'createWatermark', wfCatalog, name1, settings1 )
				:tap( function( watermarkId )
					-- capture the watermark id returned, it will be used for validation later.
					wmId1 = watermarkId
			end ) )

		--- Insert Second Doc
		local name2 = "TEST-WATERMARK-2"
		local settings2 = {
			opacity = 20,
			rotation = 80,
			position = "center",
			xOffset = 0.23,
			yOffset = 0.40,
			size = 22,
			text = "@ saravana",
			font = {
				name = "SourceSansPro",
				bold = true,
				italic = true,
			},
			shadow = {
				enabled = "off" -- [ off, simple ]
			},
			rgbColor = { 0.4, 0.33, 0.0 },
			activeWatermarkType = "text",
		}

		local wmId2

		rxTest:waitFor( callFunctionOrSubmodel(
					rxTest, wfCatalog,
					'WFCatWatermark',
					'createWatermark', wfCatalog, name2, settings2 )
				:tap( function( watermarkId )
					-- capture the watermark id returned, it will be used for validation later.
					wmId2 = watermarkId
				end ) )

		-- wrappers for comparing test output
		local expectedSettings1 = settings1
		expectedSettings1.id = wmId1
		expectedSettings1.displayName = name1
		expectedSettings1.activeWatermarkType = "text"

		local expectedSettings2 = settings2
		expectedSettings2.id = wmId2
		expectedSettings2.displayName = name2
		expectedSettings2.activeWatermarkType = "text"

		local expectedWatermarkSettingsArray = { expectedSettings1, expectedSettings2 }

		rxTest:expectSingleValue( WFCatWatermark.getAllWatermarksInfo( wfCatalog )
			:take( 1 )
			:map( function( watermarksInfoList )

				if( #expectedWatermarkSettingsArray ~= #watermarksInfoList ) then
					return false
				end

				for i = 1, #watermarksInfoList do
					watermarksInfoList[i] = removeUserCreatedAndUpdatedField(watermarksInfoList[i])
					if not ( KSTableUtils.deepEquals( expectedWatermarkSettingsArray[i], watermarksInfoList[i] ) ) then
						return false
					end
				end

				return true

			end ), true )

	end

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

	local options = {

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

		testMode = true,

		isCatalogPuller = true

	}
	------------------------------------------------------------------------

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

		for _, doc in ipairs( changedDocsResult ) do

			remapSingleDocument( doc, true )

		end

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

		return changedDocsResult

	end

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

	function WFCatWatermarkTest.rxTestSendUpdates_realOz_testNightly( rxTest )

		rxTest:setTimeLimit( 120 )

		local catalogTestSession = WFCatTestSession( rxTest )

		local catalog

		rxTest:waitFor(
			catalogTestSession:catalog()
				:tap( function( _catalog )
					check( 'WFCatalog' )
					catalog = _catalog
				end ) )

		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.

		local name = "My First Watermark"
		local settings = {
			opacity = 73,
			rotation = 90,
			position = "center",
			xOffset = 0.23,
			yOffset = 0.40,
			size = 22,
			text = "@ saravana",
			font = {
				name = "SourceSansPro",
				bold = true,
				italic = true,
			},
			shadow = {
				enabled = "off" -- [ off, simple ]
			},
			rgbColor = { 0.4, 0.33, 0.0 },
			activeWatermarkType = "text",
		}

		local wmId
		rxTest:waitFor( callFunctionOrSubmodel(
					rxTest, catalog,
					'WFCatWatermark',
					'createWatermark', catalog, name, settings )
				:tap( function( watermarkId )
					-- capture the watermark id returned, it will be used for validation later.
					wmId = watermarkId
				end ) )

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

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

		rxTest:expectSingleValue(
			catalog:docStore():changedDocs { fromChangeCounter = 0 }
				:map( remapChangedDocsFeed ),
					{
						{
							_labels = {
								oz = true,
							},
							_serverProvidedMetadata = {
								base = ">>base",
								created = ">>created",
								links = ">>links",
								revision_ids = ">>revision_ids",
								updated = ">>updated",
							},
							name = "WFCatTestSession",
							subtype = "lightroom",
							type = "catalog",
						},
						{
							_labels = {
								oz = true,
							},
							_serverProvidedMetadata = {
								base = ">>base",
								created = ">>created",
								links = ">>links",
								revision_ids = ">>revision_ids",
								updated = ">>updated",
							},
							displayName = "My First Watermark",
							font = {
								bold = true,
								italic = true,
								name = "SourceSansPro",
							},
							opacity = 73,
							position = "center",
							rgbColor = {
								0.4,
								0.33,
								0,
							},
							rotation = 90,
							shadow = {
								enabled = "off",
							},
							size = 22,
							subtype = "text",
							activeWatermarkType = "text",
							text = "@ saravana",
							type = "watermark",
							userCreated = ">>userCreated",
							userUpdated = ">>userUpdated",
							xOffset = 0.23,
							yOffset = 0.4,
						},
						continueFrom = 4,
					}
	 )

		-- modify settings
		settings.rgbColor[3] = 0.9
		rxTest:waitFor( callFunctionOrSubmodel(
					rxTest, catalog,
					'WFCatWatermark',
					'updateWatermark', catalog, wmId, name, settings ) )

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

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

		rxTest:expectSingleValue(
			catalog:docStore():changedDocs { fromChangeCounter = 0 }
				:map( remapChangedDocsFeed ),
		{
			{
				_labels = {
					oz = true,
				},
				_serverProvidedMetadata = {
					base = ">>base",
					created = ">>created",
					links = ">>links",
					revision_ids = ">>revision_ids",
					updated = ">>updated",
				},
				name = "WFCatTestSession",
				subtype = "lightroom",
				type = "catalog",
			},
			{
				_serverProvidedMetadata = {
					base = ">>base",
					created = ">>created",
					links = ">>links",
					revision_ids = ">>revision_ids",
					updated = ">>updated",
				},
				displayName = "My First Watermark",
				font = {
					bold = true,
					italic = true,
					name = "SourceSansPro",
				},
				opacity = 73,
				position = "center",
				rgbColor = { --- verify that the first value got updated to 0.9 from 0.4
					0.4,
					0.33,
					0.9,
				},
				rotation = 90,
				shadow = {
					enabled = "off",
				},
				size = 22,
				subtype = "text",
				text = "@ saravana",
				activeWatermarkType = "text",
				type = "watermark",
				userCreated = ">>userCreated",
				userUpdated = ">>userUpdated",
				xOffset = 0.23,
				yOffset = 0.4,
			},
			continueFrom = 5,
		}
	 )

	end

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

	function WFCatWatermarkTest.rxTestTruncateTextLengthExceeds1024chars( rxTest )

		rxTest:setTimeLimit( 30 )

		local wfCatalog = WFCatMockCatalog.createFakeWFCatalog( rxTest )

		local name = "Long Text Watermark"

		local longtext = [[watermark, watermark, watermark, watermark, watermark, watermark,
							watermark, watermark, watermark, watermark, watermark, watermark,
							watermark, watermark, watermark, watermark, watermark, watermark,
							watermark, watermark, watermark, watermark, watermark, watermark,
							watermark, watermark, watermark, watermark, watermark, watermark,
							watermark, watermark, watermark, watermark, watermark, watermark,
							watermark, watermark, watermark, watermark, watermark, watermark,
							watermark, watermark, watermark, watermark, watermark, watermark,
							watermark, watermark, watermark, watermark, watermark, watermark,
							watermark, watermark, watermark, watermark, watermark, watermark,
							watermark, watermark, watermark, watermark, watermark, watermark,
							watermark, watermark, watermark, watermark, watermark, watermark,
							watermark, watermark, watermark, watermark, watermark, watermark,
							watermark, watermark, watermark, watermark, watermark, watermark,
							watermark, water exceeds the 1024 limit]]

		local settings = {
			opacity = 73,
			rotation = 90,
			position = "center",
			xOffset = 0.23,
			yOffset = 0.40,
			size = 22,
			text = longtext,
			font = {
				name = "SourceSansPro",
				bold = true,
				italic = true,
			},
			shadow = {
				enabled = "off"
			},
			rgbColor = { 0.4, 0.33, 0.0 },
			activeWatermarkType = "text",
		}

		rxTest:expectSingleValue(
			 callFunctionOrSubmodel(
					rxTest, wfCatalog,
					'WFCatWatermark',
					'createWatermark',  wfCatalog, name, settings ),
			':ignore:' )

		local wmId

		rxTest:expectSingleValue( wfCatalog:indexStore():queryOnce {
			sql =  [[ SELECT docId, created FROM watermark ORDER BY created ASC ]],
			quiesceUntilUpToDate = true,
		}
		:tap( function( rows )
			 wmId = rows[1][1]
		end ), ':ignore:' )

		-- wrapper for validating the received watermarkInfo
		local expectedSettings = settings

		expectedSettings.id = wmId
		expectedSettings.displayName = name

		-- truncate the expected string to max limit of 1024
		expectedSettings.text = string.sub(longtext, 1, 1024)

		rxTest:expectSingleValue( callFunctionOrSubmodel(
					rxTest, wfCatalog,
					'WFCatWatermark',
					'getWatermarkInfo', wfCatalog, wmId )
			:map( function( watermarkInfo )

				watermarkInfo = removeUserCreatedAndUpdatedField( watermarkInfo )
				return KSTableUtils.deepEquals( expectedSettings, watermarkInfo )

			end ), true )

	end

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

	function WFCatWatermarkTest.rxTestDeleteOneWatermarkTest( rxTest )

		rxTest:setTimeLimit( 40 )

		local wfCatalog = WFCatMockCatalog.createFakeWFCatalog( rxTest )

		local name = "My First Watermark"

		local settings = {
			opacity = 73,
			rotation = 90,
			position = "center",
			xOffset = 0.23,
			yOffset = 0.40,
			size = 22,
			text = "@ saravana",
			font = {
				name = "SourceSansPro",
				bold = true,
				italic = true,
			},
			shadow = {
				enabled = "off"
			},
			rgbColor = { 0.4, 0.33, 0.0 },
			activeWatermarkType = "text"
		}

		local wmId

		rxTest:waitFor( callFunctionOrSubmodel(
					rxTest, wfCatalog,
					'WFCatWatermark',
					'createWatermark', wfCatalog, name, settings )
				:tap( function( watermarkId )
					-- capture the watermark id returned, it will be used for validation later.
					wmId = watermarkId
				end ) )

		-- wrapper for validating the received watermarkInfo
		local expectedSettings = settings

		expectedSettings.id = wmId
		expectedSettings.displayName = name


		rxTest:expectSingleValue( callFunctionOrSubmodel(
					rxTest, wfCatalog,
					'WFCatWatermark',
					'getWatermarkInfo', wfCatalog, wmId )
			:map( function( watermarkInfo )

				watermarkInfo = removeUserCreatedAndUpdatedField( watermarkInfo )
				return KSTableUtils.deepEquals( expectedSettings, watermarkInfo )

			end ), true )

		-- delete the watermark
		local watermarkIdsArray = { wmId,}

		rxTest:expectSingleValue( WFCatWatermark.deleteWatermarks( wfCatalog, watermarkIdsArray ), true )

		-- Verify doc is deleted
		rxTest:expectSingleValue( wfCatalog:docStore():docValue( wmId )
			:pluck( '_deleted' ), true )

		-- should return empty watermark info
		local emptyTbl = {}
		rxTest:expectSingleValue( callFunctionOrSubmodel(
					rxTest, wfCatalog,
					'WFCatWatermark',
					'getWatermarkInfo', wfCatalog, wmId )
			:map( function( watermarkInfo )
				return KSTableUtils.deepEquals( emptyTbl, watermarkInfo )
			end ), true )

	end

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

	function WFCatWatermarkTest.rxTestDeleteMultipleWatermarkTest( rxTest )

		rxTest:setTimeLimit( 30 )

		local wfCatalog = WFCatMockCatalog.createFakeWFCatalog( rxTest )

		--- Insert First Doc
		local name1 = "TEST-WATERMARK-1"
		local settings1 = {
			opacity = 19,
			rotation = 79,
			position = "center",
			xOffset = 0.23,
			yOffset = 0.40,
			size = 22,
			text = "@ saravana",
			font = {
				name = "SourceSansPro",
				bold = true,
				italic = true,
			},
			shadow = {
				enabled = "off" -- [ off, simple ]
			},
			rgbColor = { 0.4, 0.33, 0.0 },
			activeWatermarkType = "text",
		}

		local wmId1

		rxTest:waitFor( callFunctionOrSubmodel(
					rxTest, wfCatalog,
					'WFCatWatermark',
					'createWatermark', wfCatalog, name1, settings1 )
				:tap( function( watermarkId )
					-- capture the watermark id returned, it will be used for validation later.
					wmId1 = watermarkId
			end ) )

		--- Insert Second Doc
		local name2 = "TEST-WATERMARK-2"
		local settings2 = {
			opacity = 20,
			rotation = 80,
			position = "center",
			xOffset = 0.23,
			yOffset = 0.40,
			size = 22,
			text = "@ saravana",
			font = {
				name = "SourceSansPro",
				bold = true,
				italic = true,
			},
			shadow = {
				enabled = "off" -- [ off, simple ]
			},
			rgbColor = { 0.4, 0.33, 0.0 },
			activeWatermarkType = "text",
		}

		local wmId2

		rxTest:waitFor( callFunctionOrSubmodel(
					rxTest, wfCatalog,
					'WFCatWatermark',
					'createWatermark', wfCatalog, name2, settings2 )
				:tap( function( watermarkId )
					-- capture the watermark id returned, it will be used for validation later.
					wmId2 = watermarkId
				end ) )

		local watermarkIdsArray = { wmId1, wmId2 }

		-- check the quantity - it should be equal to the number of watermark added
		rxTest:expectSingleValue( WFCatWatermark.getAllWatermarksInfo( wfCatalog )
		:take( 1 )
		:map( function( watermarksInfoList )

			if( #watermarkIdsArray ~= #watermarksInfoList ) then
				return false
			end
			return true

		end ), true )

		-- now delete both the watermarks
		rxTest:expectSingleValue( WFCatWatermark.deleteWatermarks( wfCatalog, watermarkIdsArray ), true )

		-- Verify doc1 is deleted
		rxTest:expectSingleValue( wfCatalog:docStore():docValue( wmId1 )
			:pluck( '_deleted' ), true )

		-- Verify doc2 is deleted
		rxTest:expectSingleValue( wfCatalog:docStore():docValue( wmId2 )
			:pluck( '_deleted' ), true )

		-- check the quantity again, it should return empty watermark info as both the watermarks got deleted
		local emptyTbl = {}
		rxTest:expectSingleValue( WFCatWatermark.getAllWatermarksInfo( wfCatalog )
			:take(1)
			:map( function( watermarkInfo )
				return KSTableUtils.deepEquals( emptyTbl, watermarkInfo )
			end ), true )

	end

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

	function WFCatWatermarkTest.rxTestGetWatermarkInfoInvalidWatermarkIdTest( rxTest )

		rxTest:setTimeLimit( 30 )
		local wfCatalog = WFCatMockCatalog.createFakeWFCatalog( rxTest )

		-- Invalid Watermark id
		local wmId = "2192fe70fee04eadb6e7e68aaeb981ea"

		local emptyTbl = {}
		rxTest:expectSingleValue( callFunctionOrSubmodel(
					rxTest, wfCatalog,
					'WFCatWatermark',
					'getWatermarkInfo', wfCatalog, wmId )
			:map( function( watermarkInfo )

				return KSTableUtils.deepEquals( emptyTbl, watermarkInfo )

			end ), true )

	end

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

	function WFCatWatermarkTest.rxTestGetAllWatermarksInfoEmptyTest( rxTest )

		rxTest:setTimeLimit( 30 )

		local wfCatalog = WFCatMockCatalog.createFakeWFCatalog( rxTest )

		local emptyTbl = {}
		rxTest:expectSingleValue( WFCatWatermark.getAllWatermarksInfo( wfCatalog )
			:take(1)
			:map( function( watermarkInfo )

				return KSTableUtils.deepEquals( emptyTbl, watermarkInfo )

			end ), true )

	end

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

	function WFCatWatermarkTest.rxTestTruncateNameLengthExceeds250chars( rxTest )

		rxTest:setTimeLimit( 30 )

		local wfCatalog = WFCatMockCatalog.createFakeWFCatalog( rxTest )


		local longName = [[my very first attempt to write really, really super long watermark name
						so that it does't fit into the required max limit and freely over flows
						like a river, thus enables a great opportunity to invoke my truncate string
						length function I wont be part of the trucated string]]


		local settings = {
			opacity = 73,
			rotation = 90,
			position = "center",
			xOffset = 0.23,
			yOffset = 0.40,
			size = 22,
			text = "@saravana",
			font = {
				name = "SourceSansPro",
				bold = true,
				italic = true,
			},
			shadow = {
				enabled = "off"
			},
			rgbColor = { 0.4, 0.33, 0.0 },
			activeWatermarkType = "text",
		}

		local wmId

		rxTest:waitFor( callFunctionOrSubmodel(
					rxTest, wfCatalog,
					'WFCatWatermark',
					'createWatermark', wfCatalog, longName, settings )
				:tap( function( watermarkId )
					-- capture the watermark id returned, it will be used for validation later.
					wmId = watermarkId
				end ) )

		-- wrapper for validating the received watermarkInfo
		local expectedSettings = settings

		-- truncate the expected string to max limit of 250
		local truncatedName = string.sub(longName, 1, 250)

		expectedSettings.id = wmId
		expectedSettings.displayName = truncatedName



		rxTest:expectSingleValue( callFunctionOrSubmodel(
					rxTest, wfCatalog,
					'WFCatWatermark',
					'getWatermarkInfo', wfCatalog, wmId )
			:map( function( watermarkInfo )

				watermarkInfo = removeUserCreatedAndUpdatedField( watermarkInfo )
				return KSTableUtils.deepEquals( expectedSettings, watermarkInfo )

			end ), true )

	end

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

	function WFCatWatermarkTest.rxTestVerifyDuplicateWatermarkNameGetsRejected( rxTest )

		rxTest:setTimeLimit( 30 )

		local wfCatalog = WFCatMockCatalog.createFakeWFCatalog( rxTest )

		local name = "I claim that I am the original"

		local settings = getWatermarkTestSetttings( )

		local wmId

		rxTest:waitFor( callFunctionOrSubmodel(
					rxTest, wfCatalog,
					'WFCatWatermark',
					'createWatermark', wfCatalog, name, settings )
				:tap( function( watermarkId )
					-- capture the watermark id returned, it will be used for validation later.
					wmId = watermarkId
				end ) )

		-- wrapper for validating the received watermarkInfo
		local expectedSettings = settings

		expectedSettings.id = wmId
		expectedSettings.displayName = name


		rxTest:expectSingleValue( callFunctionOrSubmodel(
					rxTest, wfCatalog,
					'WFCatWatermark',
					'getWatermarkInfo', wfCatalog, wmId )
			:map( function( watermarkInfo )

				watermarkInfo = removeUserCreatedAndUpdatedField( watermarkInfo )
				-- verify watermark is created successfully with the given name
				return KSTableUtils.deepEquals( expectedSettings, watermarkInfo )

			end ), true )

		-- Now again, try to create another watermark with the same name.
		-- since it will result into duplicate name, Expect to receive error.
		local errorMessage = "Invalid argument, duplicate watermark name or Id received"

		rxTest:expectOnlyError(  callFunctionOrSubmodel(
					rxTest, wfCatalog,
					'WFCatWatermark',
					'createWatermark',  wfCatalog, name, settings ), errorMessage )

		-- now create another watermark with different name and this time there shouldn't be any error message
		name = "I dropped claiming that I am the original, will you allow me now"

		rxTest:waitFor(
			 callFunctionOrSubmodel(
					rxTest, wfCatalog,
					'WFCatWatermark',
					'createWatermark',  wfCatalog, name, settings )
				:tap( function( watermarkId )
					-- capture the watermark id returned, it will be used for validation later.
					wmId = watermarkId
				end ) )

		-- wrapper for validating the received watermarkInfo
		expectedSettings = settings

		expectedSettings.id = wmId
		expectedSettings.displayName = name

		rxTest:expectSingleValue( callFunctionOrSubmodel(
					rxTest, wfCatalog,
					'WFCatWatermark',
					'getWatermarkInfo', wfCatalog, wmId )
			:map( function( watermarkInfo )

				watermarkInfo = removeUserCreatedAndUpdatedField( watermarkInfo )
				-- verify watermark is created successfully with the given name
				return KSTableUtils.deepEquals( expectedSettings, watermarkInfo )

			end ), true )

	end

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

	function WFCatWatermarkTest.rxTestVerifyDuplicateWatermarkIdGetsRejected( rxTest )

		--- This test case verifies incase client passes UUID and if exist already WF rejects it.
		--  This test case also verifies that if client passes a valid UUID( doesn't exist already ) then
		--  it uses the UUID supplied by client and avoid generating internally.

		rxTest:setTimeLimit( 30 )

		local wfCatalog = WFCatMockCatalog.createFakeWFCatalog( rxTest )

		local name = "I claim that I am the original"

		local settings = getWatermarkTestSetttings( )

		local wmId

		-- generate UUID: mimic supply from client
		local optId = KSIdGenerator.uuidInLowercaseHex()
		rxTest:waitFor(
			 callFunctionOrSubmodel(
					rxTest, wfCatalog,
					'WFCatWatermark',
					'createWatermark',  wfCatalog, name, settings, nil, nil, optId )
				:tap( function( watermarkId )
					-- capture the watermark id returned, it will be used for validation later.
					wmId = watermarkId
				end ) )

		-- wrapper for validating the received watermarkInfo
		local expectedSettings = settings

		expectedSettings.id = wmId
		expectedSettings.displayName = name

		rxTest:expectSingleValue( callFunctionOrSubmodel(
					rxTest, wfCatalog,
					'WFCatWatermark',
					'getWatermarkInfo', wfCatalog, wmId )
			:map( function( watermarkInfo )

				return watermarkInfo.id == optId -- <-- verify it is same as supplied by client

			end ), true )

		-- Now again, try to create another watermark with the same id.
		-- since it will result into duplicate id, Expect to receive error.
		-- make sure the error is not due to duplicate name.
		-- so change the name:
		name = "I will give a way to validate Id"
		local errorMessage = "Invalid argument, duplicate watermark name or Id received"

		rxTest:expectOnlyError(  callFunctionOrSubmodel(
					rxTest, wfCatalog,
					'WFCatWatermark',
					'createWatermark',  wfCatalog, name, settings, nil, nil, optId ), errorMessage )

		-- now create another watermark with different id(freshly generate it)
		-- generate new UUID: mimic supply from client
		local optId_new = KSIdGenerator.uuidInLowercaseHex()

		-- make sure to change the name, so that we won't run into duplicate name issue.
		name = "I am fresh along with the id, will you allow me now"

		rxTest:waitFor(
			 callFunctionOrSubmodel(
					rxTest, wfCatalog,
					'WFCatWatermark',
					'createWatermark',  wfCatalog, name, settings, nil, nil, optId_new )
				:tap( function( watermarkId )
					-- capture the watermark id returned, it will be used for validation later.
					wmId = watermarkId
				end ) )

		rxTest:expectSingleValue( callFunctionOrSubmodel(
					rxTest, wfCatalog,
					'WFCatWatermark',
					'getWatermarkInfo', wfCatalog, wmId )
			:map( function( watermarkInfo )

				return watermarkInfo.id == optId_new -- <-- verify it is same as supplied by client

			end ), true )

	end

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

	function WFCatWatermarkTest.rxTestVerifyModifiedArraysAreOverwrittenCompletely( rxTest )

		rxTest:setTimeLimit( 30 )

		local base = {
			opacity = 73,
			rotation = 90,
			position = "center",
			xOffset = 0.23,
			yOffset = 0.40,
			size = 22,
			text = "@ saravana",
			font = {
				name = "SourceSansPro",
				bold = true,
				italic = true,
			},
			shadow = {
				enabled = "off"
			},
			rgbColor = { 0.11, 0.22, 0.33 },
			activeWatermarkType = "text",
		}

		-- modify first value of rgbColor
		local modified = {
			opacity = 73,
			rotation = 90,
			position = "center",
			xOffset = 0.23,
			yOffset = 0.40,
			size = 22,
			text = "@ saravana",
			font = {
				name = "SourceSansPro",
				bold = true,
				italic = true,
			},
			shadow = {
				enabled = "off"
			},
			rgbColor = { 0.77, 0.22, 0.33 }, -- first value modified from 0.11 to 0.77
			activeWatermarkType = "text",
		}

		local result1 = {
			rgbColor = { 0.77, 0.22, 0.33 }, -- expect the complete array and not just the single element
		}

		rxTest:expectEqual( WFTableMerge.createDiffForOz(base, modified ), result1 )

		-- modify Second Value of rgbColor
		modified.rgbColor[2] = 0.88

		local result2 = {
			rgbColor = { 0.77, 0.88, 0.33 }, -- expect the complete array and not just the single element
		}
		rxTest:expectEqual( WFTableMerge.createDiffForOz(base, modified ), result2 )

		-- modify third Value of rgbColor
		modified.rgbColor[3] = 0.99

		local result3 = {
			rgbColor = { 0.77, 0.88, 0.99 }, -- expect the complete array and not just the single element
		}
		rxTest:expectEqual( WFTableMerge.createDiffForOz(base, modified ), result3 )

		-- modify All values of rgbColor
		modified.rgbColor = { 0.1, 0.2, 0.3 }

		local resultAll = {
			rgbColor = { 0.1, 0.2, 0.3 }, -- expect the complete array and not just the single element
		}
		rxTest:expectEqual( WFTableMerge.createDiffForOz(base, modified ), resultAll )

		-- modify along with others
		modified.font.name = "Arial"
		modified.rgbColor[1] = 0.5

		local resultAlongWithOthers = {
			font = {
				name = "Arial"
			},
			rgbColor = { 0.5, 0.2, 0.3 }, -- expect the complete array and not just the single element
		}
		rxTest:expectEqual( WFTableMerge.createDiffForOz(base, modified ), resultAlongWithOthers )

	end

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

	function WFCatWatermarkTest.rxTestVerifyGetAllWatermarksInfoFiring( rxTest )

		rxTest:setTimeLimit( 30 )

		local wfCatalog = WFCatMockCatalog.createFakeWFCatalog( rxTest )

		local watermarkInfoArray = {}

		local observer = rxTest:observe(
			WFCatWatermark.getAllWatermarksInfo( wfCatalog )
				:map( function( watermarksInfoList )
					for i, watermarkInfo in ipairs( watermarksInfoList ) do
						watermarksInfoList[i] = removeUserCreatedAndUpdatedField( watermarkInfo )
					end

				return watermarksInfoList

			end ), true )
		-- This is the begining, expect the empty table
		observer:expectDataWithinTime( 3, watermarkInfoArray )

		local name1 = "watermark-1"
		local settings1 = {
			opacity = 73,
			rotation = 90,
			position = "center",
			xOffset = 0.23,
			yOffset = 0.40,
			size = 22,
			text = "@ saravana",
			font = {
				name = "SourceSansPro",
				bold = true,
				italic = true,
			},
			shadow = {
				enabled = "off" -- [ off, simple ]
			},
			rgbColor = { 0.5, 0.6, 0.7 },
			activeWatermarkType = "text",
		}

		local wmId1

		rxTest:waitFor(
			 callFunctionOrSubmodel(
					rxTest, wfCatalog,
					'WFCatWatermark',
					'createWatermark',  wfCatalog, name1, settings1 )
				:tap( function( watermarkId )
					-- capture the watermark id returned, it will be used for validation later.
					wmId1 = watermarkId
				end ) )

		-- wrappers for comparing test output
		local expectedSettings1 = settings1
		expectedSettings1.id = wmId1
		expectedSettings1.displayName = name1

		local expectedWatermarkSettingsArray1 = { expectedSettings1, }

		-- expect the firing should happen for the 1st created watermark
		observer:expectDataWithinTime( 3, expectedWatermarkSettingsArray1 )

		-- just update a single filed in the watermark settings.
		-- do not update name
		local updatedSettings = settings1
		updatedSettings.size = 30

		rxTest:waitFor(  callFunctionOrSubmodel(
					rxTest, wfCatalog,
					'WFCatWatermark',
					'updateWatermark', wfCatalog, wmId1, name1, updatedSettings ) )

		-- wrappers for comparing test output
		local expectedUpdatedSettings = expectedSettings1
		expectedUpdatedSettings.size = 30
		expectedUpdatedSettings = removeUserCreatedAndUpdatedField( expectedUpdatedSettings )

		local expectedUpdatedWatermarkSettingsArray = { expectedUpdatedSettings, }

		-- expect the firing should happen for the updated watermark
		observer:expectDataWithinTime( 3, expectedUpdatedWatermarkSettingsArray )

		-- Extension of TC to cover creation of another watermark
		local name2 = "watermark-2"
		local settings2 = {
			opacity = 73,
			rotation = 90,
			position = "center",
			xOffset = 0.23,
			yOffset = 0.40,
			size = 22,
			text = "@ saravana",
			font = {
				name = "SourceSansPro",
				bold = true,
				italic = true,
			},
			shadow = {
				enabled = "off" -- [ off, simple ]
			},
			rgbColor = { 0.5, 0.6, 0.7 },
			activeWatermarkType = "text",
		}

		local wmId2

		rxTest:waitFor(
			 callFunctionOrSubmodel(
					rxTest, wfCatalog,
					'WFCatWatermark',
					'createWatermark',  wfCatalog, name2, settings2 )
				:tap( function( watermarkId )
					-- capture the watermark id returned, it will be used for validation later.
					wmId2 = watermarkId
				end ) )

		-- wrappers for comparing test output
		local expectedSettings2 = settings2
		expectedSettings2.id = wmId2
		expectedSettings2.displayName = name2

		-- now expect two watermarks
		local expectedWatermarkSettingsArray2 = { expectedSettings1, expectedSettings2, }

		-- expect the firing should happen for the 2nd created watermark
		observer:expectDataWithinTime( 3, expectedWatermarkSettingsArray2 )

		-- Extend the TC to delete watermark
		-- delete the first watermark
		local watermarkIdsArray = { wmId1, }
		rxTest:expectSingleValue( WFCatWatermark.deleteWatermarks( wfCatalog, watermarkIdsArray ), true )

		-- since first watermark deleted expect the watermark info only for the second watermark
		expectedWatermarkSettingsArray2 = { expectedSettings2 }
		observer:expectDataWithinTime( 3, expectedWatermarkSettingsArray2 )

		watermarkIdsArray = { wmId2, }

		-- delete the second watermark as well
		rxTest:expectSingleValue( WFCatWatermark.deleteWatermarks( wfCatalog, watermarkIdsArray ), true )

		-- expect empty table
		observer:expectDataWithinTime( 3, {} )

		observer:dispose()

	end

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

	function WFCatWatermarkTest.rxTestVerifyValidImageWatermarkFilePathStoredInLocalOnly( rxTest )

	-- This test case verifies whether WF stores the file path and
	-- its derived attributes such as file size, sha256 in a _localOnly table inside the watermark doc
	-- when create watermark API being invoked with an optFilePath(valid path) argument

		rxTest:setTimeLimit( 30 )

		local wfCatalog = WFCatMockCatalog.createFakeWFCatalog( rxTest )

		local name = "My First Image Watermark"

		local settings = getWatermarkTestSetttings()

		local wmId

		local testDataPath = KSPathUtils.getStandardFilePath( "testdata" )

		local relativePath = KSPathUtils.child( 'assets', '1.5MB.jpg' )

		local absolutePath = KSPathUtils.child( testDataPath, relativePath )

		local fileSize = absolutePath and WFFileUtils.fileAttributes( absolutePath ).fileSize


		rxTest:waitFor(
			 callFunctionOrSubmodel(
					rxTest, wfCatalog,
					'WFCatWatermark',
					'createWatermark',  wfCatalog, name, settings, relativePath, '1.5MB.jpg' )
				:tap( function( watermarkId )
					-- capture the watermark id returned, it will be used for validation later.
					wmId = watermarkId
				end ) )

		rxTest:expectSingleValue( wfCatalog:docStore():docValue( wmId )
			:map( function( watermarkDoc )
				local master = derefOpt( watermarkDoc, '_localOnly', 'universal', 'files', 'original', 'master' )
				return master.relativePath == relativePath and master.fileSize == fileSize
			end ), true )

	end

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

	function WFCatWatermarkTest.rxTestVerifyInvalidFilePathRejected( rxTest )

			rxTest:setTimeLimit( 30 )

			-- check during create watermark
			local wfCatalog = WFCatMockCatalog.createFakeWFCatalog( rxTest )
			local name = "My First Image Watermark"
			local settings = getWatermarkTestSetttings()
			settings.activeWatermarkType = "image"

			local relativePath = KSPathUtils.child( 'invalid', '1.5MB.jpg' ) -- make the path invalid

			rxTest:expectOnlyError(  callFunctionOrSubmodel(
					rxTest, wfCatalog,
					'WFCatWatermark',
					'createWatermark',  wfCatalog, name, settings,
				relativePath, '1.5MB.jpg' ), "File does not exist" )

			relativePath = KSPathUtils.child( 'assets', '1.5MB.jpg' ) -- make the path valid
			local wmId

			rxTest:waitFor(
				 callFunctionOrSubmodel(
					rxTest, wfCatalog,
					'WFCatWatermark',
					'createWatermark',  wfCatalog, name, settings, relativePath, '1.5MB.jpg' )
					:tap( function( watermarkId )
						-- capture the watermark id returned, it will be used for validation later.
						wmId = watermarkId
					end ) )

			-- check during update watermark
			relativePath = KSPathUtils.child( 'invalid', '1.5MB.jpg' ) -- make the path invalid again

			rxTest:expectOnlyError(  callFunctionOrSubmodel(
					rxTest, wfCatalog,
					'WFCatWatermark',
					'updateWatermark', wfCatalog, wmId, name, settings,
				relativePath, '1.5MB.jpg' ), "File does not exist" )

		end

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

	function WFCatWatermarkTest.rxTestVerifyImageWatermarkFileSizeGreaterThan5MBRejected( rxTest )

			-- Todo: The max File size restriction to 5MB is temporary and
			-- when Oz supports chunked upload this test case needs to be removed.

			rxTest:setTimeLimit( 30 )
			local maxSizeInBytes = 5 * 1024 * 1024
			local wfCatalog = WFCatMockCatalog.createFakeWFCatalog( rxTest )
			local name = "My First Image Watermark"
			local settings = getWatermarkTestSetttings()
			settings.activeWatermarkType = "image"
			local testDataPath = KSPathUtils.getStandardFilePath( "testdata" )
			local relativePath = KSPathUtils.child('assets', '12MB.jpg' )
			local optBinaryUploadSrcFilePath = KSPathUtils.child( testDataPath, relativePath ) -- size > 5MB
			local fileSize = WFFileUtils.fileAttributes( optBinaryUploadSrcFilePath ).fileSize
			rxTest:expectOnlyError(  callFunctionOrSubmodel(
					rxTest, wfCatalog,
					'WFCatWatermark',
					'createWatermark',  wfCatalog, name, settings, relativePath, '12MB.jpg' ),
					"Image watermark file size in bytes( " .. fileSize .. " ) must not exceed " .. maxSizeInBytes )


			relativePath = KSPathUtils.child( 'assets', '1.5MB.jpg' ) -- make the path valid
			local wmId

			rxTest:waitFor(
				 callFunctionOrSubmodel(
					rxTest, wfCatalog,
					'WFCatWatermark',
					'createWatermark',  wfCatalog, name, settings, relativePath, '1.5MB.jpg' )
					:tap( function( watermarkId )
						-- capture the watermark id returned, it will be used for validation later.
						wmId = watermarkId
					end ) )

			-- check during update watermark
			relativePath = KSPathUtils.child('assets', '12MB.jpg' ) -- make the path greater than 5MB

			fileSize = WFFileUtils.fileAttributes( optBinaryUploadSrcFilePath ).fileSize
			rxTest:expectOnlyError(  callFunctionOrSubmodel(
					rxTest, wfCatalog,
					'WFCatWatermark',
					'updateWatermark', wfCatalog, wmId, name, settings, relativePath, '12MB.jpg'  ),
					"Image watermark file size in bytes( " .. fileSize .. " ) must not exceed " .. maxSizeInBytes )

		end

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

	function WFCatWatermarkTest.rxTestVerifyAutomaticPayloadValidationErrorCorrection_testNightly( rxTest )

		rxTest:setTimeLimit( 120 )

		local catalogTestSession = WFCatTestSession( rxTest )

		local catalog
		--local assetId = KSIdGenerator.uuidInLowercaseHex()

		rxTest:expectSingleValue(

			catalogTestSession:catalog()
				:tap( function( _catalog )
					catalog = _catalog
				end ),

			':ignore:' )

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

		local errorHandler = WFCatErrorHandler( catalog, {
			deleteAssets = nil,
			deleteIndividualAlbum = nil },
			{ debugMode = true } )

		rxTest:finally( function()
			errorHandler:close()
			puller:close()
			pusher:close()
		end )

		-- watch the errors table, to ensure we actually do get errors
		local outstandingErrors = rxTest:observe( catalog:indexStore():query {
				sql = [[SELECT count(*) FROM errors]],
				quiesceUntilUpToDate = true,
			}
			:pluck( 1, 1 ) )

		-- start the sync processes
		puller:getUpdatesContinuously()
		pusher:sendUpdatesContinuously()

		local name = "My First Watermark"
		local settings = {
			opacity = 73,
			rotation = 90,
			position = "center",
			xOffset = 0.23,
			yOffset = 0.40,
			size = 500, --- set to out of bound, max value is 100
			text = "@ saravana",
			font = {
				name = "SourceSansPro",
				bold = true,
				italic = true,
			},
			shadow = {
				enabled = "off" -- [ off, simple ]
			},
			rgbColor = { 0.4, 0.33, 0.0 },
			activeWatermarkType = "text",
		}

		local wmId
		rxTest:waitFor(
			 callFunctionOrSubmodel(
					rxTest, catalog,
					'WFCatWatermark',
					'createWatermark',  catalog, name, settings )
				:tap( function( watermarkId )
					-- capture the watermark id returned, it will be used for validation later.
					wmId = watermarkId
				end ) )
		-- we should get an error from Oz
		outstandingErrors:expectDataWithinTime( 60, 1 )

		-- the error handler should eventually clear the error
		outstandingErrors:expectDataWithinTime( 10, 0 )

		-- Verify watermark was uploaded
		local assetOnOz = rxTest:observe(
			catalog:docStore():ongoingDocValue( wmId )
				:map( function( docValue )
					return docValue._serverProvidedMetadata ~= nil
				end ) )

		assetOnOz:expectDataWithinTime( 40, true )

	end

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

	function WFCatWatermarkTest.rxTestVerifyAbsenceOfActiveWatermarkTypeShallResultInError( rxTest )

		rxTest:setTimeLimit( 30 )

		local wfCatalog = WFCatMockCatalog.createFakeWFCatalog( rxTest )

		local name = "VerifyAbsenceOfActiveWatermark"
		local settings = getWatermarkTestSetttings()
		settings["activeWatermarkType"] = nil -- make activeWatermarkType absent

		local errorMessage = "activeWatermarkType shall not be nil"
		rxTest:expectOnlyError(  callFunctionOrSubmodel(
					rxTest, wfCatalog,
					'WFCatWatermark',
					'createWatermark',  wfCatalog, name, settings ), errorMessage )

	end

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

	function WFCatWatermarkTest.rxTestVerifyIncorrectActiveWatermarkTypeShallResultInError( rxTest )

		rxTest:setTimeLimit( 30 )

		local wfCatalog = WFCatMockCatalog.createFakeWFCatalog( rxTest )
		local name = "VerifyIncorrectActiveWatermarkType"
		local settings = getWatermarkTestSetttings()
		settings.activeWatermarkType = "Invalid Value" -- make activeWatermarkType wrong

		local errorMessage = "Unknown activeWatermarkType received, " .. settings.activeWatermarkType
		rxTest:expectOnlyError(  callFunctionOrSubmodel(
					rxTest, wfCatalog,
					'WFCatWatermark',
					'createWatermark',  wfCatalog, name, settings ), errorMessage )

	end

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

	function WFCatWatermarkTest.rxTestVerifyEmptyTextFieldShallReturnError( rxTest )

		rxTest:setTimeLimit( 30 )

		local wfCatalog = WFCatMockCatalog.createFakeWFCatalog( rxTest )
		local name = "VerifyEmptyTextField"
		local settings = getWatermarkTestSetttings()
		settings.text = "" -- make it empty string

		local errorMessage = "Text field, if it is present, then it shall not be empty"
		rxTest:expectOnlyError(  callFunctionOrSubmodel(
					rxTest, wfCatalog,
					'WFCatWatermark',
					'createWatermark',  wfCatalog, name, settings ), errorMessage )

	end

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

	function WFCatWatermarkTest.rxTestVerifyUpdateWatermarkImageToImage( rxTest )

	-- This test case verifies whether clients able to successfully update the existing image watermark

		rxTest:setTimeLimit( 30 )

		local wfCatalog = WFCatMockCatalog.createFakeWFCatalog( rxTest )

		-- create an image watermark
		local name = "VerifyUpdateWatermarkImage"
		local settings = getWatermarkTestSetttings()
		settings.activeWatermarkType = "image"
		local wmId
		local testDataPath = KSPathUtils.getStandardFilePath( "testdata" )
		local relativePath = KSPathUtils.child( 'assets', 'IMG_7706.PNG' )
		local absolutePath = KSPathUtils.child( testDataPath, relativePath )
		local fileName = 'IMG_7706.PNG'
		local sha256 = WFCrypto.digestFile( absolutePath, 'SHA256' )
		local fileSize = absolutePath and WFFileUtils.fileAttributes( absolutePath ).fileSize

		rxTest:waitFor(
			 callFunctionOrSubmodel(
					rxTest, wfCatalog,
					'WFCatWatermark',
					'createWatermark',  wfCatalog, name, settings, relativePath, fileName )
				:tap( function( watermarkId )
					-- capture the watermark id returned, it will be used for validation later.
					wmId = watermarkId
				end ) )

		-- expect localOnly created
		rxTest:expectSingleValue( wfCatalog:docStore():docValue( wmId )
			:map( function( watermarkDoc )
				local master = derefOpt( watermarkDoc, '_localOnly', 'universal', 'files', 'original', 'master' )
				return master.relativePath == relativePath and master.fileSize == fileSize
			end ), true )

		-- expect importSource created
		--  query getWatermarkInfo and check importSource correctly included in the watermarkInfo
		rxTest:expectSingleValue( callFunctionOrSubmodel(
					rxTest, wfCatalog,
					'WFCatWatermark',
					'getWatermarkInfo', wfCatalog, wmId )
			:map( function( watermarkInfo )

				return( watermarkInfo.importSource.fileName == fileName and
						watermarkInfo.importSource.sha256 == sha256 and
						watermarkInfo.importSource.fileSize == fileSize and
						watermarkInfo.importSource.importTimestamp == watermarkInfo.userCreated )

		end ), true )

		-- update the image watermark
		relativePath = KSPathUtils.child( 'assets', '1.5MB.jpg' ) -- change to different image
		absolutePath = KSPathUtils.child( testDataPath, relativePath )
		fileSize = absolutePath and WFFileUtils.fileAttributes( absolutePath ).fileSize
		sha256 = WFCrypto.digestFile( absolutePath, 'SHA256' )
		fileName = '1.5MB.jpg'

		rxTest:waitFor(  callFunctionOrSubmodel(
					rxTest, wfCatalog,
					'WFCatWatermark',
					'updateWatermark', wfCatalog, wmId, name, settings, relativePath, fileName ) )

		-- expect localOnly updated
		rxTest:expectSingleValue( wfCatalog:docStore():docValue( wmId )
			:map( function( watermarkDoc )
				local master = derefOpt( watermarkDoc, '_localOnly', 'universal', 'files', 'original', 'master' )
				return master.relativePath == relativePath and master.fileSize == fileSize and sha256 == master.sha256
			end ), true )

		-- expect importSource updated
		--  query getWatermarkInfo and check importSource correctly included in the watermarkInfo
		rxTest:expectSingleValue( callFunctionOrSubmodel(
					rxTest, wfCatalog,
					'WFCatWatermark',
					'getWatermarkInfo', wfCatalog, wmId )
			:map( function( watermarkInfo )

				return( watermarkInfo.importSource.fileName == fileName and
						watermarkInfo.importSource.sha256 == sha256 and
						watermarkInfo.importSource.fileSize == fileSize and
						watermarkInfo.importSource.importTimestamp == watermarkInfo.userCreated )

		end ), true )

	end

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

	function WFCatWatermarkTest.rxTestVerifyUpdateWatermarkImageToText( rxTest )

	-- This test case verifies whether clients able to successfully update the existing image watermark

		rxTest:setTimeLimit( 30 )

		local wfCatalog = WFCatMockCatalog.createFakeWFCatalog( rxTest )

		-- create an image watermark
		local name = "VerifyUpdateWatermarkImageToText"
		local settings = getWatermarkTestSetttings()
		settings.activeWatermarkType = "image"
		settings["text"] = nil
		settings["font"] = nil
		local wmId

		local testDataPath = KSPathUtils.getStandardFilePath( "testdata" )
		local relativePath = KSPathUtils.child( 'assets', '1.5MB.jpg' )
		local absolutePath = KSPathUtils.child( testDataPath, relativePath )
		local fileSize = absolutePath and WFFileUtils.fileAttributes( absolutePath ).fileSize
		local fileName = '1.5MB.jpg'
		local sha256 = WFCrypto.digestFile( absolutePath, 'SHA256' )

		rxTest:waitFor(
			 callFunctionOrSubmodel(
					rxTest, wfCatalog,
					'WFCatWatermark',
					'createWatermark',  wfCatalog, name, settings, relativePath, fileName )
				:tap( function( watermarkId )
					-- capture the watermark id returned, it will be used for validation later.
					wmId = watermarkId
				end ) )

		-- expect localOnly created
		rxTest:expectSingleValue( wfCatalog:docStore():docValue( wmId )
			:map( function( watermarkDoc )
				local master = derefOpt( watermarkDoc, '_localOnly', 'universal', 'files', 'original', 'master' )

				return ( master.relativePath == relativePath and master.fileSize == fileSize and
					watermarkDoc.text == nil and watermarkDoc.font == nil )
			end ), true )

		-- expect importSource created
		--  query getWatermarkInfo and check importSource correctly included in the watermarkInfo
		local importSource
		rxTest:expectSingleValue( callFunctionOrSubmodel(
					rxTest, wfCatalog,
					'WFCatWatermark',
					'getWatermarkInfo', wfCatalog, wmId )
			:map( function( watermarkInfo )
				importSource = watermarkInfo.importSource
				return( watermarkInfo.importSource.fileName == fileName and
						watermarkInfo.importSource.sha256 == sha256 and
						watermarkInfo.importSource.fileSize == fileSize and
						watermarkInfo.importSource.importTimestamp == watermarkInfo.userCreated ) -- <-- verification point


		end ), true )

		-- update the image watermark to text
		settings.text = "@UserName"
		settings.font = {
			name = "SourceSansPro",
			bold = true,
			italic = true,
		}
		settings.size = 20
		settings.opacity = 70

		-- Also helps to verify activeWatermarkType included in the watermarkInfo
		settings.activeWatermarkType = "text"
		rxTest:waitFor(  callFunctionOrSubmodel(
					rxTest, wfCatalog,
					'WFCatWatermark',
					'updateWatermark', wfCatalog, wmId, name, settings ) )

		local expectedSettings = settings
		expectedSettings.displayName = name
		expectedSettings.id = wmId
		expectedSettings.filePath = relativePath
		expectedSettings.originalLinkAvailable = false

		-- make sure updating the object to text watermark shall not affect the previously created importSource
		expectedSettings.importSource = importSource

		rxTest:expectSingleValue( callFunctionOrSubmodel(
					rxTest, wfCatalog,
					'WFCatWatermark',
					'getWatermarkInfo', wfCatalog, wmId )
			:map( function( watermarkInfo )

				watermarkInfo = removeUserCreatedAndUpdatedField( watermarkInfo )
				return KSTableUtils.deepEquals( expectedSettings, watermarkInfo )

			end ), true )

	end

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

	function WFCatWatermarkTest.rxTestVerifyUpdateWatermarkTextToImage( rxTest )

	-- This test case verifies whether clients able to successfully update the existing image watermark

		rxTest:setTimeLimit( 30 )

		local wfCatalog = WFCatMockCatalog.createFakeWFCatalog( rxTest )

		-- create a text watermark
		local name = "VerifyUpdateWatermarkTextToImage"
		local settings = getWatermarkTestSetttings()
		local wmId

		rxTest:waitFor(
			 callFunctionOrSubmodel(
					rxTest, wfCatalog,
					'WFCatWatermark',
					'createWatermark',  wfCatalog, name, settings )
				:tap( function( watermarkId )
					-- capture the watermark id returned, it will be used for validation later.
					wmId = watermarkId
				end ) )


		-- expect no localOnly created.
		rxTest:expectSingleValue( wfCatalog:docStore():docValue( wmId )
			:map( function( watermarkDoc )
				local localOnly = derefOpt( watermarkDoc, '_localOnly' )
				return ( localOnly == nil )
			end ), true )


		local expectedSettings = settings
		expectedSettings.displayName = name
		expectedSettings.id = wmId
		rxTest:expectSingleValue( callFunctionOrSubmodel(
					rxTest, wfCatalog,
					'WFCatWatermark',
					'getWatermarkInfo', wfCatalog, wmId )
			:map( function( watermarkInfo )
				--- also check no importSource created
				watermarkInfo = removeUserCreatedAndUpdatedField( watermarkInfo )
				return ( not watermarkInfo.importSource  and KSTableUtils.deepEquals( expectedSettings, watermarkInfo ) )

			end ), true )

		-- update to Image watermark
		local testDataPath = KSPathUtils.getStandardFilePath( "testdata" )
		local relativePath = KSPathUtils.child( 'assets', '1.5MB.jpg' )
		local absolutePath = KSPathUtils.child( testDataPath, relativePath )
		local fileSize = absolutePath and WFFileUtils.fileAttributes( absolutePath ).fileSize
		local fileName = '1.5MB.jpg'
		local sha256 = WFCrypto.digestFile( absolutePath, 'SHA256' )
		settings.activeWatermarkType = "image"
		-- change other params as well
		settings.size = 20
		settings.opacity = 70

		rxTest:waitFor(  callFunctionOrSubmodel(
					rxTest, wfCatalog,
					'WFCatWatermark',
					'updateWatermark', wfCatalog, wmId, name, settings, relativePath, fileName ) )

		-- expect localOnly created.
		rxTest:expectSingleValue( wfCatalog:docStore():docValue( wmId )
			:map( function( watermarkDoc )
				local master = derefOpt( watermarkDoc, '_localOnly', 'universal', 'files', 'original', 'master' )

				return ( master.relativePath == relativePath and master.fileSize == fileSize )
			end ), true )

		local importSource
		-- expect importSource created now.
		--  query getWatermarkInfo and check importSource correctly included in the watermarkInfo
		rxTest:expectSingleValue( callFunctionOrSubmodel(
					rxTest, wfCatalog,
					'WFCatWatermark',
					'getWatermarkInfo', wfCatalog, wmId )
			:map( function( watermarkInfo )
				importSource = watermarkInfo.importSource
				return( watermarkInfo.importSource.fileName == fileName and
						watermarkInfo.importSource.sha256 == sha256 and
						watermarkInfo.importSource.fileSize == fileSize and
						watermarkInfo.importSource.importTimestamp == watermarkInfo.userCreated ) -- <-- verification point
		end ), true )

		expectedSettings.activeWatermarkType = "image"
		expectedSettings.size = 20
		expectedSettings.opacity = 70
		expectedSettings.filePath = relativePath
		expectedSettings.importSource = importSource
		expectedSettings.originalLinkAvailable = false

		-- Also verify updating to image watermark doesn't alter text and font fields or other common fields.
		rxTest:expectSingleValue( callFunctionOrSubmodel(
					rxTest, wfCatalog,
					'WFCatWatermark',
					'getWatermarkInfo', wfCatalog, wmId )
			:map( function( watermarkInfo )
				watermarkInfo = removeUserCreatedAndUpdatedField( watermarkInfo )
				return KSTableUtils.deepEquals( expectedSettings, watermarkInfo )
			end ), true )

	end

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

	function WFCatWatermarkTest.rxTestVerifyInsteadOfSubtypeActiveWatermarkTypeSentToClients( rxTest )

		-- This test case exclusively verifies activeWatermarkTye being sent
		-- to clients instead of subtype.

		rxTest:setTimeLimit( 30 )

		local wfCatalog = WFCatMockCatalog.createFakeWFCatalog( rxTest )
		local name = "VerifyInsteadOfSubTypeActiveWatermarkTypeSentToClients"
		local settings = getWatermarkTestSetttings()
		local wmId

		rxTest:waitFor(
			 callFunctionOrSubmodel(
					rxTest, wfCatalog,
					'WFCatWatermark',
					'createWatermark',  wfCatalog, name, settings )
				:tap( function( watermarkId )
					-- capture the watermark id returned, it will be used for validation later.
					wmId = watermarkId
				end ) )

		rxTest:expectSingleValue( callFunctionOrSubmodel(
					rxTest, wfCatalog,
					'WFCatWatermark',
					'getWatermarkInfo', wfCatalog, wmId )
			:map( function( watermarkInfo )

				return ( watermarkInfo.subtype == nil and watermarkInfo.activeWatermarkType == "text" )

			end ), true )

	end

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

	function WFCatWatermarkTest.rxTestVerifyFilePathBeingSentToClientsIfAvailable( rxTest )

		-- This test case verifies filePath being sent to client if available

		rxTest:setTimeLimit( 30 )

		local wfCatalog = WFCatMockCatalog.createFakeWFCatalog( rxTest )
		local name = "VerifyFilePathBeingSentToClientsIfAvailable"
		local settings = getWatermarkTestSetttings()
		local wmId

		local relativePath = KSPathUtils.child( 'assets', '1.5MB.jpg' )

		settings.activeWatermarkType = "image"

		rxTest:waitFor(
			 callFunctionOrSubmodel(
					rxTest, wfCatalog,
					'WFCatWatermark',
					'createWatermark',  wfCatalog, name, settings, relativePath, '1.5MB.jpg' )
				:tap( function( watermarkId )
					-- capture the watermark id returned, it will be used for validation later.
					wmId = watermarkId
				end ) )

		rxTest:expectSingleValue( callFunctionOrSubmodel(
					rxTest, wfCatalog,
					'WFCatWatermark',
					'getWatermarkInfo', wfCatalog, wmId )
			:map( function( watermarkInfo )

				return ( watermarkInfo.subtype == nil and
						watermarkInfo.activeWatermarkType == "image" and
						watermarkInfo.filePath == relativePath )

			end ), true )

	end

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

	function WFCatWatermarkTest.rxTestVerifyEmptySettingReceivedShouldResultInError( rxTest )

		rxTest:setTimeLimit( 30 )

		--- Verify Empty settings handling during create watermark
		local wfCatalog = WFCatMockCatalog.createFakeWFCatalog( rxTest )
		local name = "VerifyEmptySettingReceivedShouldResultInError"
		local settings = {}  -- set to empty table

		local errorMessage = "Invalid argument, empty settings received in create watermark"
		rxTest:expectOnlyError(  callFunctionOrSubmodel(
					rxTest, wfCatalog,
					'WFCatWatermark',
					'createWatermark',  wfCatalog, name, settings ), errorMessage )

		local wmId
		settings = getWatermarkTestSetttings()

		rxTest:waitFor(
			 callFunctionOrSubmodel(
					rxTest, wfCatalog,
					'WFCatWatermark',
					'createWatermark',  wfCatalog, name, settings )
				:tap( function( watermarkId )
					-- capture the watermark id returned, it will be used for validation later.
					wmId = watermarkId
				end ) )

		-- modify settings, set to empty table
		settings = {}

		--- Verify Empty settings handling during update watermark
		errorMessage = "Invalid argument, empty settings received in update watermark"
		rxTest:expectOnlyError(  callFunctionOrSubmodel(
					rxTest, wfCatalog,
					'WFCatWatermark',
					'updateWatermark', wfCatalog, wmId, name, settings ), errorMessage )

	end

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

	function WFCatWatermarkTest.rxTestVerifyInvalidActiveWatermarkTypeFallBackToText( rxTest )

		rxTest:setTimeLimit( 30 )

		local wfCatalog = WFCatMockCatalog.createFakeWFCatalog( rxTest )
		local name = "VerifyInvalidActiveWatermarkTypeFallsBackToText"
		local settings = getWatermarkTestSetttings()

		local wmId

		-- create watermark
		rxTest:waitFor(
			 callFunctionOrSubmodel(
					rxTest, wfCatalog,
					'WFCatWatermark',
					'createWatermark',  wfCatalog, name, settings )
				:tap( function( watermarkId )
					-- capture the watermark id returned, it will be used for validation later.
					wmId = watermarkId
				end ) )

		-- now corrupt the activeWatermarkType value to nil
		rxTest:expectSingleValue(
			wfCatalog:docStore():applyOverlay {
				_id = wmId,
				activeWatermarkType = { _delete = true },
			}, ':ignore:' )

		-- Ensure activeWatermarkType set to nil
		rxTest:expectSingleValue( wfCatalog:docStore():docValue( wmId )
			:map( function( watermarkDoc )
				local activeWatermarkType = watermarkDoc.activeWatermarkType
				return ( activeWatermarkType  == nil )
			end ), true )

		-- check when queried activeWatermarkType "fallback" to text
		rxTest:expectSingleValue( callFunctionOrSubmodel(
					rxTest, wfCatalog,
					'WFCatWatermark',
					'getWatermarkInfo', wfCatalog, wmId )
			:map( function( watermarkInfo )
				return ( watermarkInfo.activeWatermarkType == "text")
			end ), true )

		-- Again corrupt the activeWatermarkType value to invalid value
		rxTest:expectSingleValue(
			wfCatalog:docStore():applyOverlay {
				_id = wmId,
				activeWatermarkType = "BlaBlaBla"
			}, ':ignore:' )

		-- Ensure activeWatermarkType set to BlaBlaBla
		rxTest:expectSingleValue( wfCatalog:docStore():docValue( wmId )
			:map( function( watermarkDoc )
				local activeWatermarkType = watermarkDoc.activeWatermarkType
				return ( activeWatermarkType  == "BlaBlaBla" )
			end ), true )

		-- check when queried activeWatermarkType "fallback" to text
		rxTest:expectSingleValue( callFunctionOrSubmodel(
					rxTest, wfCatalog,
					'WFCatWatermark',
					'getWatermarkInfo', wfCatalog, wmId )
			:map( function( watermarkInfo )
				return ( watermarkInfo.activeWatermarkType == "text")
			end ), true )

	end

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

	function WFCatWatermarkTest.rxTestVerifyAllowOnlyPNGOrJPGImage( rxTest )

		-- This test case verifies WF will reject watermark image format
		-- other than PNG OR JPG while creating/updating watermark

		rxTest:setTimeLimit( 30 )

		local wfCatalog = WFCatMockCatalog.createFakeWFCatalog( rxTest )

		-- verify create watermark

		local name = "VerifyAllowOnlyPNGOrJPGImage"
		local settings = getWatermarkTestSetttings()
		settings.activeWatermarkType = "image"
		local wmId
		local testDataPath = KSPathUtils.getStandardFilePath( "testdata" )

		-- 1. Try to create watermark with invalid image format ( *.dng ) and expect error

		local relativePath =  KSPathUtils.child( 'assets', 'IMG_7706.DNG' )
		local errorMessage = "Unsupported image file extension, dng"
		rxTest:expectOnlyError(
			 callFunctionOrSubmodel(
					rxTest, wfCatalog,
					'WFCatWatermark',
					'createWatermark',  wfCatalog, name, settings, relativePath, 'IMG_7706.DNG' ), errorMessage )

		-- 2. choose supported type this time, select ( PNG ) and make sure it allows.

		relativePath = KSPathUtils.child( 'assets', 'IMG_7706.PNG' )
		local absolutePath = KSPathUtils.child( testDataPath, relativePath )
		local fileSize = absolutePath and WFFileUtils.fileAttributes( absolutePath ).fileSize
		local fileName = 'IMG_7706.PNG'

		rxTest:waitFor(
			 callFunctionOrSubmodel(
					rxTest, wfCatalog,
					'WFCatWatermark',
					'createWatermark',  wfCatalog, name, settings, relativePath, fileName )
				:tap( function( watermarkId )
					-- capture the watermark id returned, it will be used for validation later.
					wmId = watermarkId
				end ) )

		rxTest:expectSingleValue( wfCatalog:docStore():docValue( wmId )
			:map( function( watermarkDoc )
				local master = derefOpt( watermarkDoc, '_localOnly', 'universal', 'files', 'original', 'master' )
				return master.relativePath == relativePath and master.fileSize == fileSize
			end ), true )

		-- verify update watermark

		-- 3. Try to update with invalid watermark format ( *.mov ) and expect error

		relativePath = KSPathUtils.child( 'assets', '4.3MB.mov' )
		errorMessage = "Unsupported image file extension, mov"
		rxTest:expectOnlyError(
			 callFunctionOrSubmodel(
					rxTest, wfCatalog,
					'WFCatWatermark',
					'updateWatermark', wfCatalog, wmId, name, settings, relativePath, '4.3MB.mov' ), errorMessage )

		-- 4. Try to update with supported watermark format, this time choose jpg and check it is allowed.

		relativePath = KSPathUtils.child( 'assets', '1.5MB.jpg' )
		absolutePath = KSPathUtils.child( testDataPath, relativePath )
		fileSize = absolutePath and WFFileUtils.fileAttributes( absolutePath ).fileSize
		fileName = '1.5MB.jpg'

		-- checking Sha256 while creating avoided. Just checking one time is enough and saves test case run time!
		local sha256Digest = WFCrypto.digestFile( absolutePath, 'SHA256' )
		rxTest:waitFor(  callFunctionOrSubmodel(
					rxTest, wfCatalog,
					'WFCatWatermark',
					'updateWatermark', wfCatalog, wmId, name, settings, relativePath, fileName ) )

		-- verify localOnly update
		rxTest:expectSingleValue( wfCatalog:docStore():docValue( wmId )
			:map( function( watermarkDoc )
				local master = derefOpt( watermarkDoc, '_localOnly', 'universal', 'files', 'original', 'master' )
				return master.relativePath == relativePath and master.fileSize == fileSize and sha256Digest == master.sha256
			end ), true )

	end

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

	function WFCatWatermarkTest.rxTestVerifyImportSourceHandling( rxTest )

		--- This test case verifies composing and handling of importSource.

		rxTest:setTimeLimit( 30 )

		local wfCatalog = WFCatMockCatalog.createFakeWFCatalog( rxTest )

		--[[scenario-1: Create Text watermark. Expect No ImportSource has been created]]--

		-- create a text watermark
		local name = "VerifyImportSourceHandlingText"
		local settings = getWatermarkTestSetttings()
		local wmId
		rxTest:waitFor(
			 callFunctionOrSubmodel(
					rxTest, wfCatalog,
					'WFCatWatermark',
					'createWatermark',  wfCatalog, name, settings )
				:tap( function( watermarkId )
					-- capture the watermark id returned, it will be used for validation later.
					wmId = watermarkId
				end ) )

		rxTest:expectSingleValue( callFunctionOrSubmodel(
					rxTest, wfCatalog,
					'WFCatWatermark',
					'getWatermarkInfo', wfCatalog, wmId )
			:map( function( watermarkInfo )

				return ( watermarkInfo.importSource == nil ) -- <-- verification point

			end ), true )

		--[[scenario-2 : create image watermark: Expect importSource has been created]]--

		name = "VerifyImportSourceHandlingImage"
		settings.ActiveWatermarkType = "image"
		local testDataPath = KSPathUtils.getStandardFilePath( "testdata" )
		local relativePath = KSPathUtils.child( 'assets', 'IMG_7706.PNG' )
		local absolutePath = KSPathUtils.child( testDataPath, relativePath )
		local fileSize = absolutePath and WFFileUtils.fileAttributes( absolutePath ).fileSize
		local fileName = 'IMG_7706.PNG'
		local sha256 = WFCrypto.digestFile( absolutePath, 'SHA256' )

		rxTest:waitFor(
			 callFunctionOrSubmodel(
					rxTest, wfCatalog,
					'WFCatWatermark',
					'createWatermark',  wfCatalog, name, settings, relativePath, fileName )
				:tap( function( watermarkId )
					-- capture the watermark id returned, it will be used for validation later.
					wmId = watermarkId
				end ) )

		--- verify localOnly created.
		rxTest:expectSingleValue( wfCatalog:docStore():docValue( wmId )
			:map( function( watermarkDoc )
				local master = derefOpt( watermarkDoc, '_localOnly', 'universal', 'files', 'original', 'master' )
				return ( master.relativePath == relativePath and master.fileSize == fileSize )
			end ), true )

		  --  query getWatermarkInfo and check importSource correctly included in the watermarkInfo
		rxTest:expectSingleValue( callFunctionOrSubmodel(
					rxTest, wfCatalog,
					'WFCatWatermark',
					'getWatermarkInfo', wfCatalog, wmId )
			:map( function( watermarkInfo )

				return( watermarkInfo.importSource.fileName == fileName and
						watermarkInfo.importSource.sha256 == sha256 and
						watermarkInfo.importSource.fileSize == fileSize and
						watermarkInfo.importSource.importTimestamp == watermarkInfo.userCreated ) -- <-- verification point


		end ), true )

		--[[scenario 3: Update the settings and no binary update - expect that the importSource not altered]]--

		settings.rotation = 92 -- just update the settings and no update in binary
		rxTest:waitFor(  callFunctionOrSubmodel(
					rxTest, wfCatalog,
					'WFCatWatermark',
					'updateWatermark', wfCatalog, wmId, name, settings ) )

		--  query getWatermarkInfo and check importSource correctly included in the watermarkInfo
		rxTest:expectSingleValue( callFunctionOrSubmodel(
					rxTest, wfCatalog,
					'WFCatWatermark',
					'getWatermarkInfo', wfCatalog, wmId )
			:map( function( watermarkInfo )
				return( watermarkInfo.importSource.fileName == fileName and
						watermarkInfo.importSource.sha256 == sha256 and
						watermarkInfo.importSource.fileSize == fileSize and
						watermarkInfo.importSource.importTimestamp == watermarkInfo.userCreated )  -- <-- verification point

		end ), true )

		--[[scenario 4: Update the binary and check importSource has been updated accordingly]]--

		relativePath = KSPathUtils.child( 'assets', '1.5MB.jpg' )
		absolutePath = KSPathUtils.child( testDataPath, relativePath )
		fileSize = absolutePath and WFFileUtils.fileAttributes( absolutePath ).fileSize
		fileName = '1.5MB.jpg'
		sha256 = WFCrypto.digestFile( absolutePath, 'SHA256' )
		rxTest:waitFor(  callFunctionOrSubmodel(
					rxTest, wfCatalog,
					'WFCatWatermark',
					'updateWatermark', wfCatalog, wmId, name, settings, relativePath, fileName ) )

		-- verify localOnly update
		rxTest:expectSingleValue( wfCatalog:docStore():docValue( wmId )
			:map( function( watermarkDoc )
				local master = derefOpt( watermarkDoc, '_localOnly', 'universal', 'files', 'original', 'master' )
				return ( master.relativePath == relativePath and master.fileSize == fileSize )
			end ), true )

		-- verify importSource: query getWatermarkInfo and check importSource correctly included in the watermarkInfo
		rxTest:expectSingleValue( callFunctionOrSubmodel(
					rxTest, wfCatalog,
					'WFCatWatermark',
					'getWatermarkInfo', wfCatalog, wmId )
			:map( function( watermarkInfo )
				return( watermarkInfo.importSource.fileName == fileName and
						watermarkInfo.importSource.sha256 == sha256 and
						watermarkInfo.importSource.fileSize == fileSize and
						watermarkInfo.importSource.importTimestamp == watermarkInfo.userCreated ) -- <-- verification point
		end ), true )

		--[[scenario 5: change the activeWatermarkType again to text. Verify the importSource stays]]--
		settings.ActiveWatermarkType = "text"
		rxTest:waitFor(  callFunctionOrSubmodel(
					rxTest, wfCatalog,
					'WFCatWatermark',
					'updateWatermark', wfCatalog, wmId, name, settings ) )

		-- verify importSource: query getWatermarkInfo and check importSource correctly included in the watermarkInfo
		rxTest:expectSingleValue( callFunctionOrSubmodel(
					rxTest, wfCatalog,
					'WFCatWatermark',
					'getWatermarkInfo', wfCatalog, wmId )
			:map( function( watermarkInfo )
				return( watermarkInfo.importSource.fileName == fileName and
						watermarkInfo.importSource.sha256 == sha256 and
						watermarkInfo.importSource.fileSize == fileSize and
						watermarkInfo.importSource.importTimestamp == watermarkInfo.userCreated ) -- <-- verification point

		end ), true )

	end

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

	function WFCatWatermarkTest.rxTestFilePathAndFileNameExtensionMimatchShallResultInError( rxTest )

		-- This test case verifies the filePath and fileName extension are same!.
		-- Incase of mismatch return error!.

		rxTest:setTimeLimit( 30 )

		local wfCatalog = WFCatMockCatalog.createFakeWFCatalog( rxTest )

		-- create an image watermark
		local name = "VerifyUpdateWatermarkImage"
		local settings = getWatermarkTestSetttings()
		settings.activeWatermarkType = "image"

		local relativePath = KSPathUtils.child( 'assets', 'IMG_7706.PNG' ) --> path extension set to png
		local fileName = 'IMG_7706.JPG' --> file extension set to jpg

		rxTest:expectOnlyError(  callFunctionOrSubmodel(
					rxTest, wfCatalog,
					'WFCatWatermark',
					'createWatermark',  wfCatalog, name, settings,
				relativePath, fileName ), "Mismatch of filePath and fileName extension, png, jpg" )
	end

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


	function WFCatWatermarkTest.rxTestDownloadWatermarkError400_testNightly( rxTest )

		rxTest:setTimeLimit( 360 )

		-- Step 1: Create a watermark and upload to Oz

		local catalogTestSession = WFCatTestSession( rxTest )

		local catalog, catalogSpec

		rxTest:waitFor(
			catalogTestSession:catalog()
				:tap( function( _catalog )
					check( 'WFCatalog' )
					catalog = _catalog
					catalogSpec = _catalog:ozCatalogSpec()
				end ) )

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

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

		local uploader = WFCatBinaryUploader( catalog, nil, options, catalogSpec, {
					clearFromRecentlyUploadedSeconds = 1,
					waitForOtherUploadSeconds = .5,
					waitForCopyFailedSeconds = .5,
					timer = function( seconds )
						return Observable.timer( seconds, WFRxScheduler.sharedScheduler() )
					end,
				} )

		rxTest:addCloseTarget( uploader )

		uploader:DEBUG_resetDiagnosticLogErrors()

		rxTest:finally( function()
			puller:stopContinuousUpdates()
			pusher:stopContinuousUpdates()
			uploader:stopContinuousUploads()
		end )

		-- create an image watermark
		local name = "VerifyUpdateWatermarkImage"
		local settings = getWatermarkTestSetttings()
		settings.activeWatermarkType = "image"


		local testDataPath = KSPathUtils.getStandardFilePath( "testdata" )
		local relativePath = KSPathUtils.child( 'assets', 'IMG_7706.PNG' )
		local absolutePath = KSPathUtils.child( testDataPath, relativePath )
		--local fileSize = absolutePath and WFFileUtils.fileAttributes( absolutePath ).fileSize
		local fileName = 'IMG_7706.PNG'
		local sha256 = WFCrypto.digestFile( absolutePath, 'SHA256' )

		local wmId

		rxTest:waitFor(
			 callFunctionOrSubmodel(
					rxTest, catalog,
					'WFCatWatermark',
					'createWatermark',  catalog, name, settings, relativePath, fileName )
				:tap( function( watermarkId )
					-- capture the watermark id returned, it will be used for validation later.
					wmId = watermarkId
				end ) )


		-- start up sync engine

		pusher:sendUpdatesContinuously()
		puller:getUpdatesContinuously()
		uploader:startContinuousUploads()
		local href
		local docObserver = rxTest:observe( catalog:docStore():ongoingDocValue( wmId )
		:map( function( docValue )
			href = derefOpt( docValue, '_serverProvidedMetadata', 'links', '/rels/master', 'href')
			if href then
				return true
			end
		end ) )

		docObserver:expectDataWithinTime(120, true)

		-- Also verify whether originalLinkAvailable set to true
		rxTest:expectSingleValue( callFunctionOrSubmodel(
					rxTest, catalog,
					'WFCatWatermark',
					'getWatermarkInfo', catalog, wmId )
			:map( function( watermarkInfo )
				return ( watermarkInfo.originalLinkAvailable )
			end ), true )

		-- Step 2: Make sure the localOnly doesn't have the path.

		-- Step 3: corrupt the href : make invalid revision ( modify last 6 digits of revisionId )
		href = string.sub(href, 1, 125)
		href = href .. "111111/master"

		rxTest:waitFor( catalog:docStore():applyOverlay( {
				_id = wmId,
				_serverProvidedMetadata = {
					links = {
						[ '/rels/master' ] = {
							href = href
						},
					},
				 },
				_localOnly = {
					files = {
						original = {
							master = {
								relativePath = { _delete = true }
							}
						}

					}
				}
			} ) )

		-- Step 4: Attempt to download the file from Oz.


		--- saveFunc is dummy and doesn't play a role here. Don't bother much about the save function.
		local saveFunc = function( _, xmpKSData, watermarkDocVal )

			local tempFolder = WFTestFileUtils.createTestFolder( rxTest )

			local downloadFilePath = KSPathUtils.child( tempFolder, sha256 )
			fileName = KSTableUtils.derefOpt( watermarkDocVal, 'importSource', 'fileName' )
			return WFAsyncFileUtils.saveToFile( xmpKSData, downloadFilePath )

				:map( function( success )
					if success then
						return downloadFilePath, KSData.length( xmpKSData ), sha256
					end
					error( "failed to store watermark %s", fileName )
				end )

		end

		-- Step 5: As there is no valid request, it should result in bad request status_code 400
		rxTest:expectSingleValue( WFCatWatermark.downloadImageWatermark( catalog, wmId, saveFunc )

			:map( function( result )
				return ( result.status == "error" and result.headers.status_code == "400" )
			end ), true )

	end

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

	function WFCatWatermarkTest.rxTestDownloadWatermarkSuccess_testNightly( rxTest )

		rxTest:setTimeLimit( 360 )

		-- Step 1: Create a watermark and upload to Oz

		local catalogTestSession = WFCatTestSession( rxTest )

		local catalog, catalogSpec

		rxTest:waitFor(
			catalogTestSession:catalog()
				:tap( function( _catalog )
					check( 'WFCatalog' )
					catalog = _catalog
					catalogSpec = _catalog:ozCatalogSpec()
				end ) )

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

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

		local uploader = WFCatBinaryUploader( catalog, nil, options, catalogSpec, {
					clearFromRecentlyUploadedSeconds = 1,
					waitForOtherUploadSeconds = .5,
					waitForCopyFailedSeconds = .5,
					timer = function( seconds )
						return Observable.timer( seconds, WFRxScheduler.sharedScheduler() )
					end,
				} )

		rxTest:addCloseTarget( uploader )

		uploader:DEBUG_resetDiagnosticLogErrors()

		rxTest:finally( function()
			puller:stopContinuousUpdates()
			pusher:stopContinuousUpdates()
			uploader:stopContinuousUploads()
		end )

		-- create an image watermark
		local name = "VerifyUpdateWatermarkImage"
		local settings = getWatermarkTestSetttings()
		settings.activeWatermarkType = "image"


		local testDataPath = KSPathUtils.getStandardFilePath( "testdata" )
		local relativePath = KSPathUtils.child( 'assets', 'IMG_7706.PNG' )
		local absolutePath = KSPathUtils.child( testDataPath, relativePath )
		--local fileSize = absolutePath and WFFileUtils.fileAttributes( absolutePath ).fileSize
		local fileName = 'IMG_7706.PNG'
		local sha256 = WFCrypto.digestFile( absolutePath, 'SHA256' )

		local wmId

		rxTest:waitFor(
			 callFunctionOrSubmodel(
					rxTest, catalog,
					'WFCatWatermark',
					'createWatermark',  catalog, name, settings, relativePath, fileName )
				:tap( function( watermarkId )
					-- capture the watermark id returned, it will be used for validation later.
					wmId = watermarkId
				end ) )


		-- start up sync engine

		pusher:sendUpdatesContinuously()
		puller:getUpdatesContinuously()
		uploader:startContinuousUploads()
		local href
		local docObserver = rxTest:observe( catalog:docStore():ongoingDocValue( wmId )
		:map( function( docValue )
			href = derefOpt( docValue, '_serverProvidedMetadata', 'links', '/rels/master', 'href')
			if href then
				return true
			end
		end ) )

		docObserver:expectDataWithinTime(120, true)

		-- Also verify whether originalLinkAvailable set to true
		rxTest:expectSingleValue( callFunctionOrSubmodel(
					rxTest, catalog,
					'WFCatWatermark',
					'getWatermarkInfo', catalog, wmId )
			:map( function( watermarkInfo )
				return ( watermarkInfo.originalLinkAvailable )
			end ), true )

		-- Step 2: Make sure the localOnly doesn't have the path.
		rxTest:waitFor( catalog:docStore():applyOverlay( {
				_id = wmId,
				_localOnly = {
					files = {
						original = {
							master = {
								relativePath = { _delete = true }
							}
						}

					}
				}
			} ) )

		-- Step 3: Attempt to download the file from Oz.
		local downloadFilePath

		local saveFunc = function( _, xmpKSData, docValue )

			local tempFolder = WFTestFileUtils.createTestFolder( rxTest )

			downloadFilePath = KSPathUtils.child( tempFolder, sha256 )
			fileName = KSTableUtils.derefOpt( docValue, 'importSource', 'fileName' )
			return WFAsyncFileUtils.saveToFile( xmpKSData, downloadFilePath )

				:map( function( success )
					if success then
						return downloadFilePath, KSData.length( xmpKSData ), sha256
					end
					error( "failed to store watermark %s", fileName )
				end )

		end

		rxTest:waitFor( WFCatWatermark.downloadImageWatermark( catalog, wmId, saveFunc ) )

		-- Step 4: Check downloaded file path
		rxTest:expectSingleValue( catalog:docStore():docValue( wmId )

			:map( function( docValue )

				return WFCatAssetUtils.getLocalWatermarkBinaryRelativePath( docValue )

			end ), downloadFilePath )

		rxTest:expectEqual( WFFileUtils.exists( downloadFilePath ), 'file' )

	end

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

	return WFRxTestCase.wrapSuite( WFCatWatermarkTest )

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

end
