1// Copyright 2016 The etcd Authors 2// 3// Licensed under the Apache License, Version 2.0 (the "License"); 4// you may not use this file except in compliance with the License. 5// You may obtain a copy of the License at 6// 7// http://www.apache.org/licenses/LICENSE-2.0 8// 9// Unless required by applicable law or agreed to in writing, software 10// distributed under the License is distributed on an "AS IS" BASIS, 11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 12// See the License for the specific language governing permissions and 13// limitations under the License. 14 15package clientv3 16 17import ( 18 "context" 19 "fmt" 20 "go.uber.org/zap" 21 "net" 22 "testing" 23 "time" 24 25 "go.etcd.io/etcd/api/v3/v3rpc/rpctypes" 26 "go.etcd.io/etcd/client/pkg/v3/testutil" 27 "go.uber.org/zap/zaptest" 28 29 "google.golang.org/grpc" 30) 31 32func NewClient(t *testing.T, cfg Config) (*Client, error) { 33 cfg.Logger = zaptest.NewLogger(t) 34 return New(cfg) 35} 36 37func TestDialCancel(t *testing.T) { 38 testutil.RegisterLeakDetection(t) 39 40 // accept first connection so client is created with dial timeout 41 ln, err := net.Listen("unix", "dialcancel:12345") 42 if err != nil { 43 t.Fatal(err) 44 } 45 defer ln.Close() 46 47 ep := "unix://dialcancel:12345" 48 cfg := Config{ 49 Endpoints: []string{ep}, 50 DialTimeout: 30 * time.Second} 51 c, err := NewClient(t, cfg) 52 if err != nil { 53 t.Fatal(err) 54 } 55 56 // connect to ipv4 black hole so dial blocks 57 c.SetEndpoints("http://254.0.0.1:12345") 58 59 // issue Get to force redial attempts 60 getc := make(chan struct{}) 61 go func() { 62 defer close(getc) 63 // Get may hang forever on grpc's Stream.Header() if its 64 // context is never canceled. 65 c.Get(c.Ctx(), "abc") 66 }() 67 68 // wait a little bit so client close is after dial starts 69 time.Sleep(100 * time.Millisecond) 70 71 donec := make(chan struct{}) 72 go func() { 73 defer close(donec) 74 c.Close() 75 }() 76 77 select { 78 case <-time.After(5 * time.Second): 79 t.Fatalf("failed to close") 80 case <-donec: 81 } 82 select { 83 case <-time.After(5 * time.Second): 84 t.Fatalf("get failed to exit") 85 case <-getc: 86 } 87} 88 89func TestDialTimeout(t *testing.T) { 90 testutil.RegisterLeakDetection(t) 91 92 wantError := context.DeadlineExceeded 93 94 // grpc.WithBlock to block until connection up or timeout 95 testCfgs := []Config{ 96 { 97 Endpoints: []string{"http://254.0.0.1:12345"}, 98 DialTimeout: 2 * time.Second, 99 DialOptions: []grpc.DialOption{grpc.WithBlock()}, 100 }, 101 { 102 Endpoints: []string{"http://254.0.0.1:12345"}, 103 DialTimeout: time.Second, 104 DialOptions: []grpc.DialOption{grpc.WithBlock()}, 105 Username: "abc", 106 Password: "def", 107 }, 108 } 109 110 for i, cfg := range testCfgs { 111 donec := make(chan error, 1) 112 go func(cfg Config) { 113 // without timeout, dial continues forever on ipv4 black hole 114 c, err := NewClient(t, cfg) 115 if c != nil || err == nil { 116 t.Errorf("#%d: new client should fail", i) 117 } 118 donec <- err 119 }(cfg) 120 121 time.Sleep(10 * time.Millisecond) 122 123 select { 124 case err := <-donec: 125 t.Errorf("#%d: dial didn't wait (%v)", i, err) 126 default: 127 } 128 129 select { 130 case <-time.After(5 * time.Second): 131 t.Errorf("#%d: failed to timeout dial on time", i) 132 case err := <-donec: 133 if err.Error() != wantError.Error() { 134 t.Errorf("#%d: unexpected error '%v', want '%v'", i, err, wantError) 135 } 136 } 137 } 138} 139 140func TestDialNoTimeout(t *testing.T) { 141 cfg := Config{Endpoints: []string{"127.0.0.1:12345"}} 142 c, err := NewClient(t, cfg) 143 if c == nil || err != nil { 144 t.Fatalf("new client with DialNoWait should succeed, got %v", err) 145 } 146 c.Close() 147} 148 149func TestIsHaltErr(t *testing.T) { 150 if !isHaltErr(context.TODO(), fmt.Errorf("etcdserver: some etcdserver error")) { 151 t.Errorf(`error prefixed with "etcdserver: " should be Halted by default`) 152 } 153 if isHaltErr(context.TODO(), rpctypes.ErrGRPCStopped) { 154 t.Errorf("error %v should not halt", rpctypes.ErrGRPCStopped) 155 } 156 if isHaltErr(context.TODO(), rpctypes.ErrGRPCNoLeader) { 157 t.Errorf("error %v should not halt", rpctypes.ErrGRPCNoLeader) 158 } 159 ctx, cancel := context.WithCancel(context.TODO()) 160 if isHaltErr(ctx, nil) { 161 t.Errorf("no error and active context should not be Halted") 162 } 163 cancel() 164 if !isHaltErr(ctx, nil) { 165 t.Errorf("cancel on context should be Halted") 166 } 167} 168 169func TestCloseCtxClient(t *testing.T) { 170 ctx := context.Background() 171 c := NewCtxClient(ctx) 172 err := c.Close() 173 // Close returns ctx.toErr, a nil error means an open Done channel 174 if err == nil { 175 t.Errorf("failed to Close the client. %v", err) 176 } 177} 178 179func TestWithLogger(t *testing.T) { 180 ctx := context.Background() 181 c := NewCtxClient(ctx) 182 if c.lg == nil { 183 t.Errorf("unexpected nil in *zap.Logger") 184 } 185 186 c.WithLogger(nil) 187 if c.lg != nil { 188 t.Errorf("WithLogger should modify *zap.Logger") 189 } 190} 191 192func TestZapWithLogger(t *testing.T) { 193 ctx := context.Background() 194 lg := zap.NewNop() 195 c := NewCtxClient(ctx, WithZapLogger(lg)) 196 197 if c.lg != lg { 198 t.Errorf("WithZapLogger should modify *zap.Logger") 199 } 200} 201