xdsclient: add test for custom xDS resource subscription (#8834)

Fixes #8449

RELEASE NOTES: none

Author: easwars

internal/xds/clients/xdsclient/test/custom_resource_watch_test.go

@@ -0,0 +1,254 @@
+/*
+ *
+ * Copyright 2026 gRPC authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package xdsclient_test
+
+import (
+	"context"
+	"fmt"
+	"strings"
+	"testing"
+
+	"github.com/google/uuid"
+	"google.golang.org/grpc/credentials/insecure"
+	"google.golang.org/grpc/internal/xds/clients"
+	"google.golang.org/grpc/internal/xds/clients/grpctransport"
+	"google.golang.org/grpc/internal/xds/clients/internal/testutils/fakeserver"
+	"google.golang.org/grpc/internal/xds/clients/xdsclient"
+	"google.golang.org/protobuf/proto"
+	"google.golang.org/protobuf/types/known/anypb"
+	"google.golang.org/protobuf/types/known/wrapperspb"
+
+	v3discoverypb "github.com/envoyproxy/go-control-plane/envoy/service/discovery/v3"
+)
+
+// customTestResourceType is a fake resource type used for testing.
+var customTestResourceType = xdsclient.ResourceType{
+	TypeURL:                    "type.googleapis.com/google.protobuf.StringValue",
+	TypeName:                   "CustomResource",
+	AllResourcesRequiredInSotW: false,
+	Decoder:                    customTestDecoder{},
+}
+
+type customTestDecoder struct{}
+
+func (customTestDecoder) Decode(resource *xdsclient.AnyProto, _ xdsclient.DecodeOptions) (*xdsclient.DecodeResult, error) {
+	any := resource.ToAny()
+	if any.GetTypeUrl() != customTestResourceType.TypeURL {
+		return nil, fmt.Errorf("unexpected resource type: %q", any.GetTypeUrl())
+	}
+	var val wrapperspb.StringValue
+	if err := proto.Unmarshal(any.GetValue(), &val); err != nil {
+		return nil, fmt.Errorf("failed to unmarshal resource: %v", err)
+	}
+
+	// We expect the value to be "name|content". This allows us to extract the
+	// name which is required by the xDS client to match the watch.
+	str := val.GetValue()
+	name, content, found := strings.Cut(str, "|")
+	if !found {
+		return nil, fmt.Errorf("invalid format: expected 'name|content', got %q", str)
+	}
+
+	return &xdsclient.DecodeResult{
+		Name:     name,
+		Resource: &customTestResourceData{val: content},
+	}, nil
+}
+
+type customTestResourceData struct {
+	val string
+}
+
+func (c *customTestResourceData) Equal(other xdsclient.ResourceData) bool {
+	if c == nil && other == nil {
+		return true
+	}
+	if c == nil || other == nil {
+		return false
+	}
+	o, ok := other.(*customTestResourceData)
+	if !ok {
+		return false
+	}
+	return c.val == o.val
+}
+
+func (c *customTestResourceData) Bytes() []byte {
+	return []byte(c.val)
+}
+
+// customTestWatcher is a watcher for the custom resource type.
+type customTestWatcher struct {
+	updateCh chan xdsclient.ResourceData
+	errCh    chan error
+}
+
+func newCustomTestWatcher() *customTestWatcher {
+	return &customTestWatcher{
+		updateCh: make(chan xdsclient.ResourceData, 1),
+		errCh:    make(chan error, 1),
+	}
+}
+
+func (w *customTestWatcher) ResourceChanged(update xdsclient.ResourceData, onDone func()) {
+	w.updateCh <- update
+	onDone()
+}
+
+func (w *customTestWatcher) ResourceError(err error, onDone func()) {
+	w.errCh <- fmt.Errorf("ResourceError: %v", err)
+	onDone()
+}
+
+func (w *customTestWatcher) AmbientError(err error, onDone func()) {
+	w.errCh <- fmt.Errorf("AmbientError: %v", err)
+	onDone()
+}
+
+// Tests that the xDS client can watch a custom resource type that is injected
+// via the config.
+func (s) TestCustomResourceWatch(t *testing.T) {
+	const authority = "my-authority"
+	resourceName := "xdstp://" + authority + "/" + customTestResourceType.TypeURL + "/my-resource"
+
+	tests := []struct {
+		name          string
+		resourceValue string
+		wantUpdate    string
+		wantNACK      string
+	}{
+		{
+			name:          "valid_resource",
+			resourceValue: resourceName + "|hello world",
+			wantUpdate:    "hello world",
+		},
+		{
+			name:          "decode_error",
+			resourceValue: "malformed-value",
+			wantNACK:      "invalid format: expected 'name|content'",
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			// Start a fake xDS management server.
+			mgmtServer, cleanup, err := fakeserver.StartServer(nil)
+			if err != nil {
+				t.Fatalf("Failed to start fake xDS server: %v", err)
+			}
+			defer cleanup()
+
+			resourceTypes := map[string]xdsclient.ResourceType{
+				customTestResourceType.TypeURL: customTestResourceType,
+			}
+			si := clients.ServerIdentifier{
+				ServerURI:  mgmtServer.Address,
+				Extensions: grpctransport.ServerIdentifierExtension{ConfigName: "insecure"},
+			}
+
+			configs := map[string]grpctransport.Config{"insecure": {Credentials: insecure.NewBundle()}}
+			nodeID := uuid.New().String()
+			xdsClientConfig := xdsclient.Config{
+				Servers:          []xdsclient.ServerConfig{{ServerIdentifier: si}},
+				Node:             clients.Node{ID: nodeID},
+				TransportBuilder: grpctransport.NewBuilder(configs),
+				ResourceTypes:    resourceTypes,
+				Authorities: map[string]xdsclient.Authority{
+					authority: {XDSServers: []xdsclient.ServerConfig{}},
+				},
+			}
+
+			client, err := xdsclient.New(xdsClientConfig)
+			if err != nil {
+				t.Fatalf("Failed to create xDS client: %v", err)
+			}
+			defer client.Close()
+
+			watcher := newCustomTestWatcher()
+			cancelWatch := client.WatchResource(customTestResourceType.TypeURL, resourceName, watcher)
+			defer cancelWatch()
+
+			// Wait for the discovery request.
+			ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
+			defer cancel()
+
+			// Verify the request type URL.
+			v, err := mgmtServer.XDSRequestChan.Receive(ctx)
+			if err != nil {
+				t.Fatalf("Timeout when waiting for a DiscoveryRequest message: %v", err)
+			}
+			req := v.(*fakeserver.Request).Req.(*v3discoverypb.DiscoveryRequest)
+			if req.GetTypeUrl() != customTestResourceType.TypeURL {
+				t.Fatalf("DiscoveryRequest TypeUrl = %v, want %v", req.GetTypeUrl(), customTestResourceType.TypeURL)
+			}
+			if len(req.GetResourceNames()) != 1 || req.GetResourceNames()[0] != resourceName {
+				t.Fatalf("DiscoveryRequest ResourceNames = %v, want [%v]", req.GetResourceNames(), resourceName)
+			}
+
+			// Send a response with the custom resource.
+			resource, err := anypb.New(&wrapperspb.StringValue{Value: test.resourceValue})
+			if err != nil {
+				t.Fatalf("Failed to marshal resource: %v", err)
+			}
+			mgmtServer.XDSResponseChan <- &fakeserver.Response{
+				Resp: &v3discoverypb.DiscoveryResponse{
+					TypeUrl:     customTestResourceType.TypeURL,
+					VersionInfo: "1",
+					Resources:   []*anypb.Any{resource},
+				},
+			}
+
+			if test.wantNACK != "" {
+				// Verify the NACK.
+				// We expect a new DiscoveryRequest with ErrorDetail set.
+				v, err = mgmtServer.XDSRequestChan.Receive(ctx)
+				if err != nil {
+					t.Fatalf("Timeout when waiting for NACK DiscoveryRequest message: %v", err)
+				}
+				req = v.(*fakeserver.Request).Req.(*v3discoverypb.DiscoveryRequest)
+				if req.GetTypeUrl() != customTestResourceType.TypeURL {
+					t.Fatalf("DiscoveryRequest TypeUrl = %v, want %v", req.GetTypeUrl(), customTestResourceType.TypeURL)
+				}
+				if req.GetErrorDetail() == nil {
+					t.Fatalf("DiscoveryRequest ErrorDetail is nil, want non-nil")
+				}
+				if !strings.Contains(req.GetErrorDetail().GetMessage(), test.wantNACK) {
+					t.Fatalf("DiscoveryRequest ErrorDetail = %v, want substring %q", req.GetErrorDetail().GetMessage(), test.wantNACK)
+				}
+				return
+			}
+
+			// Verify the update.
+			select {
+			case <-ctx.Done():
+				t.Fatalf("Timeout waiting for resource update")
+			case err := <-watcher.errCh:
+				t.Fatalf("Received unexpected error: %v", err)
+			case got := <-watcher.updateCh:
+				gotData, ok := got.(*customTestResourceData)
+				if !ok {
+					t.Fatalf("Received unexpected data type: %T", got)
+				}
+				if gotData.val != test.wantUpdate {
+					t.Fatalf("Received resource value %q, want %q", gotData.val, test.wantUpdate)
+				}
+			}
+		})
+	}
+}