func watchUpdate(kv kvdb.Kvdb, data *watchData) error {
	var err error
	var kvp *kvdb.KVPair

	data.reader, data.writer = 0, 0
	atomic.AddInt32(&data.writer, 1)
	// whichKey = 1 : key
	// whichKey = 0 : otherKey
	atomic.SwapInt32(&data.whichKey, 1)
	data.action = kvdb.KVCreate
	fmt.Printf("-")
	kvp, err = kv.Create(data.key, []byte("bar"), 0)
	for i := 0; i < data.iterations && err == nil; i++ {
		fmt.Printf("-")

		for data.writer != data.reader {
			time.Sleep(time.Millisecond * 100)
		}
		atomic.AddInt32(&data.writer, 1)
		data.action = kvdb.KVSet
		kvp, err = kv.Put(data.key, []byte("bar"), 0)

		data.updateIndex = kvp.KVDBIndex
		assert.NoError(data.t, err, "Unexpected error in Put")
	}

	fmt.Printf("-")
	for data.writer != data.reader {
		time.Sleep(time.Millisecond * 100)
	}
	atomic.AddInt32(&data.writer, 1)
	// Delete key
	data.action = kvdb.KVDelete
	kv.Delete(data.key)

	fmt.Printf("-")
	for data.writer != data.reader {
		time.Sleep(time.Millisecond * 100)
	}
	atomic.AddInt32(&data.writer, 1)

	atomic.SwapInt32(&data.whichKey, 0)
	data.action = kvdb.KVDelete
	// Delete otherKey
	kv.Delete(data.otherKey)

	fmt.Printf("-")
	for data.writer != data.reader {
		time.Sleep(time.Millisecond * 100)
	}
	atomic.AddInt32(&data.writer, 1)

	atomic.SwapInt32(&data.whichKey, 1)
	data.action = kvdb.KVCreate
	_, err = kv.Create(data.key, []byte(data.stop), 0)

	return err
}

func getLockMethods(kv kvdb.Kvdb) []func(string) (*kvdb.KVPair, error) {
	lockMethods := make([]func(string) (*kvdb.KVPair, error), 2)
	lockMethods[0] = func(key string) (*kvdb.KVPair, error) {
		return kv.Lock(key)
	}
	lockMethods[1] = func(key string) (*kvdb.KVPair, error) {
		tag := "node:node_1,func:testfunc"
		return kv.LockWithID(key, tag)
	}
	return lockMethods
}

func deleteTree(kv kvdb.Kvdb, t *testing.T) {
	fmt.Println("deleteTree")

	prefix := "tree"
	keys := map[string]string{
		prefix + "/1cbc9a98-072a-4793-8608-01ab43db96c8": "bar",
		prefix + "/foo":                                  "baz",
	}

	for key, val := range keys {
		_, err := kv.Put(key, []byte(val), 0)
		assert.NoError(t, err, "Unexpected error on Put")
	}

	for key := range keys {
		_, err := kv.Get(key)
		assert.NoError(t, err, "Unexpected error on Get")
	}
	err := kv.DeleteTree(prefix)
	assert.NoError(t, err, "Unexpected error on DeleteTree")

	for key := range keys {
		_, err := kv.Get(key)
		assert.Error(t, err, "Get should fail on all keys after DeleteTree")
	}
}

func getInterface(kv kvdb.Kvdb, t *testing.T) {

	fmt.Println("getInterface")
	expected := struct {
		N int
		S string
	}{
		N: 10,
		S: "Ten",
	}

	actual := expected
	actual.N = 0
	actual.S = "zero"

	key := "DEADBEEF"
	_, err := kv.Delete(key)
	_, err = kv.Put(key, &expected, 0)
	assert.NoError(t, err, "Failed in Put")

	_, err = kv.GetVal(key, &actual)
	assert.NoError(t, err, "Failed in Get")

	assert.Equal(t, expected, actual, "Expected %#v but got %#v",
		expected, actual)
}

