func testBlockfileStream(t *testing.T, numBlocks int) {
	env := newTestEnv(t)
	defer env.Cleanup()
	w := newTestBlockfileWrapper(t, env)
	blockfileMgr := w.blockfileMgr
	blocks := testutil.ConstructTestBlocks(t, numBlocks)
	w.addBlocks(blocks)
	w.close()

	s, err := newBlockfileStream(blockfileMgr.rootDir, 0, 0)
	defer s.close()
	testutil.AssertNoError(t, err, "Error in constructing blockfile stream")

	blockCount := 0
	for {
		blockBytes, err := s.nextBlockBytes()
		testutil.AssertNoError(t, err, "Error in getting next block")
		if blockBytes == nil {
			break
		}
		blockCount++
	}
	// After the stream has been exhausted, both blockBytes and err should be nil
	blockBytes, err := s.nextBlockBytes()
	testutil.AssertNil(t, blockBytes)
	testutil.AssertNoError(t, err, "Error in getting next block after exhausting the file")
	testutil.AssertEquals(t, blockCount, numBlocks)
}

func TestRawLedger(t *testing.T) {
	cleanup(t)
	rawLedger := NewFSBasedRawLedger(testFolder)
	defer rawLedger.Close()
	defer cleanup(t)

	// Construct test blocks and add to raw ledger
	blocks := testutil.ConstructTestBlocks(t, 10)
	for _, block := range blocks {
		rawLedger.CommitBlock(block)
	}

	// test GetBlockchainInfo()
	bcInfo, err := rawLedger.GetBlockchainInfo()
	testutil.AssertNoError(t, err, "Error in getting BlockchainInfo")
	testutil.AssertEquals(t, bcInfo.Height, uint64(10))

	// test GetBlockByNumber()
	block, err := rawLedger.GetBlockByNumber(2)
	testutil.AssertNoError(t, err, "Error in getting block by number")
	testutil.AssertEquals(t, block, blocks[1])

	// get blocks iterator for block number starting from 3
	itr, err := rawLedger.GetBlocksIterator(3)
	testutil.AssertNoError(t, err, "Error in getting iterator")
	blockHolder, err := itr.Next()
	testutil.AssertNoError(t, err, "")
	testutil.AssertEquals(t, blockHolder.(ledger.BlockHolder).GetBlock(), blocks[2])
	// get next block from iterator. The block should be 4th block
	blockHolder, err = itr.Next()
	testutil.AssertNoError(t, err, "")
	testutil.AssertEquals(t, blockHolder.(ledger.BlockHolder).GetBlock(), blocks[3])
}

func TestBlockFileScanSmallTxLastTxIncomplete(t *testing.T) {
	env := newTestEnv(t)
	defer env.Cleanup()
	blkfileMgrWrapper := newTestBlockfileWrapper(t, env)
	bg := testutil.NewBlockGenerator(t)
	blocks := []*common.Block{}
	blocks = append(blocks, bg.NextTestBlock(0, 0))
	blocks = append(blocks, bg.NextTestBlock(0, 0))
	blocks = append(blocks, bg.NextTestBlock(0, 0))
	blkfileMgrWrapper.addBlocks(blocks)
	blkfileMgrWrapper.close()

	filePath := deriveBlockfilePath(env.conf.blockfilesDir, 0)
	_, fileSize, err := util.FileExists(filePath)
	testutil.AssertNoError(t, err, "")

	file, err := os.OpenFile(filePath, os.O_RDWR|os.O_APPEND|os.O_CREATE, 0660)
	defer file.Close()
	testutil.AssertNoError(t, err, "")
	err = file.Truncate(fileSize - 1)
	testutil.AssertNoError(t, err, "")

	_, numBlocks, err := scanForLastCompleteBlock(env.conf.blockfilesDir, 0, 0)
	testutil.AssertNoError(t, err, "")
	testutil.AssertEquals(t, numBlocks, len(blocks)-1)
}

