Merge pull request #887 from abronan/pkg_store_test

store: Add unit testing for pkg/store on Travis

.travis.yml

@@ -19,7 +19,15 @@ install:
   - go get github.com/golang/lint/golint
   - go get github.com/GeertJohan/fgt
 
+before_script:
+  - script/travis_consul.sh 0.5.2 
+  - script/travis_etcd.sh 2.0.11
+  - script/travis_zk.sh 3.4.6
+
 script:
+  - ./consul agent -server -bootstrap-expect 1 -data-dir /tmp/consul -config-file=./config.json 1>/dev/null &
+  - ./etcd/etcd --listen-client-urls 'http://0.0.0.0:4001' --advertise-client-urls 'http://127.0.0.1:4001' >/dev/null 2>&1 &
+  - ./zk/bin/zkServer.sh start ./zk/conf/zoo.cfg 1> /dev/null
   - script/validate-gofmt
   - go vet ./...
   - fgt golint ./...

pkg/store/consul.go

@@ -16,10 +16,6 @@ const (
 	// watched key has changed.  This affects the minimum time it takes to
 	// cancel a watch.
 	DefaultWatchWaitTime = 15 * time.Second
-
-	// MinimumTimeToLive is the minimum TTL value allowed by Consul for
-	// Ephemeral entries
-	MinimumTimeToLive = 10 * time.Second
 )
 
 // Consul embeds the client and watches
@@ -56,9 +52,7 @@ func InitializeConsul(endpoints []string, options *Config) (Store, error) {
 			s.setTimeout(options.ConnectionTimeout)
 		}
 		if options.EphemeralTTL != 0 {
-			if err := s.setEphemeralTTL(options.EphemeralTTL); err != nil {
-				return nil, err
-			}
+			s.setEphemeralTTL(options.EphemeralTTL)
 		}
 	}
 
@@ -87,12 +81,8 @@ func (s *Consul) setTimeout(time time.Duration) {
 }
 
 // SetEphemeralTTL sets the ttl for ephemeral nodes
-func (s *Consul) setEphemeralTTL(ttl time.Duration) error {
-	if ttl < MinimumTimeToLive {
-		return ErrInvalidTTL
-	}
+func (s *Consul) setEphemeralTTL(ttl time.Duration) {
 	s.ephemeralTTL = ttl
-	return nil
 }
 
 // createEphemeralSession creates the global session
@@ -406,3 +396,8 @@ func (s *Consul) AtomicDelete(key string, previous *KVPair) (bool, error) {
 	}
 	return true, nil
 }
+
+// Close closes the client connection
+func (s *Consul) Close() {
+	return
+}

pkg/store/consul_test.go

@@ -0,0 +1,69 @@
+package store
+
+import (
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func makeConsulClient(t *testing.T) Store {
+	client := "localhost:8500"
+
+	kv, err := NewStore(
+		CONSUL,
+		[]string{client},
+		&Config{
+			ConnectionTimeout: 3 * time.Second,
+			EphemeralTTL:      2 * time.Second,
+		},
+	)
+	if err != nil {
+		t.Fatalf("cannot create store: %v", err)
+	}
+
+	return kv
+}
+
+func TestConsulStore(t *testing.T) {
+	kv := makeConsulClient(t)
+
+	testStore(t, kv)
+}
+
+func TestCreateEphemeralSession(t *testing.T) {
+	kv := makeConsulClient(t)
+
+	consul := kv.(*Consul)
+
+	err := consul.createEphemeralSession()
+	assert.NoError(t, err)
+	assert.NotEqual(t, consul.ephemeralSession, "")
+}
+
+func TestCheckActiveSession(t *testing.T) {
+	kv := makeConsulClient(t)
+
+	consul := kv.(*Consul)
+
+	key := "foo"
+	value := []byte("bar")
+
+	// Put the first key with the Ephemeral flag
+	err := kv.Put(key, value, &WriteOptions{Ephemeral: true})
+	assert.NoError(t, err)
+
+	// Session should not be empty
+	session, err := consul.checkActiveSession(key)
+	assert.NoError(t, err)
+	assert.NotEqual(t, session, "")
+
+	// Delete the key
+	err = kv.Delete(key)
+	assert.NoError(t, err)
+
+	// Check the session again, it should return nothing
+	session, err = consul.checkActiveSession(key)
+	assert.NoError(t, err)
+	assert.Equal(t, session, "")
+}

pkg/store/etcd.go

@@ -114,7 +114,7 @@ func (s *Etcd) Get(key string) (*KVPair, error) {
 		}
 		return nil, err
 	}