func update(kv kvdb.Kvdb, t *testing.T) {
	fmt.Println("update")

	key := "update/foo"
	kv.Delete(key)

	kvp, err := kv.Update(key, []byte("bar"), 0)
	assert.Error(t, err, "Update should error on non-existent key")

	defer func() {
		kv.Delete(key)
	}()

	kvp, err = kv.Create(key, []byte("bar"), 0)
	assert.NoError(t, err, "Unexpected error on create")

	kvp, err = kv.Update(key, []byte("bar"), 0)
	assert.NoError(t, err, "Unexpected error on update")

	assert.Equal(t, kvp.Action, kvdb.KVSet,
		"Expected action KVSet, actual %v", kvp.Action)
}

func createWithTTL(kv kvdb.Kvdb, t *testing.T) {
	fmt.Println("create with ttl")
	key := "create/foo"
	kv.Delete(key)
	assert.NotNil(t, kv, "Default KVDB is not set")
	_, err := kv.Create(key, []byte("bar"), 6)
	if err != nil {
		// Consul does not support ttl less than 10
		assert.EqualError(t, err, kvdb.ErrTTLNotSupported.Error(), "ttl not supported")
		_, err := kv.Create(key, []byte("bar"), 20)
		assert.NoError(t, err, "Error on create")
		// Consul doubles the ttl value
		time.Sleep(time.Second * 20)
		_, err = kv.Get(key)
		assert.Error(t, err, "Expecting error value for expired value")

	} else {
		assert.NoError(t, err, "Error on create")
		time.Sleep(time.Second * 7)
		_, err = kv.Get(key)
		assert.Error(t, err, "Expecting error value for expired value")
	}
}

func lockBasic(kv kvdb.Kvdb, t *testing.T) {
	lockMethods := getLockMethods(kv)

	for _, lockMethod := range lockMethods {
		fmt.Println("lock")

		key := "locktest"
		kvPair, err := lockMethod(key)
		assert.NoError(t, err, "Unexpected error in lock")

		if kvPair == nil {
			return
		}

		err = kv.Unlock(kvPair)
		assert.NoError(t, err, "Unexpected error from Unlock")

		kvPair, err = lockMethod(key)
		assert.NoError(t, err, "Failed to lock after unlock")

		err = kv.Unlock(kvPair)
		assert.NoError(t, err, "Unexpected error from Unlock")
	}
}

func create(kv kvdb.Kvdb, t *testing.T) {
	fmt.Println("create")

	key := "///create/foo"
	kv.Delete(key)

	kvp, err := kv.Create(key, []byte("bar"), 0)
	require.NoError(t, err, "Error on create")

	defer func() {
		kv.Delete(key)
	}()
	assert.Equal(t, kvp.Action, kvdb.KVCreate,
		"Expected action KVCreate, actual %v", kvp.Action)

	_, err = kv.Create(key, []byte("bar"), 0)
	assert.Error(t, err, "Create on existing key should have errored.")
}

func get(kv kvdb.Kvdb, t *testing.T) {
	fmt.Println("get")

	kvPair, err := kv.Get("DEADCAFE")
	assert.Error(t, err, "Expecting error value for non-existent value")

	key := "foo/docker"
	val := "great"
	defer func() {
		kv.Delete(key)
	}()

	kvPair, err = kv.Put(key, []byte(val), 0)
	assert.NoError(t, err, "Unexpected error in Put")

	kvPair, err = kv.Get(key)
	assert.NoError(t, err, "Failed in Get")

	assert.Equal(t, key, kvPair.Key, "Key mismatch in Get")
	assert.Equal(t, string(kvPair.Value), val, "value mismatch in Get")
}

