func (s) TestClusterWatchExpiryTimer(t *testing.T) {
	apiClientCh, cleanup := overrideNewAPIClient()
	defer cleanup()
	client, err := newWithConfig(clientOpts(testXDSServer, true))
	if err != nil {
		t.Fatalf("failed to create client: %v", err)
	}
	defer client.Close()
	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
	defer cancel()
	c, err := apiClientCh.Receive(ctx)
	if err != nil {
		t.Fatalf("timeout when waiting for API client to be created: %v", err)
	}
	apiClient := c.(*testAPIClient)
	clusterUpdateCh := testutils.NewChannel()
	client.WatchCluster(testCDSName, func(u ClusterUpdate, err error) {
		clusterUpdateCh.Send(ClusterUpdateErrTuple{Update: u, Err: err})
	})
	if _, err := apiClient.addWatches[ClusterResource].Receive(ctx); err != nil {
		t.Fatalf("want new watch to start, got error %v", err)
	}
	u, err := clusterUpdateCh.Receive(ctx)
	if err != nil {
		t.Fatalf("timeout when waiting for cluster update: %v", err)
	}
	gotUpdate := u.(ClusterUpdateErrTuple)
	if gotUpdate.Err == nil || !cmp.Equal(gotUpdate.Update, ClusterUpdate{}) {
		t.Fatalf("unexpected clusterUpdate: (%v, %v), want: (ClusterUpdate{}, nil)", gotUpdate.Update, gotUpdate.Err)
	}
}
TestClusterWatchExpiryTimer tests the case where the client does not receive
an CDS response for the request that it sends out. We want the watch callback
to be invoked with an error once the watchExpiryTimer fires.
package xdsclient
import (
	"context"
	"fmt"
	"testing"
	"github.com/google/go-cmp/cmp"
	"google.golang.org/protobuf/types/known/anypb"
	"google.golang.org/grpc/internal/testutils"
)
func (s) TestClusterWatch(t *testing.T) {
	apiClientCh, cleanup := overrideNewAPIClient()
	defer cleanup()
	client, err := newWithConfig(clientOpts(testXDSServer, false))
	if err != nil {
		t.Fatalf("failed to create client: %v", err)
	}
	defer client.Close()
	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
	defer cancel()
	c, err := apiClientCh.Receive(ctx)
	if err != nil {
		t.Fatalf("timeout when waiting for API client to be created: %v", err)
	}
	apiClient := c.(*testAPIClient)
	clusterUpdateCh := testutils.NewChannel()
	cancelWatch := client.WatchCluster(testCDSName, func(update ClusterUpdate, err error) {
		clusterUpdateCh.Send(ClusterUpdateErrTuple{Update: update, Err: err})
	})
	if _, err := apiClient.addWatches[ClusterResource].Receive(ctx); err != nil {
		t.Fatalf("want new watch to start, got error %v", err)
	}
	wantUpdate := ClusterUpdate{ClusterName: testEDSName}
	client.NewClusters(map[string]ClusterUpdateErrTuple{testCDSName: {Update: wantUpdate}}, UpdateMetadata{})
	if err := verifyClusterUpdate(ctx, clusterUpdateCh, wantUpdate, nil); err != nil {
		t.Fatal(err)
	}
	newUpdate := wantUpdate
	newUpdate.Raw = &anypb.Any{}
	client.NewClusters(map[string]ClusterUpdateErrTuple{
		testCDSName:  {Update: newUpdate},
		"randomName": {},
	}, UpdateMetadata{})
	if err := verifyClusterUpdate(ctx, clusterUpdateCh, newUpdate, nil); err != nil {
		t.Fatal(err)
	}
	cancelWatch()
	client.NewClusters(map[string]ClusterUpdateErrTuple{testCDSName: {Update: wantUpdate}}, UpdateMetadata{})
	sCtx, sCancel := context.WithTimeout(ctx, defaultTestShortTimeout)
	defer sCancel()
	if u, err := clusterUpdateCh.Receive(sCtx); err != context.DeadlineExceeded {
		t.Errorf("unexpected clusterUpdate: %v, %v, want channel recv timeout", u, err)
	}
}
func (s) TestClusterTwoWatchSameResourceName(t *testing.T) {
	apiClientCh, cleanup := overrideNewAPIClient()
	defer cleanup()
	client, err := newWithConfig(clientOpts(testXDSServer, false))
	if err != nil {
		t.Fatalf("failed to create client: %v", err)
	}
	defer client.Close()
	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
	defer cancel()
	c, err := apiClientCh.Receive(ctx)
	if err != nil {
		t.Fatalf("timeout when waiting for API client to be created: %v", err)
	}
	apiClient := c.(*testAPIClient)
	var clusterUpdateChs []*testutils.Channel
	var cancelLastWatch func()
	const count = 2
	for i := 0; i < count; i++ {
		clusterUpdateCh := testutils.NewChannel()
		clusterUpdateChs = append(clusterUpdateChs, clusterUpdateCh)
		cancelLastWatch = client.WatchCluster(testCDSName, func(update ClusterUpdate, err error) {
			clusterUpdateCh.Send(ClusterUpdateErrTuple{Update: update, Err: err})
		})
		if i == 0 {
			if _, err := apiClient.addWatches[ClusterResource].Receive(ctx); err != nil {
				t.Fatalf("want new watch to start, got error %v", err)
			}
		}
	}
	wantUpdate := ClusterUpdate{ClusterName: testEDSName}
	client.NewClusters(map[string]ClusterUpdateErrTuple{testCDSName: {Update: wantUpdate}}, UpdateMetadata{})
	for i := 0; i < count; i++ {
		if err := verifyClusterUpdate(ctx, clusterUpdateChs[i], wantUpdate, nil); err != nil {
			t.Fatal(err)
		}
	}
	cancelLastWatch()
	client.NewClusters(map[string]ClusterUpdateErrTuple{testCDSName: {Update: wantUpdate}}, UpdateMetadata{})
	for i := 0; i < count; i++ {
		func() {
			sCtx, sCancel := context.WithTimeout(ctx, defaultTestShortTimeout)
			defer sCancel()
			if u, err := clusterUpdateChs[i].Receive(sCtx); err != context.DeadlineExceeded {
				t.Errorf("unexpected ClusterUpdate: %v, %v, want channel recv timeout", u, err)
			}
		}()
	}
	newUpdate := ClusterUpdate{ClusterName: testEDSName, Raw: &anypb.Any{}}
	client.NewClusters(map[string]ClusterUpdateErrTuple{testCDSName: {Update: newUpdate}}, UpdateMetadata{})
	if err := verifyClusterUpdate(ctx, clusterUpdateChs[0], newUpdate, nil); err != nil {
		t.Fatal(err)
	}
}
func (s) TestClusterThreeWatchDifferentResourceName(t *testing.T) {
	apiClientCh, cleanup := overrideNewAPIClient()
	defer cleanup()
	client, err := newWithConfig(clientOpts(testXDSServer, false))
	if err != nil {
		t.Fatalf("failed to create client: %v", err)
	}
	defer client.Close()
	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
	defer cancel()
	c, err := apiClientCh.Receive(ctx)
	if err != nil {
		t.Fatalf("timeout when waiting for API client to be created: %v", err)
	}
	apiClient := c.(*testAPIClient)
	var clusterUpdateChs []*testutils.Channel
	const count = 2
	for i := 0; i < count; i++ {
		clusterUpdateCh := testutils.NewChannel()
		clusterUpdateChs = append(clusterUpdateChs, clusterUpdateCh)
		client.WatchCluster(testCDSName+"1", func(update ClusterUpdate, err error) {
			clusterUpdateCh.Send(ClusterUpdateErrTuple{Update: update, Err: err})
		})
		if i == 0 {
			if _, err := apiClient.addWatches[ClusterResource].Receive(ctx); err != nil {
				t.Fatalf("want new watch to start, got error %v", err)
			}
		}
	}
	clusterUpdateCh2 := testutils.NewChannel()
	client.WatchCluster(testCDSName+"2", func(update ClusterUpdate, err error) {
		clusterUpdateCh2.Send(ClusterUpdateErrTuple{Update: update, Err: err})
	})
	if _, err := apiClient.addWatches[ClusterResource].Receive(ctx); err != nil {
		t.Fatalf("want new watch to start, got error %v", err)
	}
	wantUpdate1 := ClusterUpdate{ClusterName: testEDSName + "1"}
	wantUpdate2 := ClusterUpdate{ClusterName: testEDSName + "2"}
	client.NewClusters(map[string]ClusterUpdateErrTuple{
		testCDSName + "1": {Update: wantUpdate1},
		testCDSName + "2": {Update: wantUpdate2},
	}, UpdateMetadata{})
	for i := 0; i < count; i++ {
		if err := verifyClusterUpdate(ctx, clusterUpdateChs[i], wantUpdate1, nil); err != nil {
			t.Fatal(err)
		}
	}
	if err := verifyClusterUpdate(ctx, clusterUpdateCh2, wantUpdate2, nil); err != nil {
		t.Fatal(err)
	}
}
func (s) TestClusterWatchAfterCache(t *testing.T) {
	apiClientCh, cleanup := overrideNewAPIClient()
	defer cleanup()
	client, err := newWithConfig(clientOpts(testXDSServer, false))
	if err != nil {
		t.Fatalf("failed to create client: %v", err)
	}
	defer client.Close()
	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
	defer cancel()
	c, err := apiClientCh.Receive(ctx)
	if err != nil {
		t.Fatalf("timeout when waiting for API client to be created: %v", err)
	}
	apiClient := c.(*testAPIClient)
	clusterUpdateCh := testutils.NewChannel()
	client.WatchCluster(testCDSName, func(update ClusterUpdate, err error) {
		clusterUpdateCh.Send(ClusterUpdateErrTuple{Update: update, Err: err})
	})
	if _, err := apiClient.addWatches[ClusterResource].Receive(ctx); err != nil {
		t.Fatalf("want new watch to start, got error %v", err)
	}
	wantUpdate := ClusterUpdate{ClusterName: testEDSName}
	client.NewClusters(map[string]ClusterUpdateErrTuple{
		testCDSName: {Update: wantUpdate},
	}, UpdateMetadata{})
	if err := verifyClusterUpdate(ctx, clusterUpdateCh, wantUpdate, nil); err != nil {
		t.Fatal(err)
	}
	clusterUpdateCh2 := testutils.NewChannel()
	client.WatchCluster(testCDSName, func(update ClusterUpdate, err error) {
		clusterUpdateCh2.Send(ClusterUpdateErrTuple{Update: update, Err: err})
	})
	sCtx, sCancel := context.WithTimeout(ctx, defaultTestShortTimeout)
	defer sCancel()
	if n, err := apiClient.addWatches[ClusterResource].Receive(sCtx); err != context.DeadlineExceeded {
		t.Fatalf("want no new watch to start (recv timeout), got resource name: %v error %v", n, err)
	}
	if err := verifyClusterUpdate(ctx, clusterUpdateCh2, wantUpdate, nil); err != nil {
		t.Fatal(err)
	}
	sCtx, sCancel = context.WithTimeout(ctx, defaultTestShortTimeout)
	defer sCancel()
	if u, err := clusterUpdateCh.Receive(sCtx); err != context.DeadlineExceeded {
		t.Errorf("unexpected clusterUpdate: %v, %v, want channel recv timeout", u, err)
	}
}