vendor/google.golang.org/grpc/transport/handler_server_test.go

   1 /*
   2  *
   3  * Copyright 2016 gRPC authors.
   4  *
   5  * Licensed under the Apache License, Version 2.0 (the "License");
   6  * you may not use this file except in compliance with the License.
   7  * You may obtain a copy of the License at
   8  *
   9  *     http://www.apache.org/licenses/LICENSE-2.0
  10  *
  11  * Unless required by applicable law or agreed to in writing, software
  12  * distributed under the License is distributed on an "AS IS" BASIS,
  13  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  14  * See the License for the specific language governing permissions and
  15  * limitations under the License.
  16  *
  17  */
  18
  19 package transport
  20
  21 import (
  22         "errors"
  23         "fmt"
  24         "io"
  25         "net/http"
  26         "net/http/httptest"
  27         "net/url"
  28         "reflect"
  29         "sync"
  30         "testing"
  31         "time"
  32
  33         "github.com/golang/protobuf/proto"
  34         dpb "github.com/golang/protobuf/ptypes/duration"
  35         "golang.org/x/net/context"
  36         epb "google.golang.org/genproto/googleapis/rpc/errdetails"
  37         "google.golang.org/grpc/codes"
  38         "google.golang.org/grpc/metadata"
  39         "google.golang.org/grpc/status"
  40 )
  41
  42 func TestHandlerTransport_NewServerHandlerTransport(t *testing.T) {
  43         type testCase struct {
  44                 name    string
  45                 req     *http.Request
  46                 wantErr string
  47                 modrw   func(http.ResponseWriter) http.ResponseWriter
  48                 check   func(*serverHandlerTransport, *testCase) error
  49         }
  50         tests := []testCase{
  51                 {
  52                         name: "http/1.1",
  53                         req: &http.Request{
  54                                 ProtoMajor: 1,
  55                                 ProtoMinor: 1,
  56                         },
  57                         wantErr: "gRPC requires HTTP/2",
  58                 },
  59                 {
  60                         name: "bad method",
  61                         req: &http.Request{
  62                                 ProtoMajor: 2,
  63                                 Method:     "GET",
  64                                 Header:     http.Header{},
  65                                 RequestURI: "/",
  66                         },
  67                         wantErr: "invalid gRPC request method",
  68                 },
  69                 {
  70                         name: "bad content type",
  71                         req: &http.Request{
  72                                 ProtoMajor: 2,
  73                                 Method:     "POST",
  74                                 Header: http.Header{
  75                                         "Content-Type": {"application/foo"},
  76                                 },
  77                                 RequestURI: "/service/foo.bar",
  78                         },
  79                         wantErr: "invalid gRPC request content-type",
  80                 },
  81                 {
  82                         name: "not flusher",
  83                         req: &http.Request{
  84                                 ProtoMajor: 2,
  85                                 Method:     "POST",
  86                                 Header: http.Header{
  87                                         "Content-Type": {"application/grpc"},
  88                                 },
  89                                 RequestURI: "/service/foo.bar",
  90                         },
  91                         modrw: func(w http.ResponseWriter) http.ResponseWriter {
  92                                 // Return w without its Flush method
  93                                 type onlyCloseNotifier interface {
  94                                         http.ResponseWriter
  95                                         http.CloseNotifier
  96                                 }
  97                                 return struct{ onlyCloseNotifier }{w.(onlyCloseNotifier)}
  98                         },
  99                         wantErr: "gRPC requires a ResponseWriter supporting http.Flusher",
 100                 },
 101                 {
 102                         name: "not closenotifier",
 103                         req: &http.Request{
 104                                 ProtoMajor: 2,
 105                                 Method:     "POST",
 106                                 Header: http.Header{
 107                                         "Content-Type": {"application/grpc"},
 108                                 },
 109                                 RequestURI: "/service/foo.bar",
 110                         },
 111                         modrw: func(w http.ResponseWriter) http.ResponseWriter {
 112                                 // Return w without its CloseNotify method
 113                                 type onlyFlusher interface {
 114                                         http.ResponseWriter
 115                                         http.Flusher
 116                                 }
 117                                 return struct{ onlyFlusher }{w.(onlyFlusher)}
 118                         },
 119                         wantErr: "gRPC requires a ResponseWriter supporting http.CloseNotifier",
 120                 },
 121                 {
 122                         name: "valid",
 123                         req: &http.Request{
 124                                 ProtoMajor: 2,
 125                                 Method:     "POST",
 126                                 Header: http.Header{
 127                                         "Content-Type": {"application/grpc"},
 128                                 },
 129                                 URL: &url.URL{
 130                                         Path: "/service/foo.bar",
 131                                 },
 132                                 RequestURI: "/service/foo.bar",
 133                         },
 134                         check: func(t *serverHandlerTransport, tt *testCase) error {
 135                                 if t.req != tt.req {
 136                                         return fmt.Errorf("t.req = %p; want %p", t.req, tt.req)
 137                                 }
 138                                 if t.rw == nil {
 139                                         return errors.New("t.rw = nil; want non-nil")
 140                                 }
 141                                 return nil
 142                         },
 143                 },
 144                 {
 145                         name: "with timeout",
 146                         req: &http.Request{
 147                                 ProtoMajor: 2,
 148                                 Method:     "POST",
 149                                 Header: http.Header{
 150                                         "Content-Type": []string{"application/grpc"},
 151                                         "Grpc-Timeout": {"200m"},
 152                                 },
 153                                 URL: &url.URL{
 154                                         Path: "/service/foo.bar",
 155                                 },
 156                                 RequestURI: "/service/foo.bar",
 157                         },
 158                         check: func(t *serverHandlerTransport, tt *testCase) error {
 159                                 if !t.timeoutSet {
 160                                         return errors.New("timeout not set")
 161                                 }
 162                                 if want := 200 * time.Millisecond; t.timeout != want {
 163                                         return fmt.Errorf("timeout = %v; want %v", t.timeout, want)
 164                                 }
 165                                 return nil
 166                         },
 167                 },
 168                 {
 169                         name: "with bad timeout",
 170                         req: &http.Request{
 171                                 ProtoMajor: 2,
 172                                 Method:     "POST",
 173                                 Header: http.Header{
 174                                         "Content-Type": []string{"application/grpc"},
 175                                         "Grpc-Timeout": {"tomorrow"},
 176                                 },
 177                                 URL: &url.URL{
 178                                         Path: "/service/foo.bar",
 179                                 },
 180                                 RequestURI: "/service/foo.bar",
 181                         },
 182                         wantErr: `stream error: code = Internal desc = "malformed time-out: transport: timeout unit is not recognized: \"tomorrow\""`,
 183                 },
 184                 {
 185                         name: "with metadata",
 186                         req: &http.Request{
 187                                 ProtoMajor: 2,
 188                                 Method:     "POST",
 189                                 Header: http.Header{
 190                                         "Content-Type": []string{"application/grpc"},
 191                                         "meta-foo":     {"foo-val"},
 192                                         "meta-bar":     {"bar-val1", "bar-val2"},
 193                                         "user-agent":   {"x/y a/b"},
 194                                 },
 195                                 URL: &url.URL{
 196                                         Path: "/service/foo.bar",
 197                                 },
 198                                 RequestURI: "/service/foo.bar",
 199                         },
 200                         check: func(ht *serverHandlerTransport, tt *testCase) error {
 201                                 want := metadata.MD{
 202                                         "meta-bar":   {"bar-val1", "bar-val2"},
 203                                         "user-agent": {"x/y a/b"},
 204                                         "meta-foo":   {"foo-val"},
 205                                 }
 206
 207                                 if !reflect.DeepEqual(ht.headerMD, want) {
 208                                         return fmt.Errorf("metdata = %#v; want %#v", ht.headerMD, want)
 209                                 }
 210                                 return nil
 211                         },
 212                 },
 213         }
 214
 215         for _, tt := range tests {
 216                 rw := newTestHandlerResponseWriter()
 217                 if tt.modrw != nil {
 218                         rw = tt.modrw(rw)
 219                 }
 220                 got, gotErr := NewServerHandlerTransport(rw, tt.req)
 221                 if (gotErr != nil) != (tt.wantErr != "") || (gotErr != nil && gotErr.Error() != tt.wantErr) {
 222                         t.Errorf("%s: error = %v; want %q", tt.name, gotErr, tt.wantErr)
 223                         continue
 224                 }
 225                 if gotErr != nil {
 226                         continue
 227                 }
 228                 if tt.check != nil {
 229                         if err := tt.check(got.(*serverHandlerTransport), &tt); err != nil {
 230                                 t.Errorf("%s: %v", tt.name, err)
 231                         }
 232                 }
 233         }
 234 }
 235
 236 type testHandlerResponseWriter struct {
 237         *httptest.ResponseRecorder
 238         closeNotify chan bool
 239 }
 240
 241 func (w testHandlerResponseWriter) CloseNotify() <-chan bool { return w.closeNotify }
 242 func (w testHandlerResponseWriter) Flush()                   {}
 243
 244 func newTestHandlerResponseWriter() http.ResponseWriter {
 245         return testHandlerResponseWriter{
 246                 ResponseRecorder: httptest.NewRecorder(),
 247                 closeNotify:      make(chan bool, 1),
 248         }
 249 }
 250
 251 type handleStreamTest struct {
 252         t     *testing.T
 253         bodyw *io.PipeWriter
 254         req   *http.Request
 255         rw    testHandlerResponseWriter
 256         ht    *serverHandlerTransport
 257 }
 258
 259 func newHandleStreamTest(t *testing.T) *handleStreamTest {
 260         bodyr, bodyw := io.Pipe()
 261         req := &http.Request{
 262                 ProtoMajor: 2,
 263                 Method:     "POST",
 264                 Header: http.Header{
 265                         "Content-Type": {"application/grpc"},
 266                 },
 267                 URL: &url.URL{
 268                         Path: "/service/foo.bar",
 269                 },
 270                 RequestURI: "/service/foo.bar",
 271                 Body:       bodyr,
 272         }
 273         rw := newTestHandlerResponseWriter().(testHandlerResponseWriter)
 274         ht, err := NewServerHandlerTransport(rw, req)
 275         if err != nil {
 276                 t.Fatal(err)
 277         }
 278         return &handleStreamTest{
 279                 t:     t,
 280                 bodyw: bodyw,
 281                 ht:    ht.(*serverHandlerTransport),
 282                 rw:    rw,
 283         }
 284 }
 285
 286 func TestHandlerTransport_HandleStreams(t *testing.T) {
 287         st := newHandleStreamTest(t)
 288         handleStream := func(s *Stream) {
 289                 if want := "/service/foo.bar"; s.method != want {
 290                         t.Errorf("stream method = %q; want %q", s.method, want)
 291                 }
 292                 st.bodyw.Close() // no body
 293                 st.ht.WriteStatus(s, status.New(codes.OK, ""))
 294         }
 295         st.ht.HandleStreams(
 296                 func(s *Stream) { go handleStream(s) },
 297                 func(ctx context.Context, method string) context.Context { return ctx },
 298         )
 299         wantHeader := http.Header{
 300                 "Date":         nil,
 301                 "Content-Type": {"application/grpc"},
 302                 "Trailer":      {"Grpc-Status", "Grpc-Message", "Grpc-Status-Details-Bin"},
 303                 "Grpc-Status":  {"0"},
 304         }
 305         if !reflect.DeepEqual(st.rw.HeaderMap, wantHeader) {
 306                 t.Errorf("Header+Trailer Map: %#v; want %#v", st.rw.HeaderMap, wantHeader)
 307         }
 308 }
 309
 310 // Tests that codes.Unimplemented will close the body, per comment in handler_server.go.
 311 func TestHandlerTransport_HandleStreams_Unimplemented(t *testing.T) {
 312         handleStreamCloseBodyTest(t, codes.Unimplemented, "thingy is unimplemented")
 313 }
 314
 315 // Tests that codes.InvalidArgument will close the body, per comment in handler_server.go.
 316 func TestHandlerTransport_HandleStreams_InvalidArgument(t *testing.T) {
 317         handleStreamCloseBodyTest(t, codes.InvalidArgument, "bad arg")
 318 }
 319
 320 func handleStreamCloseBodyTest(t *testing.T, statusCode codes.Code, msg string) {
 321         st := newHandleStreamTest(t)
 322
 323         handleStream := func(s *Stream) {
 324                 st.ht.WriteStatus(s, status.New(statusCode, msg))
 325         }
 326         st.ht.HandleStreams(
 327                 func(s *Stream) { go handleStream(s) },
 328                 func(ctx context.Context, method string) context.Context { return ctx },
 329         )
 330         wantHeader := http.Header{
 331                 "Date":         nil,
 332                 "Content-Type": {"application/grpc"},
 333                 "Trailer":      {"Grpc-Status", "Grpc-Message", "Grpc-Status-Details-Bin"},
 334                 "Grpc-Status":  {fmt.Sprint(uint32(statusCode))},
 335                 "Grpc-Message": {encodeGrpcMessage(msg)},
 336         }
 337
 338         if !reflect.DeepEqual(st.rw.HeaderMap, wantHeader) {
 339                 t.Errorf("Header+Trailer mismatch.\n got: %#v\nwant: %#v", st.rw.HeaderMap, wantHeader)
 340         }
 341 }
 342
 343 func TestHandlerTransport_HandleStreams_Timeout(t *testing.T) {
 344         bodyr, bodyw := io.Pipe()
 345         req := &http.Request{
 346                 ProtoMajor: 2,
 347                 Method:     "POST",
 348                 Header: http.Header{
 349                         "Content-Type": {"application/grpc"},
 350                         "Grpc-Timeout": {"200m"},
 351                 },
 352                 URL: &url.URL{
 353                         Path: "/service/foo.bar",
 354                 },
 355                 RequestURI: "/service/foo.bar",
 356                 Body:       bodyr,
 357         }
 358         rw := newTestHandlerResponseWriter().(testHandlerResponseWriter)
 359         ht, err := NewServerHandlerTransport(rw, req)
 360         if err != nil {
 361                 t.Fatal(err)
 362         }
 363         runStream := func(s *Stream) {
 364                 defer bodyw.Close()
 365                 select {
 366                 case <-s.ctx.Done():
 367                 case <-time.After(5 * time.Second):
 368                         t.Errorf("timeout waiting for ctx.Done")
 369                         return
 370                 }
 371                 err := s.ctx.Err()
 372                 if err != context.DeadlineExceeded {
 373                         t.Errorf("ctx.Err = %v; want %v", err, context.DeadlineExceeded)
 374                         return
 375                 }
 376                 ht.WriteStatus(s, status.New(codes.DeadlineExceeded, "too slow"))
 377         }
 378         ht.HandleStreams(
 379                 func(s *Stream) { go runStream(s) },
 380                 func(ctx context.Context, method string) context.Context { return ctx },
 381         )
 382         wantHeader := http.Header{
 383                 "Date":         nil,
 384                 "Content-Type": {"application/grpc"},
 385                 "Trailer":      {"Grpc-Status", "Grpc-Message", "Grpc-Status-Details-Bin"},
 386                 "Grpc-Status":  {"4"},
 387                 "Grpc-Message": {encodeGrpcMessage("too slow")},
 388         }
 389         if !reflect.DeepEqual(rw.HeaderMap, wantHeader) {
 390                 t.Errorf("Header+Trailer Map mismatch.\n got: %#v\nwant: %#v", rw.HeaderMap, wantHeader)
 391         }
 392 }
 393
 394 func TestHandlerTransport_HandleStreams_MultiWriteStatus(t *testing.T) {
 395         st := newHandleStreamTest(t)
 396         handleStream := func(s *Stream) {
 397                 if want := "/service/foo.bar"; s.method != want {
 398                         t.Errorf("stream method = %q; want %q", s.method, want)
 399                 }
 400                 st.bodyw.Close() // no body
 401
 402                 var wg sync.WaitGroup
 403                 wg.Add(5)
 404                 for i := 0; i < 5; i++ {
 405                         go func() {
 406                                 defer wg.Done()
 407                                 st.ht.WriteStatus(s, status.New(codes.OK, ""))
 408                         }()
 409                 }
 410                 wg.Wait()
 411         }
 412         st.ht.HandleStreams(
 413                 func(s *Stream) { go handleStream(s) },
 414                 func(ctx context.Context, method string) context.Context { return ctx },
 415         )
 416 }
 417
 418 func TestHandlerTransport_HandleStreams_ErrDetails(t *testing.T) {
 419         errDetails := []proto.Message{
 420                 &epb.RetryInfo{
 421                         RetryDelay: &dpb.Duration{Seconds: 60},
 422                 },
 423                 &epb.ResourceInfo{
 424                         ResourceType: "foo bar",
 425                         ResourceName: "service.foo.bar",
 426                         Owner:        "User",
 427                 },
 428         }
 429
 430         statusCode := codes.ResourceExhausted
 431         msg := "you are being throttled"
 432         st, err := status.New(statusCode, msg).WithDetails(errDetails...)
 433         if err != nil {
 434                 t.Fatal(err)
 435         }
 436
 437         stBytes, err := proto.Marshal(st.Proto())
 438         if err != nil {
 439                 t.Fatal(err)
 440         }
 441
 442         hst := newHandleStreamTest(t)
 443         handleStream := func(s *Stream) {
 444                 hst.ht.WriteStatus(s, st)
 445         }
 446         hst.ht.HandleStreams(
 447                 func(s *Stream) { go handleStream(s) },
 448                 func(ctx context.Context, method string) context.Context { return ctx },
 449         )
 450         wantHeader := http.Header{
 451                 "Date":                    nil,
 452                 "Content-Type":            {"application/grpc"},
 453                 "Trailer":                 {"Grpc-Status", "Grpc-Message", "Grpc-Status-Details-Bin"},
 454                 "Grpc-Status":             {fmt.Sprint(uint32(statusCode))},
 455                 "Grpc-Message":            {encodeGrpcMessage(msg)},
 456                 "Grpc-Status-Details-Bin": {encodeBinHeader(stBytes)},
 457         }
 458
 459         if !reflect.DeepEqual(hst.rw.HeaderMap, wantHeader) {
 460                 t.Errorf("Header+Trailer mismatch.\n got: %#v\nwant: %#v", hst.rw.HeaderMap, wantHeader)
 461         }
 462 }