func watchTree(kv kvdb.Kvdb, t *testing.T) {
	fmt.Println("\nwatchTree")

	tree := "tree"

	watchData := watchData{
		t:          t,
		key:        tree + "/key",
		otherKey:   tree + "/otherKey",
		stop:       "stop",
		iterations: 2,
	}
	_, err := kv.Delete(watchData.key)
	_, err = kv.Delete(watchData.otherKey)

	// First create a tree to watch for. We should not get update for this create.
	_, err = kv.Create(watchData.otherKey, []byte("bar"), 0)
	// Let the create operation finish and then start the watch

	time.Sleep(time.Second)
	err = kv.WatchTree(tree, 0, &watchData, watchFn)
	if err != nil {
		fmt.Printf("Cannot test watchKey: %v\n", err)
		return
	}

	// Sleep for sometime before calling the watchUpdate go routine.
	time.Sleep(time.Millisecond * 100)

	go randomUpdate(kv, &watchData)
	go watchUpdate(kv, &watchData)

	for watchData.watchStopped == false {
		time.Sleep(time.Millisecond * 100)
	}
}

func collect(kv kvdb.Kvdb, t *testing.T) {
	fmt.Println("collect")

	root := "pwx/test/collect"
	firstLevel := root + "/first"
	secondLevel := root + "/second"

	kv.DeleteTree(root)
	kvp, _ := kv.Create(firstLevel, []byte("bar"), 0)
	fmt.Printf("KVP is %v", kvp)
	collector, _ := kvdb.NewUpdatesCollector(kv, secondLevel, kvp.CreatedIndex)
	time.Sleep(time.Second)
	var updates []*kvdb.KVPair
	updateFn := func(start, end int) uint64 {
		maxVersion := uint64(0)
		for i := start; i < end; i++ {
			newLeaf := strconv.Itoa(i)
			kvp1, _ := kv.Create(secondLevel+"/"+newLeaf, []byte(newLeaf), 0)
			kvp2, _ := kv.Update(firstLevel, []byte(newLeaf), 0)
			updates = append(updates, kvp1)
			maxVersion = kvp2.ModifiedIndex
		}
		return maxVersion
	}
	updateFn(0, 10)
	// Allow watch updates to come back.
	time.Sleep(time.Millisecond * 500)
	collector.Stop()

	updateFn(10, 20)
	lastKVIndex := kvp.CreatedIndex
	lastLeafIndex := -1
	cb := func(prefix string, opaque interface{}, kvp *kvdb.KVPair,
		err error) error {
		assert.True(t, err == nil, "Error is nil %v", err)
		assert.True(t, kvp.ModifiedIndex > lastKVIndex,
			"Modified index %v lower than last index %v",
			kvp.ModifiedIndex, lastKVIndex)
		lastKVIndex = kvp.ModifiedIndex
		strValue := string(kvp.Value)
		value := secondLevel + "/" + strValue
		assert.True(t, strings.Compare(kvp.Key, value) == 0,
			"Key: %v, Value: %v", kvp.Key, value)
		leafIndex, _ := strconv.Atoi(strValue)
		assert.True(t, leafIndex == lastLeafIndex+1,
			"Last leaf: %v, leaf: %v", kvp.Key,
			value)
		lastLeafIndex = leafIndex
		return nil
	}

	replayCb := make([]kvdb.ReplayCb, 1)
	replayCb[0].Prefix = secondLevel
	replayCb[0].WatchCB = cb

	_, err := collector.ReplayUpdates(replayCb)
	assert.True(t, err == nil, "Replay encountered error %v", err)
	assert.True(t, lastLeafIndex == 9, "Last leaf index %v, expected : 9",
		lastLeafIndex)

	// Test with no updates.
	thirdLevel := root + "/third"
	kvp, _ = kv.Create(thirdLevel, []byte("bar_update"), 0)
	collector, _ = kvdb.NewUpdatesCollector(kv, thirdLevel, kvp.ModifiedIndex)
	time.Sleep(2 * time.Second)
	_, err = collector.ReplayUpdates(replayCb)
	assert.True(t, err == nil, "Replay encountered error %v", err)
	assert.True(t, lastLeafIndex == 9, "Last leaf index %v, expected : 9",
		lastLeafIndex)

	// Test with kvdb returning error because update index was too old.
	fourthLevel := root + "/fourth"
	kv.Create(fourthLevel, []byte(strconv.Itoa(0)), 0)
	for i := 1; i < 2000; i++ {
		kv.Update(fourthLevel, []byte(strconv.Itoa(i)), 0)
	}
	collector, _ = kvdb.NewUpdatesCollector(kv, fourthLevel, kvp.ModifiedIndex)
	kv.Update(fourthLevel, []byte(strconv.Itoa(2000)), 0)
	time.Sleep(500 * time.Millisecond)
	cb = func(prefix string, opaque interface{}, kvp *kvdb.KVPair,
		err error) error {
		fmt.Printf("Error is %v", err)
		assert.True(t, err != nil, "Error is nil %v", err)
		return nil
	}
	replayCb[0].WatchCB = cb
	collector.ReplayUpdates(replayCb)
}

