Fix unstable tests in agent, api, and command/watch

agent/catalog_endpoint_test.go

@@ -81,6 +81,7 @@ func TestCatalogNodes(t *testing.T) {
 	t.Parallel()
 	a := NewTestAgent(t.Name(), "")
 	defer a.Shutdown()
+	testrpc.WaitForTestAgent(t, a.RPC, "dc1")
 
 	// Register node
 	args := &structs.RegisterRequest{
@@ -93,7 +94,6 @@ func TestCatalogNodes(t *testing.T) {
 	if err := a.RPC("Catalog.Register", args, &out); err != nil {
 		t.Fatalf("err: %v", err)
 	}
-	testrpc.WaitForLeader(t, a.RPC, "dc1")
 
 	req, _ := http.NewRequest("GET", "/v1/catalog/nodes?dc=dc1", nil)
 	resp := httptest.NewRecorder()
@@ -115,6 +115,7 @@ func TestCatalogNodes_MetaFilter(t *testing.T) {
 	t.Parallel()
 	a := NewTestAgent(t.Name(), "")
 	defer a.Shutdown()
+	testrpc.WaitForTestAgent(t, a.RPC, "dc1")
 
 	// Register a node with a meta field
 	args := &structs.RegisterRequest{
@@ -159,6 +160,7 @@ func TestCatalogNodes_WanTranslation(t *testing.T) {
 		acl_datacenter = ""
 	`)
 	defer a1.Shutdown()
+	testrpc.WaitForTestAgent(t, a1.RPC, "dc1")
 
 	a2 := NewTestAgent(t.Name(), `
 		datacenter = "dc2"
@@ -166,6 +168,7 @@ func TestCatalogNodes_WanTranslation(t *testing.T) {
 		acl_datacenter = ""
 	`)
 	defer a2.Shutdown()
+	testrpc.WaitForTestAgent(t, a2.RPC, "dc2")
 
 	// Wait for the WAN join.
 	addr := fmt.Sprintf("127.0.0.1:%d", a1.Config.SerfPortWAN)
@@ -251,6 +254,7 @@ func TestCatalogNodes_Blocking(t *testing.T) {
 	t.Parallel()
 	a := NewTestAgent(t.Name(), "")
 	defer a.Shutdown()
+	testrpc.WaitForTestAgent(t, a.RPC, "dc1")
 
 	// Register node
 	args := &structs.DCSpecificRequest{
@@ -320,6 +324,7 @@ func TestCatalogNodes_DistanceSort(t *testing.T) {
 	t.Parallel()
 	a := NewTestAgent(t.Name(), "")
 	defer a.Shutdown()
+	testrpc.WaitForTestAgent(t, a.RPC, "dc1")
 
 	// Register nodes.
 	args := &structs.RegisterRequest{
@@ -331,7 +336,6 @@ func TestCatalogNodes_DistanceSort(t *testing.T) {
 	if err := a.RPC("Catalog.Register", args, &out); err != nil {
 		t.Fatalf("err: %v", err)
 	}
-	testrpc.WaitForLeader(t, a.RPC, "dc1")
 
 	args = &structs.RegisterRequest{
 		Datacenter: "dc1",
@@ -815,6 +819,7 @@ func TestCatalogNodeServices(t *testing.T) {
 	t.Parallel()
 	a := NewTestAgent(t.Name(), "")
 	defer a.Shutdown()
+	testrpc.WaitForTestAgent(t, a.RPC, "dc1")
 
 	// Register node
 	args := &structs.RegisterRequest{
@@ -854,6 +859,7 @@ func TestCatalogNodeServices_ConnectProxy(t *testing.T) {
 	assert := assert.New(t)
 	a := NewTestAgent(t.Name(), "")
 	defer a.Shutdown()
+	testrpc.WaitForTestAgent(t, a.RPC, "dc1")
 
 	// Register
 	args := structs.TestRegisterRequestProxy(t)
@@ -881,6 +887,7 @@ func TestCatalogNodeServices_WanTranslation(t *testing.T) {
 		acl_datacenter = ""
 	`)
 	defer a1.Shutdown()
+	testrpc.WaitForTestAgent(t, a1.RPC, "dc1")
 
 	a2 := NewTestAgent(t.Name(), `
 		datacenter = "dc2"
@@ -888,6 +895,7 @@ func TestCatalogNodeServices_WanTranslation(t *testing.T) {
 		acl_datacenter = ""
 	`)
 	defer a2.Shutdown()
+	testrpc.WaitForTestAgent(t, a2.RPC, "dc2")
 
 	// Wait for the WAN join.
 	addr := fmt.Sprintf("127.0.0.1:%d", a1.Config.SerfPortWAN)

agent/connect_ca_endpoint_test.go

@@ -7,6 +7,8 @@ import (
 	"testing"
 	"time"
 
+	"github.com/hashicorp/consul/testrpc"
+
 	"github.com/stretchr/testify/require"
 
 	"github.com/hashicorp/consul/agent/connect"
@@ -35,6 +37,7 @@ func TestConnectCARoots_list(t *testing.T) {
 	assert := assert.New(t)
 	a := NewTestAgent(t.Name(), "")
 	defer a.Shutdown()
+	testrpc.WaitForTestAgent(t, a.RPC, "dc1")
 
 	// Set some CAs. Note that NewTestAgent already bootstraps one CA so this just
 	// adds a second and makes it active.

agent/consul/catalog_endpoint_test.go

@@ -1472,6 +1472,7 @@ func TestCatalog_ListServices_Stale(t *testing.T) {
 	defer os.RemoveAll(dir1)
 	defer s1.Shutdown()
 
+	testrpc.WaitForTestAgent(t, s1.RPC, "dc1")
 	dir2, s2 := testServerWithConfig(t, func(c *Config) {
 		c.ACLDatacenter = "dc1" // Enable ACLs!
 		c.Bootstrap = false     // Disable bootstrap

agent/consul/coordinate_endpoint_test.go

@@ -488,7 +488,7 @@ func TestCoordinate_Node(t *testing.T) {
 
 	codec := rpcClient(t, s1)
 	defer codec.Close()
-	testrpc.WaitForLeader(t, s1.RPC, "dc1")
+	testrpc.WaitForTestAgent(t, s1.RPC, "dc1")
 
 	// Register some nodes.
 	nodes := []string{"foo", "bar"}

agent/consul/stats_fetcher_test.go

@@ -28,6 +28,8 @@ func TestStatsFetcher(t *testing.T) {
 	joinLAN(t, s2, s1)
 	joinLAN(t, s3, s1)
 	testrpc.WaitForLeader(t, s1.RPC, "dc1")
+	testrpc.WaitForLeader(t, s2.RPC, "dc1")
+	testrpc.WaitForLeader(t, s3.RPC, "dc1")
 
 	members := s1.serfLAN.Members()
 	if len(members) != 3 {

agent/dns_test.go

@@ -6171,7 +6171,7 @@ func TestDNS_Compression_Recurse(t *testing.T) {
 		recursors = ["`+recursor.Addr+`"]
 	`)
 	defer a.Shutdown()
-	testrpc.WaitForLeader(t, a.RPC, "dc1")
+	testrpc.WaitForTestAgent(t, a.RPC, "dc1")
 
 	m := new(dns.Msg)
 	m.SetQuestion("apple.com.", dns.TypeANY)

agent/health_endpoint_test.go

@@ -184,6 +184,7 @@ func TestHealthNodeChecks(t *testing.T) {
 	t.Parallel()
 	a := NewTestAgent(t.Name(), "")
 	defer a.Shutdown()
+	testrpc.WaitForTestAgent(t, a.RPC, "dc1")
 
 	req, _ := http.NewRequest("GET", "/v1/health/node/nope?dc=dc1", nil)
 	resp := httptest.NewRecorder()
@@ -270,7 +271,7 @@ func TestHealthServiceChecks_NodeMetaFilter(t *testing.T) {
 	t.Parallel()
 	a := NewTestAgent(t.Name(), "")
 	defer a.Shutdown()
-	testrpc.WaitForLeader(t, a.RPC, "dc1")
+	testrpc.WaitForTestAgent(t, a.RPC, "dc1")
 
 	req, _ := http.NewRequest("GET", "/v1/health/checks/consul?dc=dc1&node-meta=somekey:somevalue", nil)
 	resp := httptest.NewRecorder()
@@ -403,7 +404,7 @@ func TestHealthServiceNodes(t *testing.T) {
 	t.Parallel()
 	a := NewTestAgent(t.Name(), "")
 	defer a.Shutdown()
-	testrpc.WaitForLeader(t, a.RPC, "dc1")
+	testrpc.WaitForTestAgent(t, a.RPC, "dc1")
 
 	req, _ := http.NewRequest("GET", "/v1/health/service/consul?dc=dc1", nil)
 	resp := httptest.NewRecorder()

agent/operator_endpoint_test.go

@@ -8,6 +8,8 @@ import (
 	"strings"
 	"testing"
 
+	"github.com/hashicorp/consul/testrpc"
+
 	"github.com/hashicorp/consul/agent/consul/autopilot"
 	"github.com/hashicorp/consul/agent/structs"
 	"github.com/hashicorp/consul/api"
@@ -291,6 +293,7 @@ func TestOperator_AutopilotGetConfiguration(t *testing.T) {
 	t.Parallel()
 	a := NewTestAgent(t.Name(), "")
 	defer a.Shutdown()
+	testrpc.WaitForTestAgent(t, a.RPC, "dc1")
 
 	body := bytes.NewBuffer(nil)
 	req, _ := http.NewRequest("GET", "/v1/operator/autopilot/configuration", body)

agent/prepared_query_endpoint_test.go

@@ -9,6 +9,8 @@ import (
 	"reflect"
 	"testing"
 
+	"github.com/hashicorp/consul/testrpc"
+
 	"github.com/hashicorp/consul/agent/structs"
 	"github.com/hashicorp/consul/types"
 	"github.com/stretchr/testify/require"
@@ -928,6 +930,7 @@ func TestPreparedQuery_Integration(t *testing.T) {
 	t.Parallel()
 	a := NewTestAgent(t.Name(), "")
 	defer a.Shutdown()
+	testrpc.WaitForTestAgent(t, a.RPC, "dc1")
 
 	// Register a node and a service.
 	{

agent/remote_exec_test.go

@@ -138,6 +138,8 @@ func testRemoteExecGetSpec(t *testing.T, hcl string, token string, shouldSucceed
 	defer a.Shutdown()
 	if dc != "" {
 		testrpc.WaitForLeader(t, a.RPC, dc)
+	} else {
+		testrpc.WaitForTestAgent(t, a.RPC, "dc1")
 	}
 	event := &remoteExecEvent{
 		Prefix:  "_rexec",
@@ -210,7 +212,7 @@ func testRemoteExecWrites(t *testing.T, hcl string, token string, shouldSucceed
 		testrpc.WaitForLeader(t, a.RPC, dc)
 	} else {
 		// For slow machines, ensure we wait a bit
-		time.Sleep(1 * time.Millisecond)
+		testrpc.WaitForLeader(t, a.RPC, "dc1")
 	}
 	event := &remoteExecEvent{
 		Prefix:  "_rexec",
@@ -268,6 +270,7 @@ func testRemoteExecWrites(t *testing.T, hcl string, token string, shouldSucceed
 func testHandleRemoteExec(t *testing.T, command string, expectedSubstring string, expectedReturnCode string) {
 	a := NewTestAgent(t.Name(), "")
 	defer a.Shutdown()
+	testrpc.WaitForTestAgent(t, a.RPC, "dc1")
 	retry.Run(t, func(r *retry.R) {
 		event := &remoteExecEvent{
 			Prefix:  "_rexec",

api/api_test.go

@@ -24,6 +24,12 @@ func makeClient(t *testing.T) (*Client, *testutil.TestServer) {
 	return makeClientWithConfig(t, nil, nil)
 }
 
+func makeClientWithoutConnect(t *testing.T) (*Client, *testutil.TestServer) {
+	return makeClientWithConfig(t, nil, func(serverConfig *testutil.TestServerConfig) {
+		serverConfig.Connect = nil
+	})
+}
+
 func makeACLClient(t *testing.T) (*Client, *testutil.TestServer) {
 	return makeClientWithConfig(t, func(clientConfig *Config) {
 		clientConfig.Token = "root"

api/lock_test.go

@@ -15,7 +15,7 @@ import (
 
 func TestAPI_LockLockUnlock(t *testing.T) {
 	t.Parallel()
-	c, s := makeClient(t)
+	c, s := makeClientWithoutConnect(t)
 	defer s.Stop()
 
 	lock, err := c.LockKey("test/lock")
@@ -73,7 +73,7 @@ func TestAPI_LockLockUnlock(t *testing.T) {
 
 func TestAPI_LockForceInvalidate(t *testing.T) {
 	t.Parallel()
-	c, s := makeClient(t)
+	c, s := makeClientWithoutConnect(t)
 	defer s.Stop()
 
 	retry.Run(t, func(r *retry.R) {
@@ -110,7 +110,7 @@ func TestAPI_LockForceInvalidate(t *testing.T) {
 
 func TestAPI_LockDeleteKey(t *testing.T) {
 	t.Parallel()
-	c, s := makeClient(t)
+	c, s := makeClientWithoutConnect(t)
 	defer s.Stop()
 
 	// This uncovered some issues around special-case handling of low index
@@ -152,7 +152,7 @@ func TestAPI_LockDeleteKey(t *testing.T) {
 
 func TestAPI_LockContend(t *testing.T) {
 	t.Parallel()
-	c, s := makeClient(t)
+	c, s := makeClientWithoutConnect(t)
 	defer s.Stop()
 
 	wg := &sync.WaitGroup{}
@@ -205,7 +205,7 @@ func TestAPI_LockContend(t *testing.T) {
 
 func TestAPI_LockDestroy(t *testing.T) {
 	t.Parallel()
-	c, s := makeClient(t)
+	c, s := makeClientWithoutConnect(t)
 	defer s.Stop()
 
 	lock, err := c.LockKey("test/lock")
@@ -274,7 +274,7 @@ func TestAPI_LockDestroy(t *testing.T) {
 
 func TestAPI_LockConflict(t *testing.T) {
 	t.Parallel()
-	c, s := makeClient(t)
+	c, s := makeClientWithoutConnect(t)
 	defer s.Stop()
 
 	sema, err := c.SemaphorePrefix("test/lock/", 2)
@@ -312,7 +312,7 @@ func TestAPI_LockConflict(t *testing.T) {
 
 func TestAPI_LockReclaimLock(t *testing.T) {
 	t.Parallel()
-	c, s := makeClient(t)
+	c, s := makeClientWithoutConnect(t)
 	defer s.Stop()
 
 	session, _, err := c.Session().Create(&SessionEntry{}, nil)
@@ -380,7 +380,7 @@ func TestAPI_LockReclaimLock(t *testing.T) {
 
 func TestAPI_LockMonitorRetry(t *testing.T) {
 	t.Parallel()
-	raw, s := makeClient(t)
+	raw, s := makeClientWithoutConnect(t)
 	defer s.Stop()
 
 	// Set up a server that always responds with 500 errors.
@@ -495,7 +495,7 @@ func TestAPI_LockMonitorRetry(t *testing.T) {
 
 func TestAPI_LockOneShot(t *testing.T) {
 	t.Parallel()
-	c, s := makeClient(t)
+	c, s := makeClientWithoutConnect(t)
 	defer s.Stop()
 
 	// Set up a lock as a one-shot.

command/watch/watch_test.go

@@ -4,6 +4,8 @@ import (
 	"strings"
 	"testing"
 
+	"github.com/hashicorp/consul/testrpc"
+
 	"github.com/hashicorp/consul/agent"
 	"github.com/mitchellh/cli"
 )
@@ -19,6 +21,7 @@ func TestWatchCommand(t *testing.T) {
 	t.Parallel()
 	a := agent.NewTestAgent(t.Name(), ``)
 	defer a.Shutdown()
+	testrpc.WaitForTestAgent(t, a.RPC, "dc1")
 
 	ui := cli.NewMockUi()
 	c := New(ui, nil)
@@ -38,6 +41,7 @@ func TestWatchCommandNoConnect(t *testing.T) {
 	t.Parallel()
 	a := agent.NewTestAgent(t.Name(), ``)
 	defer a.Shutdown()
+	testrpc.WaitForTestAgent(t, a.RPC, "dc1")
 
 	ui := cli.NewMockUi()
 	c := New(ui, nil)