vendor/google.golang.org/grpc/clientconn_test.go

   1 /*
   2  *
   3  * Copyright 2014 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 grpc
  20
  21 import (
  22         "math"
  23         "net"
  24         "testing"
  25         "time"
  26
  27         "golang.org/x/net/context"
  28
  29         "google.golang.org/grpc/connectivity"
  30         "google.golang.org/grpc/credentials"
  31         "google.golang.org/grpc/keepalive"
  32         "google.golang.org/grpc/naming"
  33         _ "google.golang.org/grpc/resolver/passthrough"
  34         "google.golang.org/grpc/test/leakcheck"
  35         "google.golang.org/grpc/testdata"
  36 )
  37
  38 func assertState(wantState connectivity.State, cc *ClientConn) (connectivity.State, bool) {
  39         ctx, cancel := context.WithTimeout(context.Background(), time.Second)
  40         defer cancel()
  41         var state connectivity.State
  42         for state = cc.GetState(); state != wantState && cc.WaitForStateChange(ctx, state); state = cc.GetState() {
  43         }
  44         return state, state == wantState
  45 }
  46
  47 func TestConnectivityStates(t *testing.T) {
  48         defer leakcheck.Check(t)
  49         servers, resolver, cleanup := startServers(t, 2, math.MaxUint32)
  50         defer cleanup()
  51         cc, err := Dial("passthrough:///foo.bar.com", WithBalancer(RoundRobin(resolver)), WithInsecure())
  52         if err != nil {
  53                 t.Fatalf("Dial(\"foo.bar.com\", WithBalancer(_)) = _, %v, want _ <nil>", err)
  54         }
  55         defer cc.Close()
  56         wantState := connectivity.Ready
  57         if state, ok := assertState(wantState, cc); !ok {
  58                 t.Fatalf("asserState(%s) = %s, false, want %s, true", wantState, state, wantState)
  59         }
  60         // Send an update to delete the server connection (tearDown addrConn).
  61         update := []*naming.Update{
  62                 {
  63                         Op:   naming.Delete,
  64                         Addr: "localhost:" + servers[0].port,
  65                 },
  66         }
  67         resolver.w.inject(update)
  68         wantState = connectivity.TransientFailure
  69         if state, ok := assertState(wantState, cc); !ok {
  70                 t.Fatalf("asserState(%s) = %s, false, want %s, true", wantState, state, wantState)
  71         }
  72         update[0] = &naming.Update{
  73                 Op:   naming.Add,
  74                 Addr: "localhost:" + servers[1].port,
  75         }
  76         resolver.w.inject(update)
  77         wantState = connectivity.Ready
  78         if state, ok := assertState(wantState, cc); !ok {
  79                 t.Fatalf("asserState(%s) = %s, false, want %s, true", wantState, state, wantState)
  80         }
  81
  82 }
  83
  84 func TestDialTimeout(t *testing.T) {
  85         defer leakcheck.Check(t)
  86         conn, err := Dial("passthrough:///Non-Existent.Server:80", WithTimeout(time.Millisecond), WithBlock(), WithInsecure())
  87         if err == nil {
  88                 conn.Close()
  89         }
  90         if err != context.DeadlineExceeded {
  91                 t.Fatalf("Dial(_, _) = %v, %v, want %v", conn, err, context.DeadlineExceeded)
  92         }
  93 }
  94
  95 func TestTLSDialTimeout(t *testing.T) {
  96         defer leakcheck.Check(t)
  97         creds, err := credentials.NewClientTLSFromFile(testdata.Path("ca.pem"), "x.test.youtube.com")
  98         if err != nil {
  99                 t.Fatalf("Failed to create credentials %v", err)
 100         }
 101         conn, err := Dial("passthrough:///Non-Existent.Server:80", WithTransportCredentials(creds), WithTimeout(time.Millisecond), WithBlock())
 102         if err == nil {
 103                 conn.Close()
 104         }
 105         if err != context.DeadlineExceeded {
 106                 t.Fatalf("Dial(_, _) = %v, %v, want %v", conn, err, context.DeadlineExceeded)
 107         }
 108 }
 109
 110 func TestDefaultAuthority(t *testing.T) {
 111         defer leakcheck.Check(t)
 112         target := "Non-Existent.Server:8080"
 113         conn, err := Dial(target, WithInsecure())
 114         if err != nil {
 115                 t.Fatalf("Dial(_, _) = _, %v, want _, <nil>", err)
 116         }
 117         defer conn.Close()
 118         if conn.authority != target {
 119                 t.Fatalf("%v.authority = %v, want %v", conn, conn.authority, target)
 120         }
 121 }
 122
 123 func TestTLSServerNameOverwrite(t *testing.T) {
 124         defer leakcheck.Check(t)
 125         overwriteServerName := "over.write.server.name"
 126         creds, err := credentials.NewClientTLSFromFile(testdata.Path("ca.pem"), overwriteServerName)
 127         if err != nil {
 128                 t.Fatalf("Failed to create credentials %v", err)
 129         }
 130         conn, err := Dial("passthrough:///Non-Existent.Server:80", WithTransportCredentials(creds))
 131         if err != nil {
 132                 t.Fatalf("Dial(_, _) = _, %v, want _, <nil>", err)
 133         }
 134         defer conn.Close()
 135         if conn.authority != overwriteServerName {
 136                 t.Fatalf("%v.authority = %v, want %v", conn, conn.authority, overwriteServerName)
 137         }
 138 }
 139
 140 func TestWithAuthority(t *testing.T) {
 141         defer leakcheck.Check(t)
 142         overwriteServerName := "over.write.server.name"
 143         conn, err := Dial("passthrough:///Non-Existent.Server:80", WithInsecure(), WithAuthority(overwriteServerName))
 144         if err != nil {
 145                 t.Fatalf("Dial(_, _) = _, %v, want _, <nil>", err)
 146         }
 147         defer conn.Close()
 148         if conn.authority != overwriteServerName {
 149                 t.Fatalf("%v.authority = %v, want %v", conn, conn.authority, overwriteServerName)
 150         }
 151 }
 152
 153 func TestWithAuthorityAndTLS(t *testing.T) {
 154         defer leakcheck.Check(t)
 155         overwriteServerName := "over.write.server.name"
 156         creds, err := credentials.NewClientTLSFromFile(testdata.Path("ca.pem"), overwriteServerName)
 157         if err != nil {
 158                 t.Fatalf("Failed to create credentials %v", err)
 159         }
 160         conn, err := Dial("passthrough:///Non-Existent.Server:80", WithTransportCredentials(creds), WithAuthority("no.effect.authority"))
 161         if err != nil {
 162                 t.Fatalf("Dial(_, _) = _, %v, want _, <nil>", err)
 163         }
 164         defer conn.Close()
 165         if conn.authority != overwriteServerName {
 166                 t.Fatalf("%v.authority = %v, want %v", conn, conn.authority, overwriteServerName)
 167         }
 168 }
 169
 170 func TestDialContextCancel(t *testing.T) {
 171         defer leakcheck.Check(t)
 172         ctx, cancel := context.WithCancel(context.Background())
 173         cancel()
 174         if _, err := DialContext(ctx, "Non-Existent.Server:80", WithBlock(), WithInsecure()); err != context.Canceled {
 175                 t.Fatalf("DialContext(%v, _) = _, %v, want _, %v", ctx, err, context.Canceled)
 176         }
 177 }
 178
 179 // blockingBalancer mimics the behavior of balancers whose initialization takes a long time.
 180 // In this test, reading from blockingBalancer.Notify() blocks forever.
 181 type blockingBalancer struct {
 182         ch chan []Address
 183 }
 184
 185 func newBlockingBalancer() Balancer {
 186         return &blockingBalancer{ch: make(chan []Address)}
 187 }
 188 func (b *blockingBalancer) Start(target string, config BalancerConfig) error {
 189         return nil
 190 }
 191 func (b *blockingBalancer) Up(addr Address) func(error) {
 192         return nil
 193 }
 194 func (b *blockingBalancer) Get(ctx context.Context, opts BalancerGetOptions) (addr Address, put func(), err error) {
 195         return Address{}, nil, nil
 196 }
 197 func (b *blockingBalancer) Notify() <-chan []Address {
 198         return b.ch
 199 }
 200 func (b *blockingBalancer) Close() error {
 201         close(b.ch)
 202         return nil
 203 }
 204
 205 func TestDialWithBlockingBalancer(t *testing.T) {
 206         defer leakcheck.Check(t)
 207         ctx, cancel := context.WithCancel(context.Background())
 208         dialDone := make(chan struct{})
 209         go func() {
 210                 DialContext(ctx, "Non-Existent.Server:80", WithBlock(), WithInsecure(), WithBalancer(newBlockingBalancer()))
 211                 close(dialDone)
 212         }()
 213         cancel()
 214         <-dialDone
 215 }
 216
 217 // securePerRPCCredentials always requires transport security.
 218 type securePerRPCCredentials struct{}
 219
 220 func (c securePerRPCCredentials) GetRequestMetadata(ctx context.Context, uri ...string) (map[string]string, error) {
 221         return nil, nil
 222 }
 223
 224 func (c securePerRPCCredentials) RequireTransportSecurity() bool {
 225         return true
 226 }
 227
 228 func TestCredentialsMisuse(t *testing.T) {
 229         defer leakcheck.Check(t)
 230         tlsCreds, err := credentials.NewClientTLSFromFile(testdata.Path("ca.pem"), "x.test.youtube.com")
 231         if err != nil {
 232                 t.Fatalf("Failed to create authenticator %v", err)
 233         }
 234         // Two conflicting credential configurations
 235         if _, err := Dial("passthrough:///Non-Existent.Server:80", WithTransportCredentials(tlsCreds), WithBlock(), WithInsecure()); err != errCredentialsConflict {
 236                 t.Fatalf("Dial(_, _) = _, %v, want _, %v", err, errCredentialsConflict)
 237         }
 238         // security info on insecure connection
 239         if _, err := Dial("passthrough:///Non-Existent.Server:80", WithPerRPCCredentials(securePerRPCCredentials{}), WithBlock(), WithInsecure()); err != errTransportCredentialsMissing {
 240                 t.Fatalf("Dial(_, _) = _, %v, want _, %v", err, errTransportCredentialsMissing)
 241         }
 242 }
 243
 244 func TestWithBackoffConfigDefault(t *testing.T) {
 245         defer leakcheck.Check(t)
 246         testBackoffConfigSet(t, &DefaultBackoffConfig)
 247 }
 248
 249 func TestWithBackoffConfig(t *testing.T) {
 250         defer leakcheck.Check(t)
 251         b := BackoffConfig{MaxDelay: DefaultBackoffConfig.MaxDelay / 2}
 252         expected := b
 253         setDefaults(&expected) // defaults should be set
 254         testBackoffConfigSet(t, &expected, WithBackoffConfig(b))
 255 }
 256
 257 func TestWithBackoffMaxDelay(t *testing.T) {
 258         defer leakcheck.Check(t)
 259         md := DefaultBackoffConfig.MaxDelay / 2
 260         expected := BackoffConfig{MaxDelay: md}
 261         setDefaults(&expected)
 262         testBackoffConfigSet(t, &expected, WithBackoffMaxDelay(md))
 263 }
 264
 265 func testBackoffConfigSet(t *testing.T, expected *BackoffConfig, opts ...DialOption) {
 266         opts = append(opts, WithInsecure())
 267         conn, err := Dial("passthrough:///foo:80", opts...)
 268         if err != nil {
 269                 t.Fatalf("unexpected error dialing connection: %v", err)
 270         }
 271         defer conn.Close()
 272
 273         if conn.dopts.bs == nil {
 274                 t.Fatalf("backoff config not set")
 275         }
 276
 277         actual, ok := conn.dopts.bs.(BackoffConfig)
 278         if !ok {
 279                 t.Fatalf("unexpected type of backoff config: %#v", conn.dopts.bs)
 280         }
 281
 282         if actual != *expected {
 283                 t.Fatalf("unexpected backoff config on connection: %v, want %v", actual, expected)
 284         }
 285 }
 286
 287 // emptyBalancer returns an empty set of servers.
 288 type emptyBalancer struct {
 289         ch chan []Address
 290 }
 291
 292 func newEmptyBalancer() Balancer {
 293         return &emptyBalancer{ch: make(chan []Address, 1)}
 294 }
 295 func (b *emptyBalancer) Start(_ string, _ BalancerConfig) error {
 296         b.ch <- nil
 297         return nil
 298 }
 299 func (b *emptyBalancer) Up(_ Address) func(error) {
 300         return nil
 301 }
 302 func (b *emptyBalancer) Get(_ context.Context, _ BalancerGetOptions) (Address, func(), error) {
 303         return Address{}, nil, nil
 304 }
 305 func (b *emptyBalancer) Notify() <-chan []Address {
 306         return b.ch
 307 }
 308 func (b *emptyBalancer) Close() error {
 309         close(b.ch)
 310         return nil
 311 }
 312
 313 func TestNonblockingDialWithEmptyBalancer(t *testing.T) {
 314         defer leakcheck.Check(t)
 315         ctx, cancel := context.WithCancel(context.Background())
 316         defer cancel()
 317         dialDone := make(chan error)
 318         go func() {
 319                 dialDone <- func() error {
 320                         conn, err := DialContext(ctx, "Non-Existent.Server:80", WithInsecure(), WithBalancer(newEmptyBalancer()))
 321                         if err != nil {
 322                                 return err
 323                         }
 324                         return conn.Close()
 325                 }()
 326         }()
 327         if err := <-dialDone; err != nil {
 328                 t.Fatalf("unexpected error dialing connection: %s", err)
 329         }
 330 }
 331
 332 func TestClientUpdatesParamsAfterGoAway(t *testing.T) {
 333         defer leakcheck.Check(t)
 334         lis, err := net.Listen("tcp", "localhost:0")
 335         if err != nil {
 336                 t.Fatalf("Failed to listen. Err: %v", err)
 337         }
 338         defer lis.Close()
 339         addr := lis.Addr().String()
 340         s := NewServer()
 341         go s.Serve(lis)
 342         defer s.Stop()
 343         cc, err := Dial(addr, WithBlock(), WithInsecure(), WithKeepaliveParams(keepalive.ClientParameters{
 344                 Time:                50 * time.Millisecond,
 345                 Timeout:             1 * time.Millisecond,
 346                 PermitWithoutStream: true,
 347         }))
 348         if err != nil {
 349                 t.Fatalf("Dial(%s, _) = _, %v, want _, <nil>", addr, err)
 350         }
 351         defer cc.Close()
 352         time.Sleep(1 * time.Second)
 353         cc.mu.RLock()
 354         defer cc.mu.RUnlock()
 355         v := cc.mkp.Time
 356         if v < 100*time.Millisecond {
 357                 t.Fatalf("cc.dopts.copts.Keepalive.Time = %v , want 100ms", v)
 358         }
 359 }