discovery: Watch tests for file and some other tests.

Signed-off-by: Andrea Luzzardi <aluzzardi@gmail.com>

discovery/discovery_test.go

@@ -8,9 +8,9 @@ import (
 
 func TestNewEntry(t *testing.T) {
 	entry, err := NewEntry("127.0.0.1:2375")
-	assert.Equal(t, entry.Host, "127.0.0.1")
-	assert.Equal(t, entry.Port, "2375")
 	assert.NoError(t, err)
+	assert.True(t, entry.Equals(&Entry{Host: "127.0.0.1", Port: "2375"}))
+	assert.Equal(t, entry.String(), "127.0.0.1:2375")
 
 	_, err = NewEntry("127.0.0.1")
 	assert.Error(t, err)
@@ -44,11 +44,39 @@ func TestCreateEntries(t *testing.T) {
 	assert.NoError(t, err)
 
 	entries, err = CreateEntries([]string{"127.0.0.1:2375", "127.0.0.2:2375", ""})
-	assert.Equal(t, len(entries), 2)
-	assert.Equal(t, entries[0].String(), "127.0.0.1:2375")
-	assert.Equal(t, entries[1].String(), "127.0.0.2:2375")
 	assert.NoError(t, err)
+	expected := Entries{
+		&Entry{Host: "127.0.0.1", Port: "2375"},
+		&Entry{Host: "127.0.0.2", Port: "2375"},
+	}
+	assert.True(t, entries.Equals(expected))
 
 	_, err = CreateEntries([]string{"127.0.0.1", "127.0.0.2"})
 	assert.Error(t, err)
 }
+
+func TestEntriesEquality(t *testing.T) {
+	entries := Entries{
+		&Entry{Host: "127.0.0.1", Port: "2375"},
+		&Entry{Host: "127.0.0.2", Port: "2375"},
+	}
+
+	// Same
+	assert.True(t, entries.Equals(Entries{
+		&Entry{Host: "127.0.0.1", Port: "2375"},
+		&Entry{Host: "127.0.0.2", Port: "2375"},
+	}))
+
+	// Different size
+	assert.False(t, entries.Equals(Entries{
+		&Entry{Host: "127.0.0.1", Port: "2375"},
+		&Entry{Host: "127.0.0.2", Port: "2375"},
+		&Entry{Host: "127.0.0.3", Port: "2375"},
+	}))
+
+	// Different content
+	assert.False(t, entries.Equals(Entries{
+		&Entry{Host: "127.0.0.1", Port: "2375"},
+		&Entry{Host: "127.0.0.42", Port: "2375"},
+	}))
+}

discovery/file/file.go

@@ -62,9 +62,14 @@ func (s *Discovery) Watch(stopCh <-chan struct{}) (<-chan discovery.Entries, <-c
 	ticker := time.NewTicker(s.heartbeat)
 
 	go func() {
+		defer close(errCh)
+		defer close(ch)
+
 		// Send the initial entries if available.
 		currentEntries, err := s.fetch()
-		if err == nil {
+		if err != nil {
+			errCh <- err
+		} else {
 			ch <- currentEntries
 		}
 

discovery/file/file_test.go

@@ -1,15 +1,24 @@
 package file
 
 import (
+	"io/ioutil"
+	"os"
 	"testing"
 
+	"github.com/docker/swarm/discovery"
 	"github.com/stretchr/testify/assert"
 )
 
 func TestInitialize(t *testing.T) {
-	discovery := &Discovery{}
+	d := &Discovery{}
-	discovery.Initialize("/path/to/file", 0)
+	d.Initialize("/path/to/file", 0)
-	assert.Equal(t, discovery.path, "/path/to/file")
+	assert.Equal(t, d.path, "/path/to/file")
+}
+
+func TestNew(t *testing.T) {
+	d, err := discovery.New("file:///path/to/file", 0)
+	assert.NoError(t, err)
+	assert.Equal(t, d.(*Discovery).path, "/path/to/file")
 }
 
 func TestContent(t *testing.T) {
@@ -18,6 +27,7 @@ func TestContent(t *testing.T) {
 2.2.2.[2:4]:2222
 `
 	ips := parseFileContent([]byte(data))
+	assert.Len(t, ips, 5)
 	assert.Equal(t, ips[0], "1.1.1.1:1111")
 	assert.Equal(t, ips[1], "1.1.1.2:1111")
 	assert.Equal(t, ips[2], "2.2.2.2:2222")
@@ -40,7 +50,53 @@ func TestParsingContentsWithComments(t *testing.T) {
 ### test ###
 `
 	ips := parseFileContent([]byte(data))
-	assert.Equal(t, 2, len(ips))
+	assert.Len(t, ips, 2)
 	assert.Equal(t, "1.1.1.1:1111", ips[0])
 	assert.Equal(t, "3.3.3.3:3333", ips[1])
 }
+
+func TestWatch(t *testing.T) {
+	data := `
+1.1.1.1:1111
+2.2.2.2:2222
+`
+	expected := discovery.Entries{
+		&discovery.Entry{Host: "1.1.1.1", Port: "1111"},
+		&discovery.Entry{Host: "2.2.2.2", Port: "2222"},
+	}
+
+	// Create a temporary file and remove it.
+	tmp, err := ioutil.TempFile(os.TempDir(), "discovery-file-test")
+	assert.NoError(t, err)
+	assert.NoError(t, tmp.Close())
+	assert.NoError(t, os.Remove(tmp.Name()))
+
+	// Set up file discovery.
+	d := &Discovery{}
+	d.Initialize(tmp.Name(), 1)
+	stopCh := make(chan struct{})
+	ch, errCh := d.Watch(stopCh)
+
+	// Make sure it fires errors since the file doesn't exist.
+	assert.Error(t, <-errCh)
+	// We have to drain the error channel otherwise Watch will get stuck.
+	go func() {
+		for _ = range errCh {
+		}
+	}()
+
+	// Write the file and make sure we get the expected value back.
+	assert.NoError(t, ioutil.WriteFile(tmp.Name(), []byte(data), 0600))
+	assert.Equal(t, <-ch, expected)
+
+	// Add a new entry and look it up.
+	data += "\n3.3.3.3:3333\n"
+	expected = append(expected, &discovery.Entry{Host: "3.3.3.3", Port: "3333"})
+	assert.NoError(t, ioutil.WriteFile(tmp.Name(), []byte(data), 0600))
+	assert.Equal(t, <-ch, expected)
+
+	// Stop and make sure it closes all channels.
+	close(stopCh)
+	assert.Nil(t, <-ch)
+	assert.Nil(t, <-errCh)
+}

discovery/kv/kv.go

@@ -54,11 +54,7 @@ func (s *Discovery) Initialize(uris string, heartbeat time.Duration) error {
 			Timeout: s.heartbeat,
 		},
 	)
-	if err != nil {
+	return err
-		return err
-	}
-
-	return nil
 }
 
 // Watch the store until either there's a store error or we receive a stop request.
@@ -98,6 +94,9 @@ func (s *Discovery) Watch(stopCh <-chan struct{}) (<-chan discovery.Entries, <-c
 	errCh := make(chan error)
 
 	go func() {
+		defer close(ch)
+		defer close(errCh)
+
 		// Forever: Create a store watch, watch until we get an error and then try again.
 		// Will only stop if we receive a stopCh request.
 		for {

discovery/kv/kv_test.go

@@ -3,18 +3,27 @@ package kv
 import (
 	"testing"
 
+	"github.com/docker/swarm/pkg/store"
 	"github.com/stretchr/testify/assert"
 )
 
 func TestInitialize(t *testing.T) {
-	discoveryService := &Discovery{}
+	d := &Discovery{backend: store.MOCK}
+	assert.EqualError(t, d.Initialize("127.0.0.1", 0), "invalid format \"127.0.0.1\", missing <path>")
 
-	assert.Equal(t, discoveryService.Initialize("127.0.0.1", 0).Error(), "invalid format \"127.0.0.1\", missing <path>")
+	d = &Discovery{backend: store.MOCK}
-
+	assert.NoError(t, d.Initialize("127.0.0.1:1234/path", 0))
-	assert.Error(t, discoveryService.Initialize("127.0.0.1/path", 0))
+	s := d.store.(*store.Mock)
-	assert.Equal(t, discoveryService.prefix, "path")
+	assert.Len(t, s.Endpoints, 1)
-
+	assert.Equal(t, s.Endpoints[0], "127.0.0.1:1234")
-	assert.Error(t, discoveryService.Initialize("127.0.0.1,127.0.0.2,127.0.0.3/path", 0))
+	assert.Equal(t, d.prefix, "path")
-	assert.Equal(t, discoveryService.prefix, "path")
 
+	d = &Discovery{backend: store.MOCK}
+	assert.NoError(t, d.Initialize("127.0.0.1:1234,127.0.0.2:1234,127.0.0.3:1234/path", 0))
+	s = d.store.(*store.Mock)
+	assert.Len(t, s.Endpoints, 3)
+	assert.Equal(t, s.Endpoints[0], "127.0.0.1:1234")
+	assert.Equal(t, s.Endpoints[1], "127.0.0.2:1234")
+	assert.Equal(t, s.Endpoints[2], "127.0.0.3:1234")
+	assert.Equal(t, d.prefix, "path")
 }

discovery/nodes/nodes.go

@@ -35,6 +35,7 @@ func (s *Discovery) Initialize(uris string, _ time.Duration) error {
 func (s *Discovery) Watch(stopCh <-chan struct{}) (<-chan discovery.Entries, <-chan error) {
 	ch := make(chan discovery.Entries)
 	go func() {
+		defer close(ch)
 		ch <- s.entries
 		<-stopCh
 	}()

discovery/nodes/nodes_test.go

@@ -3,29 +3,41 @@ package nodes
 import (
 	"testing"
 
+	"github.com/docker/swarm/discovery"
 	"github.com/stretchr/testify/assert"
 )
 
-func TestInitialise(t *testing.T) {
+func TestInitialize(t *testing.T) {
-	discovery := &Discovery{}
+	d := &Discovery{}
-	discovery.Initialize("1.1.1.1:1111,2.2.2.2:2222", 0)
+	d.Initialize("1.1.1.1:1111,2.2.2.2:2222", 0)
-	assert.Equal(t, len(discovery.entries), 2)
+	assert.Equal(t, len(d.entries), 2)
-	assert.Equal(t, discovery.entries[0].String(), "1.1.1.1:1111")
+	assert.Equal(t, d.entries[0].String(), "1.1.1.1:1111")
-	assert.Equal(t, discovery.entries[1].String(), "2.2.2.2:2222")
+	assert.Equal(t, d.entries[1].String(), "2.2.2.2:2222")
 }
 
-func TestInitialiseWithPattern(t *testing.T) {
+func TestInitializeWithPattern(t *testing.T) {
-	discovery := &Discovery{}
+	d := &Discovery{}
-	discovery.Initialize("1.1.1.[1:2]:1111,2.2.2.[2:4]:2222", 0)
+	d.Initialize("1.1.1.[1:2]:1111,2.2.2.[2:4]:2222", 0)
-	assert.Equal(t, len(discovery.entries), 5)
+	assert.Equal(t, len(d.entries), 5)
-	assert.Equal(t, discovery.entries[0].String(), "1.1.1.1:1111")
+	assert.Equal(t, d.entries[0].String(), "1.1.1.1:1111")
-	assert.Equal(t, discovery.entries[1].String(), "1.1.1.2:1111")
+	assert.Equal(t, d.entries[1].String(), "1.1.1.2:1111")
-	assert.Equal(t, discovery.entries[2].String(), "2.2.2.2:2222")
+	assert.Equal(t, d.entries[2].String(), "2.2.2.2:2222")
-	assert.Equal(t, discovery.entries[3].String(), "2.2.2.3:2222")
+	assert.Equal(t, d.entries[3].String(), "2.2.2.3:2222")
-	assert.Equal(t, discovery.entries[4].String(), "2.2.2.4:2222")
+	assert.Equal(t, d.entries[4].String(), "2.2.2.4:2222")
+}
+
+func TestWatch(t *testing.T) {
+	d := &Discovery{}
+	d.Initialize("1.1.1.1:1111,2.2.2.2:2222", 0)
+	expected := discovery.Entries{
+		&discovery.Entry{Host: "1.1.1.1", Port: "1111"},
+		&discovery.Entry{Host: "2.2.2.2", Port: "2222"},
+	}
+	ch, _ := d.Watch(nil)
+	assert.True(t, expected.Equals(<-ch))
 }
 
 func TestRegister(t *testing.T) {
-	discovery := &Discovery{}
+	d := &Discovery{}
-	assert.Error(t, discovery.Register("0.0.0.0"))
+	assert.Error(t, d.Register("0.0.0.0"))
 }

discovery/token/token.go

@@ -72,6 +72,9 @@ func (s *Discovery) Watch(stopCh <-chan struct{}) (<-chan discovery.Entries, <-c
 	errCh := make(chan error)
 
 	go func() {
+		defer close(ch)
+		defer close(errCh)
+
 		// Send the initial entries if available.
 		currentEntries, err := s.fetch()
 		if err != nil {

pkg/store/mock.go

@@ -53,7 +53,7 @@ func (s *Mock) Watch(key string, stopCh <-chan struct{}) (<-chan *KVPair, error)
 // WatchTree mock
 func (s *Mock) WatchTree(prefix string, stopCh <-chan struct{}) (<-chan []*KVPair, error) {
 	args := s.Mock.Called(prefix, stopCh)
-	return args.Get(0).(<-chan []*KVPair), args.Error(1)
+	return args.Get(0).(chan []*KVPair), args.Error(1)
 }
 
 // CreateLock mock