-	return &KVPair{result.Node.Key, []byte(result.Node.Value), result.Node.ModifiedIndex}, nil
+	return &KVPair{key, []byte(result.Node.Value), result.Node.ModifiedIndex}, nil
 }
 
 // Put a value at "key"
@@ -166,8 +166,6 @@ func (s *Etcd) Exists(key string) (bool, error) {
 // Upon creating a watch, the current value will be sent to the channel.
 // Providing a non-nil stopCh can be used to stop watching.
 func (s *Etcd) Watch(key string, stopCh <-chan struct{}) (<-chan *KVPair, error) {
-	key = normalize(key)
-
 	// Get the current value
 	current, err := s.Get(key)
 	if err != nil {
@@ -178,7 +176,7 @@ func (s *Etcd) Watch(key string, stopCh <-chan struct{}) (<-chan *KVPair, error)
 	// Note: etcd will send the current value through the channel.
 	etcdWatchCh := make(chan *etcd.Response)
 	etcdStopCh := make(chan bool)
-	go s.client.Watch(key, 0, false, etcdWatchCh, etcdStopCh)
+	go s.client.Watch(normalize(key), 0, false, etcdWatchCh, etcdStopCh)
 
 	// Adapter goroutine: The goal here is to convert wathever format etcd is
 	// using into our interface.
@@ -193,7 +191,7 @@ func (s *Etcd) Watch(key string, stopCh <-chan struct{}) (<-chan *KVPair, error)
 			select {
 			case result := <-etcdWatchCh:
 				watchCh <- &KVPair{
-					result.Node.Key,
+					key,
 					[]byte(result.Node.Value),
 					result.Node.ModifiedIndex,
 				}
@@ -211,8 +209,6 @@ func (s *Etcd) Watch(key string, stopCh <-chan struct{}) (<-chan *KVPair, error)
 // Upon creating a watch, the current value will be sent to the channel.
 // Providing a non-nil stopCh can be used to stop watching.
 func (s *Etcd) WatchTree(prefix string, stopCh <-chan struct{}) (<-chan []*KVPair, error) {
-	prefix = normalize(prefix)
-
 	// Get the current value
 	current, err := s.List(prefix)
 	if err != nil {
@@ -222,7 +218,7 @@ func (s *Etcd) WatchTree(prefix string, stopCh <-chan struct{}) (<-chan []*KVPai
 	// Start an etcd watch.
 	etcdWatchCh := make(chan *etcd.Response)
 	etcdStopCh := make(chan bool)
-	go s.client.Watch(prefix, 0, true, etcdWatchCh, etcdStopCh)
+	go s.client.Watch(normalize(prefix), 0, true, etcdWatchCh, etcdStopCh)
 
 	// Adapter goroutine: The goal here is to convert wathever format etcd is
 	// using into our interface.
@@ -296,7 +292,8 @@ func (s *Etcd) List(prefix string) ([]*KVPair, error) {
 	}
 	kv := []*KVPair{}
 	for _, n := range resp.Node.Nodes {
-		kv = append(kv, &KVPair{n.Key, []byte(n.Value), n.ModifiedIndex})
+		key := strings.TrimLeft(n.Key, "/")
+		kv = append(kv, &KVPair{key, []byte(n.Value), n.ModifiedIndex})
 	}
 	return kv, nil
 }
@@ -431,3 +428,8 @@ func (l *etcdLock) Unlock() error {
 	}
 	return nil
 }
+
+// Close closes the client connection
+func (s *Etcd) Close() {
+	return
+}

pkg/store/etcd_test.go

@@ -0,0 +1,30 @@
+package store
+
+import (
+	"testing"
+	"time"
+)
+
+func makeEtcdClient(t *testing.T) Store {
+	client := "localhost:4001"
+
+	kv, err := NewStore(
+		ETCD,
+		[]string{client},
+		&Config{
+			ConnectionTimeout: 3 * time.Second,
+			EphemeralTTL:      2 * time.Second,
+		},
+	)
+	if err != nil {
+		t.Fatalf("cannot create store: %v", err)
+	}
+
+	return kv
+}
+
+func TestEtcdStore(t *testing.T) {
+	kv := makeEtcdClient(t)
+
+	testStore(t, kv)
+}

pkg/store/mock.go

@@ -102,3 +102,8 @@ func (l *MockLock) Unlock() error {
 	args := l.Mock.Called()
 	return args.Error(0)
 }
+
+// Close mock
+func (s *Mock) Close() {
+	return
+}

pkg/store/store.go

@@ -95,6 +95,9 @@ type Store interface {
 
 	// Atomic delete of a single value
 	AtomicDelete(key string, previous *KVPair) (bool, error)
+
+	// Close the store connection
+	Close()
 }
 
 // KVPair represents {Key, Value, Lastindex} tuple

pkg/store/store_test.go

@@ -0,0 +1,401 @@
+package store
+
+import (
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func testStore(t *testing.T, kv Store) {
+	testPutGetDelete(t, kv)
+	testWatch(t, kv)
+	testWatchTree(t, kv)
+	testAtomicPut(t, kv)
+	testAtomicDelete(t, kv)
+	testLockUnlock(t, kv)
+	testPutEphemeral(t, kv)
+	testList(t, kv)
+	testDeleteTree(t, kv)
+}
+
+func testPutGetDelete(t *testing.T, kv Store) {
+	key := "foo"
+	value := []byte("bar")
+
+	// Put the key
+	err := kv.Put(key, value, nil)
+	assert.NoError(t, err)
+
+	// Get should return the value and an incremented index
+	pair, err := kv.Get(key)
+	assert.NoError(t, err)
+	if assert.NotNil(t, pair) {
+		assert.NotNil(t, pair.Value)
+	}
+	assert.Equal(t, pair.Value, value)
+	assert.NotEqual(t, pair.LastIndex, 0)
+
+	// Delete the key
+	err = kv.Delete(key)
+	assert.NoError(t, err)
+
+	// Get should fail
+	pair, err = kv.Get(key)
+	assert.Error(t, err)
+	assert.Nil(t, pair)
+}
+
+func testWatch(t *testing.T, kv Store) {
+	key := "hello"
+	value := []byte("world")
+	newValue := []byte("world!")
+
+	// Put the key
+	err := kv.Put(key, value, nil)
+	assert.NoError(t, err)
+
+	stopCh := make(<-chan struct{})
+	events, err := kv.Watch(key, stopCh)
+	assert.NoError(t, err)
+	assert.NotNil(t, events)
+
+	// Update loop
+	go func() {
+		timeout := time.After(1 * time.Second)
+		tick := time.Tick(250 * time.Millisecond)
+		for {
+			select {
+			case <-timeout:
+				return
+			case <-tick:
+				err := kv.Put(key, newValue, nil)
+				if assert.NoError(t, err) {
+					continue
+				}
+				return
+			}
+		}
+	}()
+
+	// Check for updates
+	timeout := time.After(2 * time.Second)
+	eventCount := 1
+	for {
+		select {
+		case event := <-events:
+			assert.NotNil(t, event)
+			if eventCount == 1 {
+				assert.Equal(t, event.Key, key)
+				assert.Equal(t, event.Value, value)
+			} else {
+				assert.Equal(t, event.Key, key)
+				assert.Equal(t, event.Value, newValue)
+			}
+			eventCount++
+			// We received all the events we wanted to check
+			if eventCount >= 4 {
+				return
+			}
+		case <-timeout:
+			t.Fatal("Timeout reached")
+			return
+		}
+	}
+}
+
+func testWatchTree(t *testing.T, kv Store) {
+	dir := "tree"
+
+	node1 := "tree/node1"
+	value1 := []byte("node1")
+
+	node2 := "tree/node2"
+	value2 := []byte("node2")
+
+	node3 := "tree/node3"
+	value3 := []byte("node3")
+
+	err := kv.Put(node1, value1, nil)
+	assert.NoError(t, err)
+	err = kv.Put(node2, value2, nil)
+	assert.NoError(t, err)
+	err = kv.Put(node3, value3, nil)
+	assert.NoError(t, err)
+
+	stopCh := make(<-chan struct{})
+	events, err := kv.WatchTree(dir, stopCh)
+	assert.NoError(t, err)
+	assert.NotNil(t, events)
+
+	// Update loop
+	go func() {
+		timeout := time.After(250 * time.Millisecond)
+		for {
+			select {
+			case <-timeout:
+				err := kv.Delete(node3)
+				assert.NoError(t, err)
+				return
+			}
+		}
+	}()
+
+	// Check for updates
+	timeout := time.After(4 * time.Second)
+	for {
+		select {
+		case event := <-events:
+			assert.NotNil(t, event)
+			// We received the Delete event on a child node
+			// Exit test successfully
+			if len(event) == 2 {
+				return
+			}
+		case <-timeout:
+			t.Fatal("Timeout reached")
+			return
+		}
+	}
+}
+
+func testAtomicPut(t *testing.T, kv Store) {
+	key := "hello"
+	value := []byte("world")
+
+	// Put the key
+	err := kv.Put(key, value, nil)
+	assert.NoError(t, err)
+
+	// Get should return the value and an incremented index
+	pair, err := kv.Get(key)
+	assert.NoError(t, err)
+	if assert.NotNil(t, pair) {
+		assert.NotNil(t, pair.Value)
+	}
+	assert.Equal(t, pair.Value, value)
+	assert.NotEqual(t, pair.LastIndex, 0)
+
+	// This CAS should succeed
+	success, _, err := kv.AtomicPut("hello", []byte("WORLD"), pair, nil)
+	assert.NoError(t, err)
+	assert.True(t, success)
+
+	// This CAS should fail
+	pair.LastIndex = 0
+	success, _, err = kv.AtomicPut("hello", []byte("WORLDWORLD"), pair, nil)
+	assert.Error(t, err)
+	assert.False(t, success)
+}
+
+func testAtomicDelete(t *testing.T, kv Store) {
+	key := "atomic"
+	value := []byte("world")
+
+	// Put the key
+	err := kv.Put(key, value, nil)
+	assert.NoError(t, err)
+
+	// Get should return the value and an incremented index
+	pair, err := kv.Get(key)
+	assert.NoError(t, err)
+	if assert.NotNil(t, pair) {
+		assert.NotNil(t, pair.Value)
+	}
+	assert.Equal(t, pair.Value, value)
+	assert.NotEqual(t, pair.LastIndex, 0)
+
+	tempIndex := pair.LastIndex
+
+	// AtomicDelete should fail
+	pair.LastIndex = 0
+	success, err := kv.AtomicDelete(key, pair)
+	assert.Error(t, err)
+	assert.False(t, success)
+
+	// AtomicDelete should succeed
+	pair.LastIndex = tempIndex
+	success, err = kv.AtomicDelete(key, pair)
+	assert.NoError(t, err)
+	assert.True(t, success)
+}
+
+func testLockUnlock(t *testing.T, kv Store) {
+	t.Parallel()
+
+	key := "foo"
+	value := []byte("bar")
+
+	// We should be able to create a new lock on key
+	lock, err := kv.NewLock(key, &LockOptions{Value: value})
+	assert.NoError(t, err)
+	assert.NotNil(t, lock)
+
+	// Lock should successfully succeed or block
+	lockChan, err := lock.Lock()
+	assert.NoError(t, err)
+	assert.NotNil(t, lockChan)
+
+	// Get should work
+	pair, err := kv.Get(key)
+	assert.NoError(t, err)
+	if assert.NotNil(t, pair) {
+		assert.NotNil(t, pair.Value)
+	}
+	assert.Equal(t, pair.Value, value)
+	assert.NotEqual(t, pair.LastIndex, 0)
+
+	// Unlock should succeed
+	err = lock.Unlock()
+	assert.NoError(t, err)
+
+	// Get should work
+	pair, err = kv.Get(key)
+	assert.NoError(t, err)
+	if assert.NotNil(t, pair) {
+		assert.NotNil(t, pair.Value)
+	}
+	assert.Equal(t, pair.Value, value)
+	assert.NotEqual(t, pair.LastIndex, 0)
+}
+
+// FIXME Gracefully handle Zookeeper
+func testPutEphemeral(t *testing.T, kv Store) {
+	// Zookeeper: initialize client here (Close() hangs otherwise)
+	zookeeper := false
+	if _, ok := kv.(*Zookeeper); ok {
+		zookeeper = true
+		kv = makeZkClient(t)
+	}
+
+	firstKey := "first"
+	firstValue := []byte("foo")
+
+	secondKey := "second"
+	secondValue := []byte("bar")
+
+	// Put the first key with the Ephemeral flag
+	err := kv.Put(firstKey, firstValue, &WriteOptions{Ephemeral: true})
+	assert.NoError(t, err)
+
+	// Put a second key with the Ephemeral flag
+	err = kv.Put(secondKey, secondValue, &WriteOptions{Ephemeral: true})
+	assert.NoError(t, err)
+
+	// Get on firstKey should work
+	pair, err := kv.Get(firstKey)
+	assert.NoError(t, err)
+	assert.NotNil(t, pair)
+
+	// Get on secondKey should work
+	pair, err = kv.Get(secondKey)
+	assert.NoError(t, err)
+	assert.NotNil(t, pair)
+
+	// Zookeeper: close client connection
+	if zookeeper {
+		kv.Close()
+	}
+
+	// Let the session expire
+	time.Sleep(5 * time.Second)
+
+	// Zookeeper: re-create the client
+	if zookeeper {
+		kv = makeZkClient(t)
+	}
+
+	// Get on firstKey shouldn't work
+	pair, err = kv.Get(firstKey)
+	assert.Error(t, err)
+	assert.Nil(t, pair)
+
+	// Get on secondKey shouldn't work
+	pair, err = kv.Get(secondKey)
+	assert.Error(t, err)
+	assert.Nil(t, pair)
+}
+
+func testList(t *testing.T, kv Store) {
+	prefix := "nodes"
+
+	firstKey := "nodes/first"
+	firstValue := []byte("first")
+
+	secondKey := "nodes/second"
+	secondValue := []byte("second")
+
+	// Put the first key
+	err := kv.Put(firstKey, firstValue, nil)
+	assert.NoError(t, err)
+
+	// Put the second key
+	err = kv.Put(secondKey, secondValue, nil)
+	assert.NoError(t, err)
+
+	// List should work and return the two correct values
+	pairs, err := kv.List(prefix)
+	assert.NoError(t, err)
+	if assert.NotNil(t, pairs) {
+		assert.Equal(t, len(pairs), 2)
+	}
+
+	// Check pairs, those are not necessarily in Put order
+	for _, pair := range pairs {
+		if pair.Key == firstKey {
+			assert.Equal(t, pair.Value, firstValue)
+		}
+		if pair.Key == secondKey {
+			assert.Equal(t, pair.Value, secondValue)
+		}
+	}
+}
+
+func testDeleteTree(t *testing.T, kv Store) {
+	prefix := "nodes"
+
+	firstKey := "nodes/first"
+	firstValue := []byte("first")
+
+	secondKey := "nodes/second"
+	secondValue := []byte("second")
+
+	// Put the first key
+	err := kv.Put(firstKey, firstValue, nil)
+	assert.NoError(t, err)
+
+	// Put the second key
+	err = kv.Put(secondKey, secondValue, nil)
+	assert.NoError(t, err)
+
+	// Get should work on the first Key
+	pair, err := kv.Get(firstKey)
+	assert.NoError(t, err)
+	if assert.NotNil(t, pair) {
+		assert.NotNil(t, pair.Value)
+	}
+	assert.Equal(t, pair.Value, firstValue)
+	assert.NotEqual(t, pair.LastIndex, 0)
+
+	// Get should work on the second Key
+	pair, err = kv.Get(secondKey)
+	assert.NoError(t, err)
+	if assert.NotNil(t, pair) {
+		assert.NotNil(t, pair.Value)
+	}
+	assert.Equal(t, pair.Value, secondValue)
+	assert.NotEqual(t, pair.LastIndex, 0)
+
+	// Delete Values under directory `nodes`
+	err = kv.DeleteTree(prefix)
+	assert.NoError(t, err)
+
+	// Get should fail on both keys
+	pair, err = kv.Get(firstKey)
+	assert.Error(t, err)
+	assert.Nil(t, pair)
+
+	pair, err = kv.Get(secondKey)
+	assert.Error(t, err)
+	assert.Nil(t, pair)
+}

pkg/store/zookeeper.go

@@ -296,7 +296,7 @@ func (l *zookeeperLock) Lock() (<-chan struct{}, error) {
 		_, err = l.client.Set(l.key, l.value, -1)
 	}
 
-	return nil, err
+	return make(chan struct{}), err
 }
 
 // Unlock released the lock. It is an error to call this
@@ -304,3 +304,8 @@ func (l *zookeeperLock) Lock() (<-chan struct{}, error) {
 func (l *zookeeperLock) Unlock() error {
 	return l.lock.Unlock()
 }
+
+// Close closes the client connection
+func (s *Zookeeper) Close() {
+	s.client.Close()
+}

pkg/store/zookeeper_test.go

@@ -0,0 +1,30 @@
+package store
+
+import (
+	"testing"
+	"time"
+)
+
+func makeZkClient(t *testing.T) Store {
+	client := "localhost:2181"
+
+	kv, err := NewStore(
+		ZK,
+		[]string{client},
+		&Config{
+			ConnectionTimeout: 3 * time.Second,
+			EphemeralTTL:      2 * time.Second,
+		},
+	)
+	if err != nil {
+		t.Fatalf("cannot create store: %v", err)
+	}
+
+	return kv
+}
+
+func TestZkStore(t *testing.T) {
+	kv := makeZkClient(t)
+
+	testStore(t, kv)
+}

script/travis_consul.sh

@@ -0,0 +1,18 @@
+#!/bin/bash
+
+if [  $# -gt 0 ] ; then
+    CONSUL_VERSION="$1"
+else
+    CONSUL_VERSION="0.5.2"
+fi
+
+# install consul
+wget "https://dl.bintray.com/mitchellh/consul/${CONSUL_VERSION}_linux_amd64.zip"
+unzip "${CONSUL_VERSION}_linux_amd64.zip"
+
+# make config for minimum ttl
+touch config.json
+echo "{\"session_ttl_min\": \"2s\"}" >> config.json
+
+# check
+./consul --version

script/travis_etcd.sh

@@ -0,0 +1,11 @@
+#!/bin/bash
+
+if [  $# -gt 0 ] ; then
+    ETCD_VERSION="$1"
+else
+    ETCD_VERSION="2.0.11"
+fi
+
+curl -L https://github.com/coreos/etcd/releases/download/v$ETCD_VERSION/etcd-v$ETCD_VERSION-linux-amd64.tar.gz -o etcd-v$ETCD_VERSION-linux-amd64.tar.gz
+tar xzvf etcd-v$ETCD_VERSION-linux-amd64.tar.gz
+mv etcd-v$ETCD_VERSION-linux-amd64 etcd

script/travis_zk.sh

@@ -0,0 +1,12 @@
+#!/bin/bash
+
+if [  $# -gt 0 ] ; then
+    ZK_VERSION="$1"
+else
+    ZK_VERSION="3.4.6"
+fi
+
+wget "http://mirrors.ukfast.co.uk/sites/ftp.apache.org/zookeeper/stable/zookeeper-${ZK_VERSION}.tar.gz"
+tar -xvf "zookeeper-${ZK_VERSION}.tar.gz"
+mv zookeeper-$ZK_VERSION zk
+mv ./zk/conf/zoo_sample.cfg ./zk/conf/zoo.cfg