Repository: grpc/grpc-go
File: internal/xds/xdsdepmgr/xds_dependency_manager_test.go (Go)

Code before:
		done:     make(chan struct{}),
	}

	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
	defer cancel()

	resources := e2e.DefaultClientResources(e2e.ResourceParams{
		NodeID:     nodeID,
		DialTarget: defaultTestServiceName,
		Host:       "localhost",
		Port:       8080,
		SecLevel:   e2e.SecurityLevelNone,
	})
	resources.Clusters[0].LrsServer = &v3corepb.ConfigSource{ConfigSourceSpecifier: &v3corepb.ConfigSource_Ads{}}

	if err := mgmtServer.Update(ctx, resources); err != nil {
		t.Fatal(err)
	}

	dm := xdsdepmgr.New(defaultTestServiceName, defaultTestServiceName, xdsClient, watcher)
	defer dm.Close()

	// Defer closing the watcher's done channel to prevent a potential test
	// hang. The management server can send errors repeatedly, which could
	// otherwise block the dependency manager's update.
	defer close(watcher.done)

	wantXdsConfig := makeXDSConfig(resources.Routes[0].Name, resources.Clusters[0].Name, resources.Clusters[0].EdsClusterConfig.ServiceName, "localhost:8080")
	wantXdsConfig.Clusters[resources.Clusters[0].Name] = &xdsresource.ClusterResult{Err: fmt.Errorf("[xDS node id: %v]: %v", nodeID, fmt.Errorf("unsupported config_source_specifier *corev3.ConfigSource_Ads in lrs_server field"))}

	if err := verifyXDSConfig(ctx, watcher.updateCh, watcher.errorCh, wantXdsConfig); err != nil {
		t.Fatal(err)
	}
}

// Tests the case where the dependency manager receives a cluster resource
// ambient error. A valid cluster resource is sent first, then an invalid
// one and then the valid resource again. The valid resource is sent again
// to make sure that the ambient error reaches the dependency manager since
// there is no other way to wait for it.
func (s) TestClusterAmbientError(t *testing.T) {
	// Expect a warning log for the ambient error.
	grpctest.ExpectWarning("Cluster resource ambient error")

	nodeID, mgmtServer, xdsClient := setupManagementServerAndClient(t, false)

	watcher := &testWatcher{
		updateCh: make(chan *xdsresource.XDSConfig),
		errorCh:  make(chan error),
	}


Review comment (bug):
Does it make more sense to instead change the `testWatcher` implementation to handle this by making the update and error channels be of type `testutils.Channel` instead of them being vanilla channels. With the former, you would use `SendContext` to send a value on the channel and use the test's context. That way, when the test ends (success or failure), the test's context will be cancelled, which will call any pending calls to `SendContext` to get unblocked.

Code after:
	}
	if err := verifyXDSConfig(ctx, watcher.updateCh, watcher.errorCh, wantXdsConfig); err != nil {
		t.Fatal(err)
	}
}

// Tests the case where the dependency manager receives a cluster resource error
// and verifies that Update is called with XDSConfig containing cluster error.
func (s) TestClusterResourceError(t *testing.T) {
	nodeID, mgmtServer, xdsClient := setupManagementServerAndClient(t, false)

	watcher := newTestWatcher()

	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
	defer cancel()

	resources := e2e.DefaultClientResources(e2e.ResourceParams{
		NodeID:     nodeID,
		DialTarget: defaultTestServiceName,
		Host:       "localhost",
		Port:       8080,
		SecLevel:   e2e.SecurityLevelNone,
	})
	resources.Clusters[0].LrsServer = &v3corepb.ConfigSource{ConfigSourceSpecifier: &v3corepb.ConfigSource_Ads{}}

	if err := mgmtServer.Update(ctx, resources); err != nil {
		t.Fatal(err)
	}

	dm := xdsdepmgr.New(defaultTestServiceName, defaultTestServiceName, xdsClient, watcher)
	defer dm.Close()

	// Defer closing the watcher to prevent a potential hang. The management
	// server may send repeated errors, triggering updates that hold the
	// dependency manager's mutex. This defer is defined last so it executes
	// first (before dm.Close()). If we don't stop the watcher, dm.Close() will
	// deadlock waiting for the mutex currently held by the blocking Update
	// call.
	defer watcher.close()

	wantXdsConfig := makeXDSConfig(resources.Routes[0].Name, resources.Clusters[0].Name, resources.Clusters[0].EdsClusterConfig.ServiceName, "localhost:8080")
	wantXdsConfig.Clusters[resources.Clusters[0].Name] = &xdsresource.ClusterResult{Err: fmt.Errorf("[xDS node id: %v]: %v", nodeID, fmt.Errorf("unsupported config_source_specifier *corev3.ConfigSource_Ads in lrs_server field"))}

	if err := verifyXDSConfig(ctx, watcher.updateCh, watcher.errorCh, wantXdsConfig); err != nil {
		t.Fatal(err)
	}
}

// Tests the case where the dependency manager receives a cluster resource
// ambient error. A valid cluster resource is sent first, then an invalid
// one and then the valid resource again. The valid resource is sent again