func testBlockStream(t *testing.T, numFiles int) {
	env := newTestEnv(t)
	defer env.Cleanup()
	w := newTestBlockfileWrapper(t, env)
	defer w.close()
	blockfileMgr := w.blockfileMgr

	numBlocksInEachFile := 10
	bg := testutil.NewBlockGenerator(t)
	for i := 0; i < numFiles; i++ {
		blocks := bg.NextTestBlocks(numBlocksInEachFile)
		w.addBlocks(blocks)
		blockfileMgr.moveToNextFile()
	}
	s, err := newBlockStream(blockfileMgr.rootDir, 0, 0, numFiles-1)
	defer s.close()
	testutil.AssertNoError(t, err, "Error in constructing new block stream")
	blockCount := 0
	for {
		blockBytes, err := s.nextBlockBytes()
		testutil.AssertNoError(t, err, "Error in getting next block")
		if blockBytes == nil {
			break
		}
		blockCount++
	}
	// After the stream has been exhausted, both blockBytes and err should be nil
	blockBytes, err := s.nextBlockBytes()
	testutil.AssertNil(t, blockBytes)
	testutil.AssertNoError(t, err, "Error in getting next block after exhausting the file")
	testutil.AssertEquals(t, blockCount, numFiles*numBlocksInEachFile)
}

// TestBasicRW tests basic read-write
func TestBasicRW(t *testing.T, db statedb.VersionedDB) {
	db.Open()
	defer db.Close()
	val, err := db.GetState("ns", "key1")
	testutil.AssertNoError(t, err, "")
	testutil.AssertNil(t, val)

	batch := statedb.NewUpdateBatch()
	vv1 := statedb.VersionedValue{Value: []byte("value1"), Version: version.NewHeight(1, 1)}
	vv2 := statedb.VersionedValue{Value: []byte("value2"), Version: version.NewHeight(1, 2)}
	vv3 := statedb.VersionedValue{Value: []byte("value3"), Version: version.NewHeight(1, 3)}
	vv4 := statedb.VersionedValue{Value: []byte{}, Version: version.NewHeight(1, 4)}
	batch.Put("ns1", "key1", vv1.Value, vv1.Version)
	batch.Put("ns1", "key2", vv2.Value, vv2.Version)
	batch.Put("ns2", "key3", vv3.Value, vv3.Version)
	batch.Put("ns2", "key4", vv4.Value, vv4.Version)
	savePoint := version.NewHeight(2, 5)
	db.ApplyUpdates(batch, savePoint)

	vv, _ := db.GetState("ns1", "key1")
	testutil.AssertEquals(t, vv, &vv1)

	vv, _ = db.GetState("ns2", "key4")
	testutil.AssertEquals(t, vv, &vv4)

	sp, err := db.GetLatestSavePoint()
	testutil.AssertNoError(t, err, "")
	testutil.AssertEquals(t, sp, savePoint)
}

// TestDeletes tests deteles
func TestDeletes(t *testing.T, db statedb.VersionedDB) {
	db.Open()
	defer db.Close()
	batch := statedb.NewUpdateBatch()
	vv1 := statedb.VersionedValue{Value: []byte("value1"), Version: version.NewHeight(1, 1)}
	vv2 := statedb.VersionedValue{Value: []byte("value2"), Version: version.NewHeight(1, 2)}
	vv3 := statedb.VersionedValue{Value: []byte("value1"), Version: version.NewHeight(1, 3)}
	vv4 := statedb.VersionedValue{Value: []byte("value2"), Version: version.NewHeight(1, 4)}

	batch.Put("ns", "key1", vv1.Value, vv1.Version)
	batch.Put("ns", "key2", vv2.Value, vv2.Version)
	batch.Put("ns", "key3", vv2.Value, vv3.Version)
	batch.Put("ns", "key4", vv2.Value, vv4.Version)
	batch.Delete("ns", "key3", version.NewHeight(1, 5))
	savePoint := version.NewHeight(1, 5)
	err := db.ApplyUpdates(batch, savePoint)
	testutil.AssertNoError(t, err, "")
	vv, _ := db.GetState("ns", "key2")
	testutil.AssertEquals(t, vv, &vv2)

	vv, err = db.GetState("ns", "key3")
	testutil.AssertNoError(t, err, "")
	testutil.AssertNil(t, vv)

	batch = statedb.NewUpdateBatch()
	batch.Delete("ns", "key2", version.NewHeight(1, 6))
	err = db.ApplyUpdates(batch, savePoint)
	testutil.AssertNoError(t, err, "")
	vv, err = db.GetState("ns", "key2")
	testutil.AssertNoError(t, err, "")
	testutil.AssertNil(t, vv)
}