func grantRevokeUser(kvRootUser kvdb.Kvdb, datastoreInit kvdb.DatastoreInit, t *testing.T) {
	fmt.Println("grantRevokeUser")

	kvRootUser.Create("allow1/foo", []byte("bar"), 0)
	kvRootUser.Create("allow2/foo", []byte("bar"), 0)
	kvRootUser.Create("disallow/foo", []byte("bar"), 0)
	err := kvRootUser.GrantUserAccess("test", kvdb.ReadWritePermission, "allow1/*")
	assert.NoError(t, err, "Error in Grant User")
	err = kvRootUser.GrantUserAccess("test", kvdb.ReadWritePermission, "allow2/*")
	assert.NoError(t, err, "Error in Grant User")
	err = kvRootUser.GrantUserAccess("test", kvdb.ReadWritePermission, "disallow/*")
	assert.NoError(t, err, "Error in Grant User")
	err = kvRootUser.RevokeUsersAccess("test", kvdb.ReadWritePermission, "disallow/*")
	assert.NoError(t, err, "Error in Revoke User")

	options := make(map[string]string)
	options[kvdb.UsernameKey] = "test"
	options[kvdb.PasswordKey] = "test123"
	options[kvdb.CAFileKey] = "/etc/pwx/pwx-ca.crt"
	options[kvdb.CertFileKey] = "/etc/pwx/pwx-user-cert.crt"
	options[kvdb.CertKeyFileKey] = "/etc/pwx/pwx-user-key.key"
	machines := []string{"https://192.168.56.101:2379"}
	kvTestUser, _ := datastoreInit("pwx/test", machines, options, fatalErrorCb())

	actual := "actual"
	_, err = kvTestUser.Put("allow1/foo", []byte(actual), 0)
	assert.NoError(t, err, "Error in writing to allowed tree")
	kvPair, err := kvTestUser.Get("allow1/foo")
	assert.NoError(t, err, "Error in accessing allowed values")
	if err == nil {
		assert.Equal(t, string(kvPair.Value), "actual")
	}

	_, err = kvTestUser.Put("allow2/foo", []byte(actual), 0)
	assert.NoError(t, err, "Error in writing to allowed tree")
	kvPair, err = kvTestUser.Get("allow2/foo")
	assert.NoError(t, err, "Error in accessing allowed values")
	if err == nil {
		assert.Equal(t, string(kvPair.Value), "actual")
	}

	actual2 := "actual2"
	_, err = kvTestUser.Put("disallow/foo", []byte(actual2), 0)
	assert.Error(t, err, "Expected error in writing to disallowed tree")
	kvPair, err = kvTestUser.Get("disallow/foo")
	assert.Error(t, err, "Expected error in accessing disallowed values")

	kvRootUser.DeleteTree("allow1")
	kvRootUser.DeleteTree("allow2")
	kvRootUser.DeleteTree("disallow")
}

