Some checks failed
codecov / upload (push) Failing after 13m59s
CodeQL / Analyze (push) Failing after 1s
Testing / vet-proto (push) Failing after 1s
Testing / tests (arm, latest) (push) Waiting to run
Testing / extras (latest) (push) Failing after 1s
Testing / static checks (latest-1) (push) Failing after 1s
Testing / tests (-race, latest) (push) Failing after 1s
Testing / tests (i386, latest) (push) Failing after 1s
Testing / tests (latest) (push) Failing after 1s
Testing / tests (latest-1) (push) Failing after 1s
Lock Threads / lock (push) Failing after 1s
govulncheck / govulncheck (.) (push) Failing after 1s
govulncheck / govulncheck (./cmd/protoc-gen-go-grpc) (push) Failing after 1s
govulncheck / govulncheck (./gcp/observability) (push) Failing after 1s
govulncheck / govulncheck (./security/advancedtls) (push) Failing after 1s
govulncheck / govulncheck (./stats/opencensus) (push) Failing after 1s
Stale bot / stale (push) Failing after 1s
588 lines
20 KiB
Go
588 lines
20 KiB
Go
/*
|
|
*
|
|
* Copyright 2020 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 test
|
|
|
|
import (
|
|
"context"
|
|
"io"
|
|
"runtime/pprof"
|
|
"sync/atomic"
|
|
"testing"
|
|
|
|
"google.golang.org/grpc"
|
|
"google.golang.org/grpc/codes"
|
|
"google.golang.org/grpc/internal"
|
|
"google.golang.org/grpc/internal/envconfig"
|
|
"google.golang.org/grpc/internal/stubserver"
|
|
"google.golang.org/grpc/internal/testutils"
|
|
"google.golang.org/grpc/status"
|
|
|
|
testgrpc "google.golang.org/grpc/interop/grpc_testing"
|
|
testpb "google.golang.org/grpc/interop/grpc_testing"
|
|
)
|
|
|
|
type ctxKey string
|
|
|
|
// TestServerReturningContextError verifies that if a context error is returned
|
|
// by the service handler, the status will have the correct status code, not
|
|
// Unknown.
|
|
func (s) TestServerReturningContextError(t *testing.T) {
|
|
ss := &stubserver.StubServer{
|
|
EmptyCallF: func(context.Context, *testpb.Empty) (*testpb.Empty, error) {
|
|
return nil, context.DeadlineExceeded
|
|
},
|
|
FullDuplexCallF: func(testgrpc.TestService_FullDuplexCallServer) error {
|
|
return context.DeadlineExceeded
|
|
},
|
|
}
|
|
if err := ss.Start(nil); err != nil {
|
|
t.Fatalf("Error starting endpoint server: %v", err)
|
|
}
|
|
defer ss.Stop()
|
|
|
|
ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
|
|
defer cancel()
|
|
_, err := ss.Client.EmptyCall(ctx, &testpb.Empty{})
|
|
if s, ok := status.FromError(err); !ok || s.Code() != codes.DeadlineExceeded {
|
|
t.Fatalf("ss.Client.EmptyCall() got error %v; want <status with Code()=DeadlineExceeded>", err)
|
|
}
|
|
|
|
stream, err := ss.Client.FullDuplexCall(ctx)
|
|
if err != nil {
|
|
t.Fatalf("unexpected error starting the stream: %v", err)
|
|
}
|
|
_, err = stream.Recv()
|
|
if s, ok := status.FromError(err); !ok || s.Code() != codes.DeadlineExceeded {
|
|
t.Fatalf("ss.Client.FullDuplexCall().Recv() got error %v; want <status with Code()=DeadlineExceeded>", err)
|
|
}
|
|
|
|
}
|
|
|
|
func pprofCtxCollectLabels(ctx context.Context) map[string]string {
|
|
seenLabels := map[string]string{}
|
|
pprof.ForLabels(ctx, func(k, val string) bool {
|
|
seenLabels[k] = val
|
|
return true
|
|
})
|
|
return seenLabels
|
|
}
|
|
|
|
// TestServerSetGoroutineLabelsInContext verifies that when enabled, the
|
|
// grpc.method runtime/pprof goroutine label gets set in the context that's
|
|
// passed to the handlers.
|
|
func (s) TestServerSetGoroutineLabelsInContext(t *testing.T) {
|
|
testutils.SetEnvConfig(t, &envconfig.LabelServerGoroutines, envconfig.GoroutineLabelServerMethod)
|
|
ss := &stubserver.StubServer{
|
|
EmptyCallF: func(ctx context.Context, _ *testpb.Empty) (*testpb.Empty, error) {
|
|
ctxLabels := pprofCtxCollectLabels(ctx)
|
|
if val, ok := ctxLabels["grpc.method"]; !ok {
|
|
t.Errorf("missing \"grpc.method\" label; found labels: %v", ctxLabels)
|
|
} else if wantVal := "/grpc.testing.TestService/EmptyCall"; val != wantVal {
|
|
t.Errorf("unexpected value for \"grpc.method\" label %q; want %q", ctxLabels["grpc.method"], wantVal)
|
|
}
|
|
return &testpb.Empty{}, nil
|
|
},
|
|
FullDuplexCallF: func(stream testgrpc.TestService_FullDuplexCallServer) error {
|
|
ctxLabels := pprofCtxCollectLabels(stream.Context())
|
|
if val, ok := ctxLabels["grpc.method"]; !ok {
|
|
t.Errorf("missing \"grpc.method\" label; found labels: %v", ctxLabels)
|
|
} else if wantVal := "/grpc.testing.TestService/FullDuplexCall"; val != wantVal {
|
|
t.Errorf("unexpected value for \"grpc.method\" label %q; want %q", ctxLabels["grpc.method"], wantVal)
|
|
}
|
|
return nil
|
|
},
|
|
}
|
|
if err := ss.Start(nil); err != nil {
|
|
t.Fatalf("Error starting endpoint server: %v", err)
|
|
}
|
|
defer ss.Stop()
|
|
|
|
ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
|
|
defer cancel()
|
|
if _, err := ss.Client.EmptyCall(ctx, &testpb.Empty{}); err != nil {
|
|
t.Fatalf("ss.Client.EmptyCall() got error %v; want OK", err)
|
|
}
|
|
|
|
stream, err := ss.Client.FullDuplexCall(ctx)
|
|
if err != nil {
|
|
t.Fatalf("unexpected error starting the stream: %v", err)
|
|
}
|
|
if _, err = stream.Recv(); err != io.EOF {
|
|
t.Fatalf("ss.Client.FullDuplexCall().Recv() got error %v; want io.EOF", err)
|
|
}
|
|
}
|
|
|
|
// TestServerSetGoroutineLabelsInContextEnvVarDisabled verifies that when
|
|
// disable, the grpc.method runtime/pprof goroutine label does _not_ get set in
|
|
// the context that's passed to the handlers.
|
|
func (s) TestServerSetGoroutineLabelsInContextEnvVarDisabled(t *testing.T) {
|
|
testutils.SetEnvConfig(t, &envconfig.LabelServerGoroutines, 0)
|
|
ss := &stubserver.StubServer{
|
|
EmptyCallF: func(ctx context.Context, _ *testpb.Empty) (*testpb.Empty, error) {
|
|
ctxLabels := pprofCtxCollectLabels(ctx)
|
|
if val, ok := ctxLabels["grpc.method"]; ok {
|
|
t.Errorf("\"grpc.method\" label set with value %q; found labels: %v", val, ctxLabels)
|
|
}
|
|
return &testpb.Empty{}, nil
|
|
},
|
|
FullDuplexCallF: func(stream testgrpc.TestService_FullDuplexCallServer) error {
|
|
ctxLabels := pprofCtxCollectLabels(stream.Context())
|
|
if val, ok := ctxLabels["grpc.method"]; ok {
|
|
t.Errorf("\"grpc.method\" label set with value %q; found labels: %v", val, ctxLabels)
|
|
}
|
|
return nil
|
|
},
|
|
}
|
|
if err := ss.Start(nil); err != nil {
|
|
t.Fatalf("Error starting endpoint server: %v", err)
|
|
}
|
|
defer ss.Stop()
|
|
|
|
ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
|
|
defer cancel()
|
|
if _, err := ss.Client.EmptyCall(ctx, &testpb.Empty{}); err != nil {
|
|
t.Fatalf("ss.Client.EmptyCall() got error %v; want OK", err)
|
|
}
|
|
|
|
stream, err := ss.Client.FullDuplexCall(ctx)
|
|
if err != nil {
|
|
t.Fatalf("unexpected error starting the stream: %v", err)
|
|
}
|
|
if _, err = stream.Recv(); err != io.EOF {
|
|
t.Fatalf("ss.Client.FullDuplexCall().Recv() got error %v; want io.EOF", err)
|
|
}
|
|
}
|
|
|
|
func (s) TestChainUnaryServerInterceptor(t *testing.T) {
|
|
var (
|
|
firstIntKey = ctxKey("firstIntKey")
|
|
secondIntKey = ctxKey("secondIntKey")
|
|
)
|
|
|
|
firstInt := func(ctx context.Context, req any, _ *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (any, error) {
|
|
if ctx.Value(firstIntKey) != nil {
|
|
return nil, status.Errorf(codes.Internal, "first interceptor should not have %v in context", firstIntKey)
|
|
}
|
|
if ctx.Value(secondIntKey) != nil {
|
|
return nil, status.Errorf(codes.Internal, "first interceptor should not have %v in context", secondIntKey)
|
|
}
|
|
|
|
firstCtx := context.WithValue(ctx, firstIntKey, 0)
|
|
resp, err := handler(firstCtx, req)
|
|
if err != nil {
|
|
return nil, status.Errorf(codes.Internal, "failed to handle request at firstInt")
|
|
}
|
|
|
|
simpleResp, ok := resp.(*testpb.SimpleResponse)
|
|
if !ok {
|
|
return nil, status.Errorf(codes.Internal, "failed to get *testpb.SimpleResponse at firstInt")
|
|
}
|
|
return &testpb.SimpleResponse{
|
|
Payload: &testpb.Payload{
|
|
Type: simpleResp.GetPayload().GetType(),
|
|
Body: append(simpleResp.GetPayload().GetBody(), '1'),
|
|
},
|
|
}, nil
|
|
}
|
|
|
|
secondInt := func(ctx context.Context, req any, _ *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (any, error) {
|
|
if ctx.Value(firstIntKey) == nil {
|
|
return nil, status.Errorf(codes.Internal, "second interceptor should have %v in context", firstIntKey)
|
|
}
|
|
if ctx.Value(secondIntKey) != nil {
|
|
return nil, status.Errorf(codes.Internal, "second interceptor should not have %v in context", secondIntKey)
|
|
}
|
|
|
|
secondCtx := context.WithValue(ctx, secondIntKey, 1)
|
|
resp, err := handler(secondCtx, req)
|
|
if err != nil {
|
|
return nil, status.Errorf(codes.Internal, "failed to handle request at secondInt")
|
|
}
|
|
|
|
simpleResp, ok := resp.(*testpb.SimpleResponse)
|
|
if !ok {
|
|
return nil, status.Errorf(codes.Internal, "failed to get *testpb.SimpleResponse at secondInt")
|
|
}
|
|
return &testpb.SimpleResponse{
|
|
Payload: &testpb.Payload{
|
|
Type: simpleResp.GetPayload().GetType(),
|
|
Body: append(simpleResp.GetPayload().GetBody(), '2'),
|
|
},
|
|
}, nil
|
|
}
|
|
|
|
lastInt := func(ctx context.Context, req any, _ *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (any, error) {
|
|
if ctx.Value(firstIntKey) == nil {
|
|
return nil, status.Errorf(codes.Internal, "last interceptor should have %v in context", firstIntKey)
|
|
}
|
|
if ctx.Value(secondIntKey) == nil {
|
|
return nil, status.Errorf(codes.Internal, "last interceptor should not have %v in context", secondIntKey)
|
|
}
|
|
|
|
resp, err := handler(ctx, req)
|
|
if err != nil {
|
|
return nil, status.Errorf(codes.Internal, "failed to handle request at lastInt at lastInt")
|
|
}
|
|
|
|
simpleResp, ok := resp.(*testpb.SimpleResponse)
|
|
if !ok {
|
|
return nil, status.Errorf(codes.Internal, "failed to get *testpb.SimpleResponse at lastInt")
|
|
}
|
|
return &testpb.SimpleResponse{
|
|
Payload: &testpb.Payload{
|
|
Type: simpleResp.GetPayload().GetType(),
|
|
Body: append(simpleResp.GetPayload().GetBody(), '3'),
|
|
},
|
|
}, nil
|
|
}
|
|
|
|
sopts := []grpc.ServerOption{
|
|
grpc.ChainUnaryInterceptor(firstInt, secondInt, lastInt),
|
|
}
|
|
|
|
ss := &stubserver.StubServer{
|
|
UnaryCallF: func(context.Context, *testpb.SimpleRequest) (*testpb.SimpleResponse, error) {
|
|
payload, err := newPayload(testpb.PayloadType_COMPRESSABLE, 0)
|
|
if err != nil {
|
|
return nil, status.Errorf(codes.Aborted, "failed to make payload: %v", err)
|
|
}
|
|
|
|
return &testpb.SimpleResponse{
|
|
Payload: payload,
|
|
}, nil
|
|
},
|
|
}
|
|
if err := ss.Start(sopts); err != nil {
|
|
t.Fatalf("Error starting endpoint server: %v", err)
|
|
}
|
|
defer ss.Stop()
|
|
|
|
ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
|
|
defer cancel()
|
|
resp, err := ss.Client.UnaryCall(ctx, &testpb.SimpleRequest{})
|
|
if s, ok := status.FromError(err); !ok || s.Code() != codes.OK {
|
|
t.Fatalf("ss.Client.UnaryCall(ctx, _) = %v, %v; want nil, <status with Code()=OK>", resp, err)
|
|
}
|
|
|
|
respBytes := resp.Payload.GetBody()
|
|
if string(respBytes) != "321" {
|
|
t.Fatalf("invalid response: want=%s, but got=%s", "321", resp)
|
|
}
|
|
}
|
|
|
|
func (s) TestChainOnBaseUnaryServerInterceptor(t *testing.T) {
|
|
baseIntKey := ctxKey("baseIntKey")
|
|
|
|
baseInt := func(ctx context.Context, req any, _ *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (any, error) {
|
|
if ctx.Value(baseIntKey) != nil {
|
|
return nil, status.Errorf(codes.Internal, "base interceptor should not have %v in context", baseIntKey)
|
|
}
|
|
|
|
baseCtx := context.WithValue(ctx, baseIntKey, 1)
|
|
return handler(baseCtx, req)
|
|
}
|
|
|
|
chainInt := func(ctx context.Context, req any, _ *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (any, error) {
|
|
if ctx.Value(baseIntKey) == nil {
|
|
return nil, status.Errorf(codes.Internal, "chain interceptor should have %v in context", baseIntKey)
|
|
}
|
|
|
|
return handler(ctx, req)
|
|
}
|
|
|
|
sopts := []grpc.ServerOption{
|
|
grpc.UnaryInterceptor(baseInt),
|
|
grpc.ChainUnaryInterceptor(chainInt),
|
|
}
|
|
|
|
ss := &stubserver.StubServer{
|
|
EmptyCallF: func(context.Context, *testpb.Empty) (*testpb.Empty, error) {
|
|
return &testpb.Empty{}, nil
|
|
},
|
|
}
|
|
if err := ss.Start(sopts); err != nil {
|
|
t.Fatalf("Error starting endpoint server: %v", err)
|
|
}
|
|
defer ss.Stop()
|
|
|
|
ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
|
|
defer cancel()
|
|
resp, err := ss.Client.EmptyCall(ctx, &testpb.Empty{})
|
|
if s, ok := status.FromError(err); !ok || s.Code() != codes.OK {
|
|
t.Fatalf("ss.Client.EmptyCall(ctx, _) = %v, %v; want nil, <status with Code()=OK>", resp, err)
|
|
}
|
|
}
|
|
|
|
func (s) TestChainStreamServerInterceptor(t *testing.T) {
|
|
callCounts := make([]atomic.Int32, 4)
|
|
|
|
firstInt := func(srv any, stream grpc.ServerStream, _ *grpc.StreamServerInfo, handler grpc.StreamHandler) error {
|
|
if callCounts[0].Load() != 0 {
|
|
return status.Errorf(codes.Internal, "callCounts[0] should be 0, but got=%d", callCounts[0].Load())
|
|
}
|
|
if callCounts[1].Load() != 0 {
|
|
return status.Errorf(codes.Internal, "callCounts[1] should be 0, but got=%d", callCounts[1].Load())
|
|
}
|
|
if callCounts[2].Load() != 0 {
|
|
return status.Errorf(codes.Internal, "callCounts[2] should be 0, but got=%d", callCounts[2].Load())
|
|
}
|
|
if callCounts[3].Load() != 0 {
|
|
return status.Errorf(codes.Internal, "callCounts[3] should be 0, but got=%d", callCounts[3].Load())
|
|
}
|
|
callCounts[0].Add(1)
|
|
return handler(srv, stream)
|
|
}
|
|
|
|
secondInt := func(srv any, stream grpc.ServerStream, _ *grpc.StreamServerInfo, handler grpc.StreamHandler) error {
|
|
if callCounts[0].Load() != 1 {
|
|
return status.Errorf(codes.Internal, "callCounts[0] should be 1, but got=%d", callCounts[0].Load())
|
|
}
|
|
if callCounts[1].Load() != 0 {
|
|
return status.Errorf(codes.Internal, "callCounts[1] should be 0, but got=%d", callCounts[1].Load())
|
|
}
|
|
if callCounts[2].Load() != 0 {
|
|
return status.Errorf(codes.Internal, "callCounts[2] should be 0, but got=%d", callCounts[2].Load())
|
|
}
|
|
if callCounts[3].Load() != 0 {
|
|
return status.Errorf(codes.Internal, "callCounts[3] should be 0, but got=%d", callCounts[3].Load())
|
|
}
|
|
callCounts[1].Add(1)
|
|
return handler(srv, stream)
|
|
}
|
|
|
|
lastInt := func(srv any, stream grpc.ServerStream, _ *grpc.StreamServerInfo, handler grpc.StreamHandler) error {
|
|
if callCounts[0].Load() != 1 {
|
|
return status.Errorf(codes.Internal, "callCounts[0] should be 1, but got=%d", callCounts[0].Load())
|
|
}
|
|
if callCounts[1].Load() != 1 {
|
|
return status.Errorf(codes.Internal, "callCounts[1] should be 1, but got=%d", callCounts[1].Load())
|
|
}
|
|
if callCounts[2].Load() != 0 {
|
|
return status.Errorf(codes.Internal, "callCounts[2] should be 0, but got=%d", callCounts[2].Load())
|
|
}
|
|
if callCounts[3].Load() != 0 {
|
|
return status.Errorf(codes.Internal, "callCounts[3] should be 0, but got=%d", callCounts[3].Load())
|
|
}
|
|
callCounts[2].Add(1)
|
|
return handler(srv, stream)
|
|
}
|
|
|
|
sopts := []grpc.ServerOption{
|
|
grpc.ChainStreamInterceptor(firstInt, secondInt, lastInt),
|
|
}
|
|
|
|
ss := &stubserver.StubServer{
|
|
FullDuplexCallF: func(testgrpc.TestService_FullDuplexCallServer) error {
|
|
if callCounts[0].Load() != 1 {
|
|
return status.Errorf(codes.Internal, "callCounts[0] should be 1, but got=%d", callCounts[0].Load())
|
|
}
|
|
if callCounts[1].Load() != 1 {
|
|
return status.Errorf(codes.Internal, "callCounts[1] should be 1, but got=%d", callCounts[1].Load())
|
|
}
|
|
if callCounts[2].Load() != 1 {
|
|
return status.Errorf(codes.Internal, "callCounts[2] should be 0, but got=%d", callCounts[2].Load())
|
|
}
|
|
if callCounts[3].Load() != 0 {
|
|
return status.Errorf(codes.Internal, "callCounts[3] should be 0, but got=%d", callCounts[3].Load())
|
|
}
|
|
callCounts[3].Add(1)
|
|
return nil
|
|
},
|
|
}
|
|
if err := ss.Start(sopts); err != nil {
|
|
t.Fatalf("Error starting endpoint server: %v", err)
|
|
}
|
|
defer ss.Stop()
|
|
|
|
ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
|
|
defer cancel()
|
|
stream, err := ss.Client.FullDuplexCall(ctx)
|
|
if err != nil {
|
|
t.Fatalf("failed to FullDuplexCall: %v", err)
|
|
}
|
|
|
|
_, err = stream.Recv()
|
|
if err != io.EOF {
|
|
t.Fatalf("failed to recv from stream: %v", err)
|
|
}
|
|
|
|
if callCounts[3].Load() != 1 {
|
|
t.Fatalf("callCounts[3] should be 1, but got=%d", callCounts[3].Load())
|
|
}
|
|
}
|
|
|
|
// Test verifies that only unary interceptors are invoked for unary RPCs and
|
|
// only streaming interceptors are invoked for streaming RPCs.
|
|
func (s) TestInterceptorSegregation(t *testing.T) {
|
|
var unaryCalled, streamCalled atomic.Bool
|
|
|
|
unaryInt := func(ctx context.Context, req any, _ *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (any, error) {
|
|
unaryCalled.Store(true)
|
|
return handler(ctx, req)
|
|
}
|
|
streamInt := func(srv any, ss grpc.ServerStream, _ *grpc.StreamServerInfo, handler grpc.StreamHandler) error {
|
|
streamCalled.Store(true)
|
|
return handler(srv, ss)
|
|
}
|
|
sopts := []grpc.ServerOption{grpc.UnaryInterceptor(unaryInt), grpc.StreamInterceptor(streamInt)}
|
|
ss := &stubserver.StubServer{
|
|
UnaryCallF: func(context.Context, *testpb.SimpleRequest) (*testpb.SimpleResponse, error) {
|
|
return &testpb.SimpleResponse{}, nil
|
|
},
|
|
FullDuplexCallF: func(testgrpc.TestService_FullDuplexCallServer) error {
|
|
return nil
|
|
},
|
|
}
|
|
if err := ss.Start(sopts); err != nil {
|
|
t.Fatalf("Error starting endpoint server: %v", err)
|
|
}
|
|
defer ss.Stop()
|
|
|
|
// Make a unary RPC and ensure that only the unary interceptor was invoked.
|
|
ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
|
|
defer cancel()
|
|
if _, err := ss.Client.UnaryCall(ctx, &testpb.SimpleRequest{}); err != nil {
|
|
t.Fatalf("ss.Client.UnaryCall failed: %v", err)
|
|
}
|
|
if !unaryCalled.Load() {
|
|
t.Error("Unary interceptor was not called for Unary RPC")
|
|
}
|
|
if streamCalled.Load() {
|
|
t.Error("Stream interceptor was called for Unary RPC")
|
|
}
|
|
|
|
// Make a streaming RPC and ensure that only the streaming interceptor was
|
|
// invoked.
|
|
unaryCalled.Store(false)
|
|
streamCalled.Store(false)
|
|
stream, err := ss.Client.FullDuplexCall(ctx)
|
|
if err != nil {
|
|
t.Fatalf("ss.Client.FullDuplexCall failed: %v", err)
|
|
}
|
|
if _, err := stream.Recv(); err != io.EOF {
|
|
t.Fatalf("stream.Recv() returned %v, want io.EOF", err)
|
|
}
|
|
if unaryCalled.Load() {
|
|
t.Error("Unary interceptor was called for Streaming RPC")
|
|
}
|
|
if !streamCalled.Load() {
|
|
t.Error("Stream interceptor was not called for Streaming RPC")
|
|
}
|
|
}
|
|
|
|
type wrappedTestStreamKey struct{}
|
|
|
|
type wrappedTestStream struct {
|
|
grpc.ServerStream
|
|
}
|
|
|
|
func (w *wrappedTestStream) Context() context.Context {
|
|
return context.WithValue(w.ServerStream.Context(), wrappedTestStreamKey{}, w)
|
|
}
|
|
|
|
// Test verifies that an internal xDS filter wrapper option configured on the
|
|
// server gets invoked and can wrap the ServerStream for both Unary and
|
|
// Streaming RPCs.
|
|
func (s) TestXDSFilterWrapperOption(t *testing.T) {
|
|
var wrapperCallCount atomic.Int32
|
|
wrapper := func(ss grpc.ServerStream) (grpc.ServerStream, error) {
|
|
wrapperCallCount.Add(1)
|
|
return &wrappedTestStream{ServerStream: ss}, nil
|
|
}
|
|
|
|
opt := internal.XDSFilterWrapperOption.(func(func(grpc.ServerStream) (grpc.ServerStream, error)) grpc.ServerOption)(wrapper)
|
|
|
|
ss := &stubserver.StubServer{
|
|
EmptyCallF: func(ctx context.Context, _ *testpb.Empty) (*testpb.Empty, error) {
|
|
if _, ok := ctx.Value(wrappedTestStreamKey{}).(*wrappedTestStream); !ok {
|
|
return nil, status.Errorf(codes.Internal, "context value is %T, want *wrappedTestStream", ctx.Value(wrappedTestStreamKey{}))
|
|
}
|
|
return &testpb.Empty{}, nil
|
|
},
|
|
FullDuplexCallF: func(stream testgrpc.TestService_FullDuplexCallServer) error {
|
|
if _, ok := stream.Context().Value(wrappedTestStreamKey{}).(*wrappedTestStream); !ok {
|
|
return status.Errorf(codes.Internal, "context value is %T, want *wrappedTestStream", stream.Context().Value(wrappedTestStreamKey{}))
|
|
}
|
|
return nil
|
|
},
|
|
}
|
|
if err := ss.Start([]grpc.ServerOption{opt}); err != nil {
|
|
t.Fatalf("Error starting endpoint server: %v", err)
|
|
}
|
|
defer ss.Stop()
|
|
|
|
ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
|
|
defer cancel()
|
|
if _, err := ss.Client.EmptyCall(ctx, &testpb.Empty{}); err != nil {
|
|
t.Fatalf("EmptyCall failed: %v", err)
|
|
}
|
|
if got := wrapperCallCount.Load(); got != 1 {
|
|
t.Fatalf("XDSFilterWrapperOption callback call count for Unary RPC got %d, want 1", got)
|
|
}
|
|
|
|
stream, err := ss.Client.FullDuplexCall(ctx)
|
|
if err != nil {
|
|
t.Fatalf("FullDuplexCall failed: %v", err)
|
|
}
|
|
if _, err = stream.Recv(); err != io.EOF {
|
|
t.Fatalf("Recv failed: %v", err)
|
|
}
|
|
if got := wrapperCallCount.Load(); got != 2 {
|
|
t.Fatalf("XDSFilterWrapperOption callback call count for Streaming RPC got %d, want 2", got)
|
|
}
|
|
}
|
|
|
|
// Test verifies that if an internal xDS filter wrapper returns an error,
|
|
// the RPC is rejected early with that status error before executing
|
|
// handlers.
|
|
func (s) TestXDSFilterWrapperOption_EarlyRejection(t *testing.T) {
|
|
wrapper := func(grpc.ServerStream) (grpc.ServerStream, error) {
|
|
return nil, status.Error(codes.PermissionDenied, "early rejection by internal wrapper")
|
|
}
|
|
|
|
opt := internal.XDSFilterWrapperOption.(func(func(grpc.ServerStream) (grpc.ServerStream, error)) grpc.ServerOption)(wrapper)
|
|
|
|
ss := &stubserver.StubServer{
|
|
EmptyCallF: func(context.Context, *testpb.Empty) (*testpb.Empty, error) {
|
|
return &testpb.Empty{}, nil
|
|
},
|
|
FullDuplexCallF: func(testgrpc.TestService_FullDuplexCallServer) error {
|
|
return nil
|
|
},
|
|
}
|
|
if err := ss.Start([]grpc.ServerOption{opt}); err != nil {
|
|
t.Fatalf("Error starting endpoint server: %v", err)
|
|
}
|
|
defer ss.Stop()
|
|
|
|
ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
|
|
defer cancel()
|
|
if _, err := ss.Client.EmptyCall(ctx, &testpb.Empty{}); status.Code(err) != codes.PermissionDenied {
|
|
t.Fatalf("EmptyCall failed with error %v; want PermissionDenied", err)
|
|
}
|
|
|
|
stream, err := ss.Client.FullDuplexCall(ctx)
|
|
if err != nil {
|
|
t.Fatalf("FullDuplexCall failed: %v", err)
|
|
}
|
|
if _, err = stream.Recv(); status.Code(err) != codes.PermissionDenied {
|
|
t.Fatalf("stream.Recv() got error %v; want PermissionDenied", err)
|
|
}
|
|
}
|