func TestTxRWSetMarshalUnmarshal(t *testing.T) {
	txRW := &TxReadWriteSet{}
	nsRW1 := &NsReadWriteSet{"ns1",
		[]*KVRead{&KVRead{"key1", version.NewHeight(1, 1)}},
		[]*KVWrite{&KVWrite{"key2", false, []byte("value2")}}}

	nsRW2 := &NsReadWriteSet{"ns2",
		[]*KVRead{&KVRead{"key3", version.NewHeight(1, 2)}},
		[]*KVWrite{&KVWrite{"key4", true, nil}}}

	nsRW3 := &NsReadWriteSet{"ns3",
		[]*KVRead{&KVRead{"key5", version.NewHeight(1, 3)}},
		[]*KVWrite{&KVWrite{"key6", false, []byte("value6")}, &KVWrite{"key7", false, []byte("value7")}}}

	txRW.NsRWs = append(txRW.NsRWs, nsRW1, nsRW2, nsRW3)

	b, err := txRW.Marshal()
	testutil.AssertNoError(t, err, "Error while marshalling changeset")

	deserializedRWSet := &TxReadWriteSet{}
	err = deserializedRWSet.Unmarshal(b)
	testutil.AssertNoError(t, err, "Error while unmarshalling changeset")
	t.Logf("Unmarshalled changeset = %#+v", deserializedRWSet.NsRWs[0].Writes[0].IsDelete)
	testutil.AssertEquals(t, deserializedRWSet, txRW)
}

func TestDirNotEmptyAndFileExists(t *testing.T) {

	cleanup(DbPathTest)
	defer cleanup(DbPathTest)

	//create the directory
	dirEmpty, err := CreateDirIfMissing(DbPathTest)
	testutil.AssertNoError(t, err, fmt.Sprintf("Error when trying to create a test db directory at [%s]", DbPathTest))
	testutil.AssertEquals(t, dirEmpty, true)

	//test file does not exists and size is returned correctly
	exists2, size2, err2 := FileExists(DbFileTest)
	testutil.AssertNoError(t, err2, fmt.Sprintf("Error when trying to determine if file exist when it does not at [%s]", DbFileTest))
	testutil.AssertEquals(t, size2, int64(0))
	testutil.AssertEquals(t, exists2, false) //test file that does not exists reports false

	//create file
	testStr := "This is some test data in a file"
	sizeOfFileCreated, err3 := createAndWriteAFile(testStr)
	testutil.AssertNoError(t, err3, fmt.Sprintf("Error when trying to create and write to file at [%s]", DbFileTest))
	testutil.AssertEquals(t, sizeOfFileCreated, len(testStr)) //test file size returned is correct

	//test that the file exists and size is returned correctly
	exists, size, err4 := FileExists(DbFileTest)
	testutil.AssertNoError(t, err4, fmt.Sprintf("Error when trying to determine if file exist at [%s]", DbFileTest))
	testutil.AssertEquals(t, size, int64(sizeOfFileCreated))
	testutil.AssertEquals(t, exists, true) //test file that does exists reports true

	//test that if the directory is not empty
	dirEmpty5, err5 := DirEmpty(DbPathTest)
	testutil.AssertNoError(t, err5, fmt.Sprintf("Error when detecting if empty at db directory [%s]", DbPathTest))
	testutil.AssertEquals(t, dirEmpty5, false) //test directory is empty is returning false
}

//TestSavepoint tests the recordSavepoint and GetBlockNumfromSavepoint methods for recording and reading a savepoint document
func TestSavepoint(t *testing.T) {

	//Only run the tests if CouchDB is explitily enabled in the code,
	//otherwise CouchDB may not be installed and all the tests would fail
	//TODO replace this with external config property rather than config within the code
	if ledgerconfig.IsCouchDBEnabled() == true {

		env := newTestEnv(t)

		txMgr := NewCouchDBTxMgr(env.conf,
			env.couchDBAddress,    //couchDB Address
			env.couchDatabaseName, //couchDB db name
			env.couchUsername,     //enter couchDB id
			env.couchPassword)     //enter couchDB pw

		// record savepoint
		txMgr.blockNum = 5
		err := txMgr.recordSavepoint()
		testutil.AssertNoError(t, err, fmt.Sprintf("Error when saving recordpoint data"))

		// read the savepoint
		blockNum, err := txMgr.GetBlockNumFromSavepoint()
		testutil.AssertNoError(t, err, fmt.Sprintf("Error when saving recordpoint data"))
		testutil.AssertEquals(t, txMgr.blockNum, blockNum)

		txMgr.Shutdown()
	}
}