func removeUser(kv kvdb.Kvdb, t *testing.T) {
	fmt.Println("removeUser")

	err := kv.RemoveUser("test")
	assert.NoError(t, err, "Error in Removing User")
}

func addUser(kv kvdb.Kvdb, t *testing.T) {
	fmt.Println("addUser")

	err := kv.AddUser("test", "test123")
	assert.NoError(t, err, "Error in Adding User")
}

func cas(kv kvdb.Kvdb, t *testing.T) {
	fmt.Println("\ncas")

	key := "foo/docker"
	val := "great"
	defer func() {
		kv.Delete(key)
	}()

	kvPair, err := kv.Put(key, []byte(val), 0)
	assert.NoError(t, err, "Unxpected error in Put")

	kvPair, err = kv.Get(key)
	assert.NoError(t, err, "Failed in Get")

	_, err = kv.CompareAndSet(kvPair, kvdb.KVFlags(0), []byte("badval"))
	assert.Error(t, err, "CompareAndSet should fail on an incorrect previous value")
	//assert.EqualError(t, err, kvdb.ErrValueMismatch.Error(), "CompareAndSet should return value mismatch error")

	copyKVPair := *kvPair
	copyKVPair.ModifiedIndex++
	_, err = kv.CompareAndSet(&copyKVPair, kvdb.KVModifiedIndex, nil)
	assert.Error(t, err, "CompareAndSet should fail on an incorrect modified index")

	//kvPair.ModifiedIndex--
	copyKVPair.ModifiedIndex--
	kvPair, err = kv.CompareAndSet(&copyKVPair, kvdb.KVModifiedIndex, nil)
	assert.NoError(t, err, "CompareAndSet should succeed on an correct modified index")

	kvPairNew, err := kv.CompareAndSet(kvPair, kvdb.KVFlags(0), []byte(val))
	if err != nil {
		// consul does not handle this kind of compare and set
		assert.EqualError(t, err, kvdb.ErrNotSupported.Error(), "Invalid error returned : %v", err)
	} else {
		assert.NoError(t, err, "CompareAndSet should succeed on an correct value")
	}

	if kvPairNew != nil {
		kvPair = kvPairNew
	}

	kvPair, err = kv.CompareAndSet(kvPair, kvdb.KVModifiedIndex, []byte(val))
	assert.NoError(t, err, "CompareAndSet should succeed on an correct value and modified index")
}

func watchWithIndex(kv kvdb.Kvdb, t *testing.T) {
	fmt.Println("\nwatchWithIndex")

	tree := "indexTree"
	subtree := tree + "/subtree"
	key := subtree + "/watchWithIndex"
	key1 := subtree + "/watchWithIndex21"

	kv.DeleteTree(tree)

	kvp, err := kv.Create(key, []byte("bar"), 0)
	assert.NoError(t, err, "Unexpected error in create: %v", err)

	time.Sleep(time.Millisecond * 100)

	waitIndex := kvp.ModifiedIndex + 2
	watchData := watchData{
		t:          t,
		key:        key,
		localIndex: waitIndex,
	}

	err = kv.WatchTree(tree, waitIndex, &watchData, watchWithIndexCb)
	if err != nil {
		fmt.Printf("Cannot test watchTree for watchWithIndex: %v", err)
		return
	}
	// Should not get updates for these
	kv.Put(key, []byte("bar1"), 0)
	kv.Put(key, []byte("bar1"), 0)
	// Should get updates for these
	kv.Put(key, []byte("bar2"), 0)
	kv.Create(key1, []byte("bar"), 0)
	kv.Delete(key)
}

func randomUpdate(kv kvdb.Kvdb, w *watchData) {
	for w.watchStopped == false {
		kv.Put("randomKey", []byte("bar"), 0)
		time.Sleep(time.Millisecond * 80)
	}
}

