Repository: grpc/grpc-go
File: xds/internal/xdsclient/tests/lds_watchers_test.go (Go)

Code before:
	ldsCancel1 := xdsresource.WatchListener(client, ldsName, lw1)
	defer ldsCancel1()

	// Configure the management server to return a single listener
	// resource, corresponding to the one we registered a watch for.
	resources := e2e.UpdateOptions{
		NodeID:         nodeID,
		Listeners:      []*v3listenerpb.Listener{e2e.DefaultClientListener(ldsName, rdsName)},
		SkipValidation: true,
	}
	ctx, cancel := context.WithTimeout(context.Background(), 1000*defaultTestTimeout)
	defer cancel()
	if err := mgmtServer.Update(ctx, resources); err != nil {
		t.Fatalf("Failed to update management server with resources: %v, err: %v", resources, err)
	}

	// Verify the contents of the received update.
	wantUpdate := listenerUpdateErrTuple{
		update: xdsresource.ListenerUpdate{
			RouteConfigName: rdsName,
			HTTPFilters:     []xdsresource.HTTPFilter{{Name: "router"}},
		},
	}
	if err := verifyListenerUpdate(ctx, lw1.updateCh, wantUpdate); err != nil {
		t.Fatal(err)
	}

	// Configure the management server to return a single listener resource
	// which is expected to be NACKed by the client.
	resources = e2e.UpdateOptions{
		NodeID:         nodeID,
		Listeners:      []*v3listenerpb.Listener{badListenerResource(t, ldsName)},
		SkipValidation: true,
	}
	if err := mgmtServer.Update(ctx, resources); err != nil {
		t.Fatalf("Failed to update management server with resources: %v, err: %v", resources, err)
	}

	// Verify that the expected error is propagated to the existing watcher.
	u, err := lw1.updateCh.Receive(ctx)
	if err != nil {
		t.Fatalf("timeout when waiting for a listener resource from the management server: %v", err)
	}
	gotErr := u.(listenerUpdateErrTuple).err
	if gotErr == nil || !strings.Contains(gotErr.Error(), wantListenerNACKErr) {
		t.Fatalf("Update received with error: %v, want %q", gotErr, wantListenerNACKErr)
	}

	// Register another watch for the same resource. This should get the update
	// and error from the cache.
	lw2 := newListenerWatcherMultiple(2)

Review comment (refactor):
Should we implement a helper for this?

```golang
func verifyListenerError(ctx context.Context, updateCh *testutils.Channel, wantErr string) error {
	u, err := updateCh.Receive(ctx)
	if err != nil {
		return fmt.Errorf("timeout when waiting for a listener update from the management server: %v", err)
	}
	gotErr := u.(listenerUpdateErrTuple).err
	if gotErr == nil || !strings.Contains(gotErr.Error(), wantErr) {
		return fmt.Errorf("update received with error: %v, want %q", gotErr, wantErr)
	}
}
```

Code after:
	defer close()

	// Register a watch for a listener resource and have the watch
	// callback push the received update on to a channel.
	lw := newListenerWatcher()
	ldsCancel := xdsresource.WatchListener(client, ldsName, lw)
	defer ldsCancel()

	// Configure the management server to return a single listener resource
	// which is expected to be NACKed by the client.
	resources := e2e.UpdateOptions{
		NodeID:         nodeID,
		Listeners:      []*v3listenerpb.Listener{badListenerResource(t, ldsName)},
		SkipValidation: true,
	}
	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
	defer cancel()
	if err := mgmtServer.Update(ctx, resources); err != nil {
		t.Fatalf("Failed to update management server with resources: %v, err: %v", resources, err)
	}

	// Verify that the expected error is propagated to the watcher.
	// Verify that the expected error is propagated to the existing watcher.
	if err := verifyUnknownListenerError(ctx, lw.updateCh, wantListenerNACKErr); err != nil {
		t.Fatal(err)
	}

	// Verify that the expected error is propagated to the new watcher as well.
	lw2 := newListenerWatcher()
	ldsCancel2 := xdsresource.WatchListener(client, ldsName, lw2)
	defer ldsCancel2()
	// Verify that the expected error is propagated to the existing watcher.
	if err := verifyUnknownListenerError(ctx, lw2.updateCh, wantListenerNACKErr); err != nil {
		t.Fatal(err)
	}
}

// TestLDSWatch_ResourceCaching_WithNACKError covers the case where a watch is
// registered for a resource which is already present in the cache with an old
// good update as well as latest NACK error. The test verifies that new watcher
// receives both good update and error without a new resource request being
// sent to the management server.
func TestLDSWatch_ResourceCaching_NACKError(t *testing.T) {
	firstRequestReceived := false
	firstAckReceived := grpcsync.NewEvent()
	secondAckReceived := grpcsync.NewEvent()
	secondRequestReceived := grpcsync.NewEvent()

	mgmtServer := e2e.StartManagementServer(t, e2e.ManagementServerOptions{
		OnStreamRequest: func(id int64, req *v3discoverypb.DiscoveryRequest) error {
			// The first request has an empty version string.