func TestDBBadJSON(t *testing.T) {

	if ledgerconfig.IsCouchDBEnabled() == true {

		cleanup()
		defer cleanup()

		//create a new connection
		db, err := CreateConnectionDefinition(connectURL, database, username, password)
		testutil.AssertNoError(t, err, fmt.Sprintf("Error when trying to create database connection definition"))

		//create a new database
		_, errdb := db.CreateDatabaseIfNotExist()
		testutil.AssertNoError(t, errdb, fmt.Sprintf("Error when trying to create database"))

		//Retrieve the info for the new database and make sure the name matches
		dbResp, _, errdb := db.GetDatabaseInfo()
		testutil.AssertNoError(t, errdb, fmt.Sprintf("Error when trying to retrieve database information"))
		testutil.AssertEquals(t, dbResp.DbName, database)

		badJSON := []byte(`{"asset_name"}`)

		//Save the test document
		_, saveerr := db.SaveDoc("1", "", badJSON, nil)
		testutil.AssertError(t, saveerr, fmt.Sprintf("Error should have been thrown for a bad JSON"))

	}

}

func TestBlockSerialization(t *testing.T) {
	block := testutil.ConstructTestBlock(t, 10, 100)
	bb, _, err := serializeBlock(block)
	testutil.AssertNoError(t, err, "")
	deserializedBlock, err := deserializeBlock(bb)
	testutil.AssertNoError(t, err, "")
	testutil.AssertEquals(t, deserializedBlock, block)
}

//InitTestLedger provides a ledger for testing. This method creates a fresh db and constructs a ledger instance on that.
func InitTestLedger(t *testing.T) *Ledger {
	testDBWrapper.CleanDB(t)
	_, err := GetLedger()
	testutil.AssertNoError(t, err, "Error while constructing ledger")
	newLedger, err := GetNewLedger()
	testutil.AssertNoError(t, err, "Error while constructing ledger")
	ledger = newLedger
	return newLedger
}

func (w *testBlockfileMgrWrapper) testGetBlockByNumber(blocks []*common.Block, startingNum uint64) {
	for i := 0; i < len(blocks); i++ {
		b, err := w.blockfileMgr.retrieveBlockByNumber(startingNum + uint64(i))
		testutil.AssertNoError(w.t, err, fmt.Sprintf("Error while retrieving [%d]th block from blockfileMgr", i))
		testutil.AssertEquals(w.t, b, blocks[i])
	}
	// test getting the last block
	b, err := w.blockfileMgr.retrieveBlockByNumber(math.MaxUint64)
	iLastBlock := len(blocks) - 1
	testutil.AssertNoError(w.t, err, fmt.Sprintf("Error while retrieving last block from blockfileMgr"))
	testutil.AssertEquals(w.t, b, blocks[iLastBlock])
}

func TestNilTxRWSet(t *testing.T) {
	txRW := &TxReadWriteSet{}
	nsRW1 := &NsReadWriteSet{"ns1",
		[]*KVRead{&KVRead{"key1", nil}},
		[]*KVWrite{&KVWrite{"key1", false, []byte("value1")}}}
	txRW.NsRWs = append(txRW.NsRWs, nsRW1)
	b, err := txRW.Marshal()
	testutil.AssertNoError(t, err, "Error while marshalling changeset")

	deserializedRWSet := &TxReadWriteSet{}
	err = deserializedRWSet.Unmarshal(b)
	testutil.AssertNoError(t, err, "Error while unmarshalling changeset")
	t.Logf("Unmarshalled changeset = %#+v", deserializedRWSet.NsRWs[0].Writes[0].IsDelete)
	testutil.AssertEquals(t, deserializedRWSet, txRW)
}

func (h *txMgrTestHelper) validateAndCommitRWSet(txRWSet []byte) {
	block := h.bg.NextBlock([][]byte{txRWSet}, false)
	err := h.txMgr.ValidateAndPrepare(block, true)
	testutil.AssertNoError(h.t, err, "")
	txsFltr := util.NewFilterBitArrayFromBytes(block.Metadata.Metadata[common.BlockMetadataIndex_TRANSACTIONS_FILTER])
	invalidTxNum := 0
	for i := 0; i < len(block.Data.Data); i++ {
		if txsFltr.IsSet(uint(i)) {
			invalidTxNum++
		}
	}
	testutil.AssertEquals(h.t, invalidTxNum, 0)
	err = h.txMgr.Commit()
	testutil.AssertNoError(h.t, err, "")
}