func watchKey(kv kvdb.Kvdb, t *testing.T) {
	fmt.Println("\nwatchKey")

	watchData := watchData{
		t:          t,
		key:        "tree/key1",
		otherKey:   "tree/otherKey1",
		stop:       "stop",
		iterations: 2,
	}

	kv.Delete(watchData.key)
	kv.Delete(watchData.otherKey)
	// First create a key. We should not get update for this create.
	_, err := kv.Create(watchData.otherKey, []byte("bar"), 0)
	// Let the create operation finish and then start the watch
	time.Sleep(time.Second)

	err = kv.WatchKey(watchData.otherKey, 0, &watchData, watchFn)
	if err != nil {
		fmt.Printf("Cannot test watchKey: %v\n", err)
		return
	}

	err = kv.WatchKey(watchData.key, 0, &watchData, watchFn)
	if err != nil {
		fmt.Printf("Cannot test watchKey: %v\n", err)
		return
	}

	go watchUpdate(kv, &watchData)

	for watchData.watchStopped == false {
		time.Sleep(time.Millisecond * 100)
	}

	// Stop the second watch
	atomic.SwapInt32(&watchData.whichKey, 0)
	watchData.action = kvdb.KVCreate
	_, err = kv.Create(watchData.otherKey, []byte(watchData.stop), 0)
}

func deleteKey(kv kvdb.Kvdb, t *testing.T) {
	fmt.Println("deleteKey")

	key := "delete_key"
	_, err := kv.Delete(key)

	_, err = kv.Put(key, []byte("delete_me"), 0)
	assert.NoError(t, err, "Unexpected error on Put")

	_, err = kv.Get(key)
	assert.NoError(t, err, "Unexpected error on Get")

	_, err = kv.Delete(key)
	assert.NoError(t, err, "Unexpected error on Delete")

	_, err = kv.Get(key)
	assert.Error(t, err, "Get should fail on deleted key")

	_, err = kv.Delete(key)
	assert.Error(t, err, "Delete should fail on non existent key")
}

func enumerate(kv kvdb.Kvdb, t *testing.T) {

	fmt.Println("enumerate")

	prefix := "enumerate"
	keys := map[string]string{
		prefix + "/1cbc9a98-072a-4793-8608-01ab43db96c8": "bar",
		prefix + "/foo":                                  "baz",
	}

	kv.DeleteTree(prefix)
	defer func() {
		kv.DeleteTree(prefix)
	}()

	errPairs, err := kv.Enumerate(prefix)
	assert.Equal(t, 0, len(errPairs), "Expected 0 pairs")

	folderKey := prefix + "/folder/"
	_, err = kv.Put(folderKey, []byte(""), 0)
	assert.NoError(t, err, "Unexpected error on Put")
	kvPairs, err := kv.Enumerate(folderKey)
	assert.Equal(t, nil, err, "Unexpected error on Enumerate")
	kv.DeleteTree(prefix)

	for key, val := range keys {
		_, err := kv.Put(key, []byte(val), 0)
		assert.NoError(t, err, "Unexpected error on Put")
	}
	kvPairs, err = kv.Enumerate(prefix)
	assert.NoError(t, err, "Unexpected error on Enumerate")

	assert.Equal(t, len(kvPairs), len(keys),
		"Expecting %d keys under %s got: %d",
		len(keys), prefix, len(kvPairs))

	for i := range kvPairs {
		v, ok := keys[kvPairs[i].Key]
		assert.True(t, ok, "unexpected kvpair (%s)->(%s)",
			kvPairs[i].Key, kvPairs[i].Value)
		assert.Equal(t, v, string(kvPairs[i].Value),
			"Invalid kvpair (%s)->(%s) expect value %s",
			kvPairs[i].Key, kvPairs[i].Value, v)
	}
}