Merge pull request #3610 from hashicorp/fix-skipped-tests

Fix skipped tests

command/agent/agent_test.go

@@ -2,11 +2,11 @@ package agent
 
 import (
 	"fmt"
-	"log"
 	"os"
 	"os/exec"
 	"path/filepath"
 	"strings"
+	"sync"
 	"testing"
 
 	"github.com/hashicorp/consul/agent"
@@ -81,30 +81,25 @@ func TestConfigFail(t *testing.T) {
 
 func TestRetryJoin(t *testing.T) {
 	t.Parallel()
-	t.Skip("fs: skipping tests that use cmd.Run until signal handling is fixed")
 	a := agent.NewTestAgent(t.Name(), "")
 	defer a.Shutdown()
 
-	cfg2 := agent.TestConfig()
-	tmpDir := testutil.TempDir(t, "consul")
-	defer os.RemoveAll(tmpDir)
-
-	doneCh := make(chan struct{})
 	shutdownCh := make(chan struct{})
 
-	defer func() {
-		close(shutdownCh)
-		<-doneCh
-	}()
+	var wg sync.WaitGroup
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
 
-	ui := cli.NewMockUi()
-	cmd := New(ui, "", version.Version, "", "", shutdownCh)
+		tmpDir := testutil.TempDir(t, "consul")
+		defer os.RemoveAll(tmpDir)
 
 		args := []string{
 			"-server",
 			"-bind", a.Config.BindAddr.String(),
 			"-data-dir", tmpDir,
-		"-node", fmt.Sprintf(`"%s"`, cfg2.NodeName),
+			"-node", "Node 11111111-1111-1111-1111-111111111111",
+			"-node-id", "11111111-1111-1111-1111-111111111111",
 			"-advertise", a.Config.BindAddr.String(),
 			"-retry-join", a.Config.SerfBindAddrLAN.String(),
 			"-retry-interval", "1s",
@@ -112,12 +107,16 @@ func TestRetryJoin(t *testing.T) {
 			"-retry-interval-wan", "1s",
 		}
 
-	go func() {
-		if code := cmd.Run(args); code != 0 {
-			log.Printf("bad: %d", code)
+		ui := cli.NewMockUi()
+		cmd := New(ui, "", version.Version, "", "", shutdownCh)
+		// closing shutdownCh triggers a SIGINT which triggers shutdown without leave
+		// which will return 1
+		if code := cmd.Run(args); code != 1 {
+			t.Log(ui.ErrorWriter.String())
+			t.Fatalf("bad: %d", code)
 		}
-		close(doneCh)
 	}()
+
 	retry.Run(t, func(r *retry.R) {
 		if got, want := len(a.LANMembers()), 2; got != want {
 			r.Fatalf("got %d LAN members want %d", got, want)
@@ -126,25 +125,23 @@ func TestRetryJoin(t *testing.T) {
 			r.Fatalf("got %d WAN members want %d", got, want)
 		}
 	})
+
+	close(shutdownCh)
+	wg.Wait()
 }
 
 func TestRetryJoinFail(t *testing.T) {
 	t.Parallel()
-	t.Skip("fs: skipping tests that use cmd.Run until signal handling is fixed")
-	cfg := agent.TestConfig()
 	tmpDir := testutil.TempDir(t, "consul")
 	defer os.RemoveAll(tmpDir)
 
-	shutdownCh := make(chan struct{})
-	defer close(shutdownCh)
-
 	ui := cli.NewMockUi()
-	cmd := New(ui, "", "", "", "", shutdownCh)
+	cmd := New(ui, "", "", "", "", nil)
 
 	args := []string{
-		"-bind", cfg.BindAddr.String(),
+		"-bind", "127.0.0.1",
 		"-data-dir", tmpDir,
-		"-retry-join", cfg.SerfBindAddrLAN.String(),
+		"-retry-join", "127.0.0.1:99",
 		"-retry-max", "1",
 		"-retry-interval", "10ms",
 	}
@@ -156,22 +153,17 @@ func TestRetryJoinFail(t *testing.T) {
 
 func TestRetryJoinWanFail(t *testing.T) {
 	t.Parallel()
-	t.Skip("fs: skipping tests that use cmd.Run until signal handling is fixed")
-	cfg := agent.TestConfig()
 	tmpDir := testutil.TempDir(t, "consul")
 	defer os.RemoveAll(tmpDir)
 
-	shutdownCh := make(chan struct{})
-	defer close(shutdownCh)
-
 	ui := cli.NewMockUi()
-	cmd := New(ui, "", "", "", "", shutdownCh)
+	cmd := New(ui, "", "", "", "", nil)
 
 	args := []string{
 		"-server",
-		"-bind", cfg.BindAddr.String(),
+		"-bind", "127.0.0.1",
 		"-data-dir", tmpDir,
-		"-retry-join-wan", cfg.SerfBindAddrWAN.String(),
+		"-retry-join-wan", "127.0.0.1:99",
 		"-retry-max-wan", "1",
 		"-retry-interval-wan", "10ms",
 	}

watch/funcs_test.go

@@ -1,356 +1,379 @@
-package watch
+package watch_test
 
 import (
-	"os"
+	"encoding/json"
+	"errors"
+	"sync"
 	"testing"
 	"time"
 
+	"github.com/hashicorp/consul/agent"
 	consulapi "github.com/hashicorp/consul/api"
+	"github.com/hashicorp/consul/watch"
 )
 
-var consulAddr string
+var errBadContent = errors.New("bad content")
+var errTimeout = errors.New("timeout")
 
-func init() {
-	consulAddr = os.Getenv("CONSUL_ADDR")
+var timeout = 5 * time.Second
+
+func makeInvokeCh() chan error {
+	ch := make(chan error)
+	time.AfterFunc(timeout, func() { ch <- errTimeout })
+	return ch
 }
 
 func TestKeyWatch(t *testing.T) {
-	if consulAddr == "" {
-		t.Skip()
-	}
+	t.Parallel()
+	a := agent.NewTestAgent(t.Name(), ``)
+	defer a.Shutdown()
+
+	invoke := makeInvokeCh()
 	plan := mustParse(t, `{"type":"key", "key":"foo/bar/baz"}`)
-	invoke := 0
 	plan.Handler = func(idx uint64, raw interface{}) {
-		if invoke == 0 {
 		if raw == nil {
-				return
+			return // ignore
 		}
 		v, ok := raw.(*consulapi.KVPair)
-			if !ok || v == nil || string(v.Value) != "test" {
-				t.Fatalf("Bad: %#v", raw)
+		if !ok || v == nil {
+			return // ignore
 		}
-			invoke++
+		if string(v.Value) != "test" {
+			invoke <- errBadContent
+			return
 		}
+		invoke <- nil
 	}
 
+	var wg sync.WaitGroup
+	wg.Add(1)
 	go func() {
-		defer plan.Stop()
-		time.Sleep(20 * time.Millisecond)
+		defer wg.Done()
+		kv := a.Client().KV()
 
-		kv := plan.client.KV()
+		time.Sleep(20 * time.Millisecond)
 		pair := &consulapi.KVPair{
 			Key:   "foo/bar/baz",
 			Value: []byte("test"),
 		}
-		_, err := kv.Put(pair, nil)
-		if err != nil {
-			t.Fatalf("err: %v", err)
-		}
-
-		// Wait for the query to run
-		time.Sleep(20 * time.Millisecond)
-		plan.Stop()
-
-		// Delete the key
-		_, err = kv.Delete("foo/bar/baz", nil)
-		if err != nil {
+		if _, err := kv.Put(pair, nil); err != nil {
 			t.Fatalf("err: %v", err)
 		}
 	}()
 
-	err := plan.Run(consulAddr)
-	if err != nil {
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		if err := plan.Run(a.HTTPAddr()); err != nil {
+			t.Fatalf("err: %v", err)
+		}
+	}()
+
+	if err := <-invoke; err != nil {
 		t.Fatalf("err: %v", err)
 	}
 
-	if invoke == 0 {
-		t.Fatalf("bad: %v", invoke)
-	}
+	plan.Stop()
+	wg.Wait()
 }
 
 func TestKeyWatch_With_PrefixDelete(t *testing.T) {
-	if consulAddr == "" {
-		t.Skip()
-	}
+	t.Parallel()
+	a := agent.NewTestAgent(t.Name(), ``)
+	defer a.Shutdown()
+
+	invoke := makeInvokeCh()
 	plan := mustParse(t, `{"type":"key", "key":"foo/bar/baz"}`)
-	invoke := 0
-	deletedKeyWatchInvoked := 0
 	plan.Handler = func(idx uint64, raw interface{}) {
-		if raw == nil && deletedKeyWatchInvoked == 0 {
-			deletedKeyWatchInvoked++
+		if raw == nil {
+			return // ignore
+		}
+		v, ok := raw.(*consulapi.KVPair)
+		if !ok || v == nil {
+			return // ignore
+		}
+		if string(v.Value) != "test" {
+			invoke <- errBadContent
 			return
 		}
-		if invoke == 0 {
-			v, ok := raw.(*consulapi.KVPair)
-			if !ok || v == nil || string(v.Value) != "test" {
-				t.Fatalf("Bad: %#v", raw)
-			}
-			invoke++
-		}
+		invoke <- nil
 	}
 
+	var wg sync.WaitGroup
+	wg.Add(1)
 	go func() {
-		defer plan.Stop()
-		time.Sleep(20 * time.Millisecond)
+		defer wg.Done()
+		kv := a.Client().KV()
 
-		kv := plan.client.KV()
+		time.Sleep(20 * time.Millisecond)
 		pair := &consulapi.KVPair{
 			Key:   "foo/bar/baz",
 			Value: []byte("test"),
 		}
-		_, err := kv.Put(pair, nil)
-		if err != nil {
+		if _, err := kv.Put(pair, nil); err != nil {
 			t.Fatalf("err: %v", err)
 		}
+	}()
 
-		// Wait for the query to run
-		time.Sleep(20 * time.Millisecond)
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		if err := plan.Run(a.HTTPAddr()); err != nil {
+			t.Fatalf("err: %v", err)
+		}
+	}()
 
-		// Delete the key
-		_, err = kv.DeleteTree("foo/bar", nil)
-		if err != nil {
+	if err := <-invoke; err != nil {
 		t.Fatalf("err: %v", err)
 	}
 
 	plan.Stop()
-	}()
-
-	err := plan.Run(consulAddr)
-	if err != nil {
-		t.Fatalf("err: %v", err)
-	}
-	if invoke != 1 {
-		t.Fatalf("expected watch plan to be invoked once but got %v", invoke)
-	}
-
-	if deletedKeyWatchInvoked != 1 {
-		t.Fatalf("expected watch plan to be invoked once on delete but got %v", deletedKeyWatchInvoked)
-	}
+	wg.Wait()
 }
 
 func TestKeyPrefixWatch(t *testing.T) {
-	if consulAddr == "" {
-		t.Skip()
-	}
+	t.Parallel()
+	a := agent.NewTestAgent(t.Name(), ``)
+	defer a.Shutdown()
+
+	invoke := makeInvokeCh()
 	plan := mustParse(t, `{"type":"keyprefix", "prefix":"foo/"}`)
-	invoke := 0
 	plan.Handler = func(idx uint64, raw interface{}) {
-		if invoke == 0 {
 		if raw == nil {
-				return
+			return // ignore
 		}
 		v, ok := raw.(consulapi.KVPairs)
-			if ok && v == nil {
+		if !ok || len(v) == 0 {
 			return
 		}
-			if !ok || v == nil || string(v[0].Key) != "foo/bar" {
-				t.Fatalf("Bad: %#v", raw)
-			}
-			invoke++
+		if string(v[0].Key) != "foo/bar" {
+			invoke <- errBadContent
+			return
 		}
+		invoke <- nil
 	}
 
+	var wg sync.WaitGroup
+	wg.Add(1)
 	go func() {
-		defer plan.Stop()
-		time.Sleep(20 * time.Millisecond)
+		defer wg.Done()
+		kv := a.Client().KV()
 
-		kv := plan.client.KV()
+		time.Sleep(20 * time.Millisecond)
 		pair := &consulapi.KVPair{
 			Key: "foo/bar",
 		}
-		_, err := kv.Put(pair, nil)
-		if err != nil {
-			t.Fatalf("err: %v", err)
-		}
-
-		// Wait for the query to run
-		time.Sleep(20 * time.Millisecond)
-		plan.Stop()
-
-		// Delete the key
-		_, err = kv.Delete("foo/bar", nil)
-		if err != nil {
+		if _, err := kv.Put(pair, nil); err != nil {
 			t.Fatalf("err: %v", err)
 		}
 	}()
 
-	err := plan.Run(consulAddr)
-	if err != nil {
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		if err := plan.Run(a.HTTPAddr()); err != nil {
+			t.Fatalf("err: %v", err)
+		}
+	}()
+
+	if err := <-invoke; err != nil {
 		t.Fatalf("err: %v", err)
 	}
 
-	if invoke == 0 {
-		t.Fatalf("bad: %v", invoke)
-	}
+	plan.Stop()
+	wg.Wait()
 }
 
 func TestServicesWatch(t *testing.T) {
-	if consulAddr == "" {
-		t.Skip()
-	}
+	t.Parallel()
+	a := agent.NewTestAgent(t.Name(), ``)
+	defer a.Shutdown()
+
+	invoke := makeInvokeCh()
 	plan := mustParse(t, `{"type":"services"}`)
-	invoke := 0
 	plan.Handler = func(idx uint64, raw interface{}) {
-		if invoke == 0 {
 		if raw == nil {
-				return
+			return // ignore
 		}
 		v, ok := raw.(map[string][]string)
-			if !ok || v["consul"] == nil {
-				t.Fatalf("Bad: %#v", raw)
+		if !ok || len(v) == 0 {
+			return // ignore
 		}
-			invoke++
+		if v["consul"] == nil {
+			invoke <- errBadContent
+			return
 		}
+		invoke <- nil
 	}
 
+	var wg sync.WaitGroup
+	wg.Add(1)
 	go func() {
-		time.Sleep(20 * time.Millisecond)
-		plan.Stop()
+		defer wg.Done()
+		agent := a.Client().Agent()
 
-		agent := plan.client.Agent()
+		time.Sleep(20 * time.Millisecond)
 		reg := &consulapi.AgentServiceRegistration{
 			ID:   "foo",
 			Name: "foo",
 		}
-		agent.ServiceRegister(reg)
-		time.Sleep(20 * time.Millisecond)
-		agent.ServiceDeregister("foo")
+		if err := agent.ServiceRegister(reg); err != nil {
+			t.Fatalf("err: %v", err)
+		}
 	}()
 
-	err := plan.Run(consulAddr)
-	if err != nil {
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		if err := plan.Run(a.HTTPAddr()); err != nil {
+			t.Fatalf("err: %v", err)
+		}
+	}()
+
+	if err := <-invoke; err != nil {
 		t.Fatalf("err: %v", err)
 	}
 
-	if invoke == 0 {
-		t.Fatalf("bad: %v", invoke)
-	}
+	plan.Stop()
+	wg.Wait()
 }
 
 func TestNodesWatch(t *testing.T) {
-	if consulAddr == "" {
-		t.Skip()
-	}
+	t.Parallel()
+	a := agent.NewTestAgent(t.Name(), ``)
+	defer a.Shutdown()
+
+	invoke := makeInvokeCh()
 	plan := mustParse(t, `{"type":"nodes"}`)
-	invoke := 0
 	plan.Handler = func(idx uint64, raw interface{}) {
-		if invoke == 0 {
 		if raw == nil {
-				return
+			return // ignore
 		}
 		v, ok := raw.([]*consulapi.Node)
 		if !ok || len(v) == 0 {
-				t.Fatalf("Bad: %#v", raw)
-			}
-			invoke++
+			return // ignore
 		}
+		invoke <- nil
 	}
 
+	var wg sync.WaitGroup
+	wg.Add(1)
 	go func() {
-		time.Sleep(20 * time.Millisecond)
-		plan.Stop()
+		defer wg.Done()
+		catalog := a.Client().Catalog()
 
-		catalog := plan.client.Catalog()
+		time.Sleep(20 * time.Millisecond)
 		reg := &consulapi.CatalogRegistration{
 			Node:       "foobar",
 			Address:    "1.1.1.1",
 			Datacenter: "dc1",
 		}
-		catalog.Register(reg, nil)
-		time.Sleep(20 * time.Millisecond)
-		dereg := &consulapi.CatalogDeregistration{
-			Node:       "foobar",
-			Address:    "1.1.1.1",
-			Datacenter: "dc1",
+		if _, err := catalog.Register(reg, nil); err != nil {
+			t.Fatalf("err: %v", err)
 		}
-		catalog.Deregister(dereg, nil)
 	}()
 
-	err := plan.Run(consulAddr)
-	if err != nil {
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		if err := plan.Run(a.HTTPAddr()); err != nil {
+			t.Fatalf("err: %v", err)
+		}
+	}()
+
+	if err := <-invoke; err != nil {
 		t.Fatalf("err: %v", err)
 	}
 
-	if invoke == 0 {
-		t.Fatalf("bad: %v", invoke)
-	}
+	plan.Stop()
+	wg.Wait()
 }
 
 func TestServiceWatch(t *testing.T) {
-	if consulAddr == "" {
-		t.Skip()
-	}
+	t.Parallel()
+	a := agent.NewTestAgent(t.Name(), ``)
+	defer a.Shutdown()
+
+	invoke := makeInvokeCh()
 	plan := mustParse(t, `{"type":"service", "service":"foo", "tag":"bar", "passingonly":true}`)
-	invoke := 0
 	plan.Handler = func(idx uint64, raw interface{}) {
-		if invoke == 0 {
 		if raw == nil {
-				return
+			return // ignore
 		}
 		v, ok := raw.([]*consulapi.ServiceEntry)
-			if ok && len(v) == 0 {
+		if !ok || len(v) == 0 {
+			return // ignore
+		}
+		if v[0].Service.ID != "foo" {
+			invoke <- errBadContent
 			return
 		}
-			if !ok || v[0].Service.ID != "foo" {
-				t.Fatalf("Bad: %#v", raw)
-			}
-			invoke++
-		}
+		invoke <- nil
 	}
 
+	var wg sync.WaitGroup
+
+	wg.Add(1)
 	go func() {
-		time.Sleep(20 * time.Millisecond)
+		defer wg.Done()
+		agent := a.Client().Agent()
 
-		agent := plan.client.Agent()
+		time.Sleep(20 * time.Millisecond)
 		reg := &consulapi.AgentServiceRegistration{
 			ID:   "foo",
 			Name: "foo",
 			Tags: []string{"bar"},
 		}
-		agent.ServiceRegister(reg)
-
-		time.Sleep(20 * time.Millisecond)
-		plan.Stop()
-
-		agent.ServiceDeregister("foo")
+		if err := agent.ServiceRegister(reg); err != nil {
+			t.Fatalf("err: %v", err)
+		}
 	}()
 
-	err := plan.Run(consulAddr)
-	if err != nil {
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		if err := plan.Run(a.HTTPAddr()); err != nil {
+			t.Fatalf("err: %v", err)
+		}
+	}()
+
+	if err := <-invoke; err != nil {
 		t.Fatalf("err: %v", err)
 	}
 
-	if invoke == 0 {
-		t.Fatalf("bad: %v", invoke)
-	}
+	plan.Stop()
+	wg.Wait()
 }
 
 func TestChecksWatch_State(t *testing.T) {
-	if consulAddr == "" {
-		t.Skip()
-	}
+	t.Parallel()
+	a := agent.NewTestAgent(t.Name(), ``)
+	defer a.Shutdown()
+
+	invoke := makeInvokeCh()
 	plan := mustParse(t, `{"type":"checks", "state":"warning"}`)
-	invoke := 0
 	plan.Handler = func(idx uint64, raw interface{}) {
-		if invoke == 0 {
 		if raw == nil {
-				return
+			return // ignore
 		}
 		v, ok := raw.([]*consulapi.HealthCheck)
-			if len(v) == 0 {
+		if !ok || len(v) == 0 {
+			return // ignore
+		}
+		if v[0].CheckID != "foobar" || v[0].Status != "warning" {
+			invoke <- errBadContent
 			return
 		}
-			if !ok || v[0].CheckID != "foobar" {
-				t.Fatalf("Bad: %#v", raw)
-			}
-			invoke++
-		}
+		invoke <- nil
 	}
 
+	var wg sync.WaitGroup
+	wg.Add(1)
 	go func() {
-		time.Sleep(20 * time.Millisecond)
+		defer wg.Done()
+		catalog := a.Client().Catalog()
 
-		catalog := plan.client.Catalog()
+		time.Sleep(20 * time.Millisecond)
 		reg := &consulapi.CatalogRegistration{
 			Node:       "foobar",
 			Address:    "1.1.1.1",
@@ -362,55 +385,56 @@ func TestChecksWatch_State(t *testing.T) {
 				Status:  consulapi.HealthWarning,
 			},
 		}
-		catalog.Register(reg, nil)
-
-		time.Sleep(20 * time.Millisecond)
-		plan.Stop()
-
-		dereg := &consulapi.CatalogDeregistration{
-			Node:       "foobar",
-			Address:    "1.1.1.1",
-			Datacenter: "dc1",
+		if _, err := catalog.Register(reg, nil); err != nil {
+			t.Fatalf("err: %v", err)
 		}
-		catalog.Deregister(dereg, nil)
 	}()
 
-	err := plan.Run(consulAddr)
-	if err != nil {
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		if err := plan.Run(a.HTTPAddr()); err != nil {
+			t.Fatalf("err: %v", err)
+		}
+	}()
+
+	if err := <-invoke; err != nil {
 		t.Fatalf("err: %v", err)
 	}
 
-	if invoke == 0 {
-		t.Fatalf("bad: %v", invoke)
-	}
+	plan.Stop()
+	wg.Wait()
 }
 
 func TestChecksWatch_Service(t *testing.T) {
-	if consulAddr == "" {
-		t.Skip()
-	}
+	t.Parallel()
+	a := agent.NewTestAgent(t.Name(), ``)
+	defer a.Shutdown()
+
+	invoke := makeInvokeCh()
 	plan := mustParse(t, `{"type":"checks", "service":"foobar"}`)
-	invoke := 0
 	plan.Handler = func(idx uint64, raw interface{}) {
-		if invoke == 0 {
 		if raw == nil {
-				return
+			return // ignore
 		}
 		v, ok := raw.([]*consulapi.HealthCheck)
-			if len(v) == 0 {
+		if !ok || len(v) == 0 {
+			return // ignore
+		}
+		if v[0].CheckID != "foobar" {
+			invoke <- errBadContent
 			return
 		}
-			if !ok || v[0].CheckID != "foobar" {
-				t.Fatalf("Bad: %#v", raw)
-			}
-			invoke++
-		}
+		invoke <- nil
 	}
 
+	var wg sync.WaitGroup
+	wg.Add(1)
 	go func() {
-		time.Sleep(20 * time.Millisecond)
+		defer wg.Done()
+		catalog := a.Client().Catalog()
 
-		catalog := plan.client.Catalog()
+		time.Sleep(20 * time.Millisecond)
 		reg := &consulapi.CatalogRegistration{
 			Node:       "foobar",
 			Address:    "1.1.1.1",
@@ -427,68 +451,86 @@ func TestChecksWatch_Service(t *testing.T) {
 				ServiceID: "foobar",
 			},
 		}
-		_, err := catalog.Register(reg, nil)
-		if err != nil {
+		if _, err := catalog.Register(reg, nil); err != nil {
 			t.Fatalf("err: %v", err)
 		}
-
-		time.Sleep(20 * time.Millisecond)
-		plan.Stop()
-
-		dereg := &consulapi.CatalogDeregistration{
-			Node:       "foobar",
-			Address:    "1.1.1.1",
-			Datacenter: "dc1",
-		}
-		catalog.Deregister(dereg, nil)
 	}()
 
-	err := plan.Run(consulAddr)
-	if err != nil {
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		if err := plan.Run(a.HTTPAddr()); err != nil {
+			t.Fatalf("err: %v", err)
+		}
+	}()
+
+	if err := <-invoke; err != nil {
 		t.Fatalf("err: %v", err)
 	}
 
-	if invoke == 0 {
-		t.Fatalf("bad: %v", invoke)
-	}
+	plan.Stop()
+	wg.Wait()
 }
 
 func TestEventWatch(t *testing.T) {
-	if consulAddr == "" {
-		t.Skip()
-	}
+	t.Parallel()
+	a := agent.NewTestAgent(t.Name(), ``)
+	defer a.Shutdown()
+
+	invoke := makeInvokeCh()
 	plan := mustParse(t, `{"type":"event", "name": "foo"}`)
-	invoke := 0
 	plan.Handler = func(idx uint64, raw interface{}) {
-		if invoke == 0 {
 		if raw == nil {
 			return
 		}
 		v, ok := raw.([]*consulapi.UserEvent)
-			if !ok || len(v) == 0 || string(v[len(v)-1].Name) != "foo" {
-				t.Fatalf("Bad: %#v", raw)
+		if !ok || len(v) == 0 {
+			return // ignore
 		}
-			invoke++
+		if string(v[len(v)-1].Name) != "foo" {
+			invoke <- errBadContent
+			return
 		}
+		invoke <- nil
 	}
 
+	var wg sync.WaitGroup
+	wg.Add(1)
 	go func() {
-		defer plan.Stop()
-		time.Sleep(20 * time.Millisecond)
+		defer wg.Done()
+		event := a.Client().Event()
 
-		event := plan.client.Event()
+		time.Sleep(20 * time.Millisecond)
 		params := &consulapi.UserEvent{Name: "foo"}
 		if _, _, err := event.Fire(params, nil); err != nil {
 			t.Fatalf("err: %v", err)
 		}
 	}()
 
-	err := plan.Run(consulAddr)
-	if err != nil {
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		if err := plan.Run(a.HTTPAddr()); err != nil {
+			t.Fatalf("err: %v", err)
+		}
+	}()
+
+	if err := <-invoke; err != nil {
 		t.Fatalf("err: %v", err)
 	}
 
-	if invoke == 0 {
-		t.Fatalf("bad: %v", invoke)
+	plan.Stop()
+	wg.Wait()
 }
+
+func mustParse(t *testing.T, q string) *watch.Plan {
+	var params map[string]interface{}
+	if err := json.Unmarshal([]byte(q), &params); err != nil {
+		t.Fatal(err)
+	}
+	plan, err := watch.Parse(params)
+	if err != nil {
+		t.Fatalf("err: %v", err)
+	}
+	return plan
 }

watch/plan_test.go

@@ -27,6 +27,7 @@ func mustParse(t *testing.T, q string) *Plan {
 }
 
 func TestRun_Stop(t *testing.T) {
+	t.Parallel()
 	plan := mustParse(t, `{"type":"noop"}`)
 
 	var expect uint64 = 1

watch/watch_test.go

@@ -7,6 +7,7 @@ import (
 )
 
 func TestParseBasic(t *testing.T) {
+	t.Parallel()
 	params := makeParams(t, `{"type":"key", "datacenter":"dc2", "token":"12345", "key":"foo"}`)
 	p, err := Parse(params)
 	if err != nil {
@@ -24,6 +25,7 @@ func TestParseBasic(t *testing.T) {
 }
 
 func TestParse_exempt(t *testing.T) {
+	t.Parallel()
 	params := makeParams(t, `{"type":"key", "key":"foo", "handler": "foobar"}`)
 	p, err := ParseExempt(params, []string{"handler"})
 	if err != nil {