func testBlockIndexSync(t *testing.T, numBlocks int, numBlocksToIndex int, syncByRestart bool) {
	env := newTestEnv(t)
	defer env.Cleanup()
	blkfileMgrWrapper := newTestBlockfileWrapper(t, env)
	defer blkfileMgrWrapper.close()
	blkfileMgr := blkfileMgrWrapper.blockfileMgr
	origIndex := blkfileMgr.index
	// construct blocks for testing
	blocks := testutil.ConstructTestBlocks(t, numBlocks)
	// add a few blocks
	blkfileMgrWrapper.addBlocks(blocks[:numBlocksToIndex])

	// Plug-in a noop index and add remaining blocks
	blkfileMgr.index = &noopIndex{}
	blkfileMgrWrapper.addBlocks(blocks[numBlocksToIndex:])

	// Plug-in back the original index
	blkfileMgr.index = origIndex
	// The first set of blocks should be present in the orginal index
	for i := 1; i <= numBlocksToIndex; i++ {
		block, err := blkfileMgr.retrieveBlockByNumber(uint64(i))
		testutil.AssertNoError(t, err, fmt.Sprintf("block [%d] should have been present in the index", i))
		testutil.AssertEquals(t, block, blocks[i-1])
	}

	// The last set of blocks should not be present in the original index
	for i := numBlocksToIndex + 1; i <= numBlocks; i++ {
		_, err := blkfileMgr.retrieveBlockByNumber(uint64(i))
		testutil.AssertSame(t, err, blkstorage.ErrNotFoundInIndex)
	}

	// perform index sync
	if syncByRestart {
		blkfileMgrWrapper.close()
		blkfileMgrWrapper = newTestBlockfileWrapper(t, env)
		defer blkfileMgrWrapper.close()
		blkfileMgr = blkfileMgrWrapper.blockfileMgr
	} else {
		blkfileMgr.syncIndex()
	}

	// Now, last set of blocks should also be present in original index
	for i := numBlocksToIndex + 1; i <= numBlocks; i++ {
		block, err := blkfileMgr.retrieveBlockByNumber(uint64(i))
		testutil.AssertNoError(t, err, fmt.Sprintf("block [%d] should have been present in the index", i))
		testutil.AssertEquals(t, block, blocks[i-1])
	}
}

func (stateTrieTestWrapper *stateTrieTestWrapper) PrepareWorkingSetAndComputeCryptoHash(stateDelta *statemgmt.StateDelta) []byte {
	stateTrieTestWrapper.stateTrie.PrepareWorkingSet(stateDelta)
	cryptoHash, err := stateTrieTestWrapper.stateTrie.ComputeCryptoHash()
	testutil.AssertNoError(stateTrieTestWrapper.t, err, "Error while computing crypto hash")
	stateTrieTestWrapper.t.Logf("Cryptohash = [%x]", cryptoHash)
	return cryptoHash
}

func TestCreatingDBDirWhenDirDoesAndDoesNotExists(t *testing.T) {

	cleanup(DbPathTest) //invoked prior to test to make sure test is executed in desired environment
	defer cleanup(DbPathTest)

	//test creating a directory without path separator passed
	dirEmpty, err := CreateDirIfMissing(DbPathTest)
	testutil.AssertNoError(t, err, fmt.Sprintf("Error when trying to create a test db directory at [%s]", DbPathTest))
	testutil.AssertEquals(t, dirEmpty, true)

	//test creating directory AGAIN, that is the directory already exists
	dirEmpty2, err2 := CreateDirIfMissing(DbPathTest)
	testutil.AssertNoError(t, err2, fmt.Sprintf("Error not handling existing directory when trying to create a test db directory at [%s]", DbPathTest))
	testutil.AssertEquals(t, dirEmpty2, true)

}

func newStateImplTestWrapper(t testing.TB) *stateImplTestWrapper {
	var configMap map[string]interface{}
	stateImpl := NewStateImpl()
	err := stateImpl.Initialize(configMap)
	testutil.AssertNoError(t, err, "Error while constrcuting stateImpl")
	return &stateImplTestWrapper{configMap, stateImpl, t}
}

func newStateImplTestWrapperWithCustomConfig(t testing.TB, numBuckets int, maxGroupingAtEachLevel int) *stateImplTestWrapper {
	configMap := map[string]interface{}{ConfigNumBuckets: numBuckets, ConfigMaxGroupingAtEachLevel: maxGroupingAtEachLevel}
	stateImpl := NewStateImpl()
	err := stateImpl.Initialize(configMap)
	testutil.AssertNoError(t, err, "Error while constrcuting stateImpl")
	return &stateImplTestWrapper{configMap, stateImpl, t}
}