upstream-grpc-go/internal/xds/clients/xdsclient/test/custom_resource_watch_test.go

/*
 *
 * 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)
				}
			}
		})
	}
}