cubefs

1
// Copyright 2022 The CubeFS 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
12
// implied. See the License for the specific language governing
13
// permissions and limitations under the License.
14

15
package rpc
16

17
import (
18
	"context"
19
	"encoding/json"
20
	"io/ioutil"
21
	"net/http"
22
	"net/http/httptest"
23
	"testing"
24
	"time"
25

26
	"github.com/stretchr/testify/require"
27

28
	"github.com/cubefs/cubefs/blobstore/common/crc32block"
29
	"github.com/cubefs/cubefs/blobstore/common/rpc/auth"
30
)
31

32
var (
33
	testServer *httptest.Server
34
	first      = true
35
)
36

37
type handler struct{}
38

39
func (s *handler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
40
	uri := r.URL.RequestURI()
41
	switch uri {
42
	case "/retry":
43
		b, err := ioutil.ReadAll(r.Body)
44
		if err != nil || int64(len(b)) != r.ContentLength || first {
45
			ReplyErr(w, 500, "test retry")
46
			first = false
47
		}
48
	case "/crc":
49
		doAfterCrc(w, r)
50
	case "/crcbody":
51
		w.Header().Set(HeaderAckCrcEncoded, "1")
52
		w.WriteHeader(500)
53
	case "/json":
54
		callWithJSON(w, r)
55
	case "/timeout":
56
		timeout(w, r)
57
	case "/notfound":
58
		ReplyWith(w, 404, "", []byte("404 page not found"))
59
	default:
60
		{
61
			marshal, err := json.Marshal(ret{Name: "Test_GetWith"})
62
			if err == nil {
63
				w.Write(marshal)
64
				return
65
			}
66
		}
67
	}
68
}
69

70
type testTimeoutReader struct{}
71

72
func (t *testTimeoutReader) Read(p []byte) (n int, err error) {
73
	time.Sleep(time.Millisecond * 20)
74
	return 0, nil
75
}
76

77
func doAfterCrc(w http.ResponseWriter, req *http.Request) {
78
	dec := crc32block.NewBodyDecoder(req.Body)
79
	defer dec.Close()
80
	dataByte := make([]byte, 1024)
81
	read, err := dec.Read(dataByte)
82
	w.Header().Set(HeaderAckCrcEncoded, "1")
83
	if err != nil {
84
		return
85
	}
86
	w.Write(dataByte[:read])
87
}
88

89
func callWithJSON(w http.ResponseWriter, req *http.Request) {
90
	r := &ret{}
91
	err := json.NewDecoder(req.Body).Decode(r)
92
	if err != nil {
93
		w.WriteHeader(http.StatusBadRequest)
94
		return
95
	}
96
	r.Name = r.Name + "+Test"
97
	marshal, err := json.Marshal(r)
98
	if err != nil {
99
		w.WriteHeader(http.StatusGatewayTimeout)
100
		return
101
	}
102
	w.Write(marshal)
103
}
104

105
func timeout(w http.ResponseWriter, req *http.Request) {
106
	r := &ret{}
107
	err := json.NewDecoder(req.Body).Decode(r)
108
	if err != nil {
109
		return
110
	}
111
	for i := 0; i < 10*1024*1024*1024; i++ {
112
		w.Write([]byte("time out"))
113
	}
114
}
115

116
var simpleCfg = &Config{
117
	ClientTimeoutMs: 10000,
118
	Tc: TransportConfig{
119
		DialTimeoutMs:           1000,
120
		ResponseHeaderTimeoutMs: 3000,
121
		MaxConnsPerHost:         100,
122
		MaxIdleConns:            1000,
123
		MaxIdleConnsPerHost:     10,
124
		IdleConnTimeoutMs:       60000,
125
		DisableCompression:      true,
126
		Auth: auth.Config{
127
			EnableAuth: true,
128
			Secret:     "test",
129
		},
130
	},
131
}
132
var simpleClient = NewClient(simpleCfg)
133

134
func init() {
135
	testServer = httptest.NewServer(&handler{})
136
}
137

138
func TestClient_NewClient(t *testing.T) {
139
	NewClient(nil)
140
	{
141
		cfg := Config{}
142
		NewClient(&cfg)
143
		require.Equal(t, 10, cfg.Tc.MaxConnsPerHost)
144
		require.Equal(t, int64(0), cfg.Tc.DialTimeoutMs)
145
		require.False(t, cfg.Tc.Auth.EnableAuth)
146
	}
147
	{
148
		cfg := Config{}
149
		cfg.Tc.Auth.EnableAuth = true
150
		cfg.Tc.Auth.Secret = "true"
151
		NewClient(&cfg)
152
		require.Equal(t, 10, cfg.Tc.MaxConnsPerHost)
153
		require.Equal(t, int64(0), cfg.Tc.DialTimeoutMs)
154
		require.True(t, cfg.Tc.Auth.EnableAuth)
155
		require.Equal(t, "true", cfg.Tc.Auth.Secret)
156
	}
157
	{
158
		cfg := Config{}
159
		cfg.Tc.MaxConnsPerHost = 4
160
		NewClient(&cfg)
161
		require.Equal(t, 4, cfg.Tc.MaxConnsPerHost)
162
		require.Equal(t, 0, cfg.Tc.MaxIdleConnsPerHost)
163
		require.Equal(t, int64(0), cfg.Tc.DialTimeoutMs)
164
		require.False(t, cfg.Tc.Auth.EnableAuth)
165
	}
166
}
167

168
func TestClient_GetWith(t *testing.T) {
169
	ctx := context.Background()
170
	result := &ret{}
171
	err := simpleClient.GetWith(ctx, testServer.URL, result)
172
	require.NoError(t, err)
173
	require.NotNil(t, result)
174
}
175

176
func TestClient_PostWithCrc(t *testing.T) {
177
	ctx := context.Background()
178
	result := &ret{}
179
	err := simpleClient.PostWith(ctx, testServer.URL+"/crc", result,
180
		&ret{Name: "TestClient_PostWithCrc"}, WithCrcEncode())
181
	require.NoError(t, err)
182
	require.NotNil(t, result)
183
	require.Equal(t, "TestClient_PostWithCrc", result.Name)
184
}
185

186
func TestClient_PostWithNoCrc(t *testing.T) {
187
	ctx := context.Background()
188
	result := &ret{}
189
	err := simpleClient.PostWith(ctx, testServer.URL+"/json", result, &ret{Name: "TestClient_PostWithNoCrc"})
190
	require.NoError(t, err)
191
	require.NotNil(t, result)
192
	require.Equal(t, "TestClient_PostWithNoCrc+Test", result.Name)
193
}
194

195
func TestClient_PostWithNoneBody(t *testing.T) {
196
	ctx := context.Background()
197
	err := simpleClient.PostWith(ctx, testServer.URL+"/json", nil, NoneBody)
198
	require.Equal(t, DetectStatusCode(err), http.StatusBadRequest)
199
}
200

201
func TestClient_Delete(t *testing.T) {
202
	ctx := context.Background()
203
	resp, err := simpleClient.Delete(ctx, testServer.URL)
204
	require.NoError(t, err)
205
	resp.Body.Close()
206
	require.Equal(t, http.StatusOK, resp.StatusCode)
207
}
208

209
func TestClient_PutWithNoCrc(t *testing.T) {
210
	ctx := context.Background()
211
	result := &ret{}
212
	err := simpleClient.PutWith(ctx, testServer.URL+"/json", result, &ret{Name: "TestClient_PutWithNoCrc"})
213
	require.NoError(t, err)
214
	require.NotNil(t, result)
215
	require.Equal(t, "TestClient_PutWithNoCrc+Test", result.Name)
216
}
217

218
func TestClient_PutWithCrc(t *testing.T) {
219
	ctx := context.Background()
220
	result := &ret{}
221
	err := simpleClient.PostWith(ctx, testServer.URL+"/crc", result,
222
		&ret{Name: "TestClient_PutWithCrc"}, WithCrcEncode())
223
	require.NoError(t, err)
224
	require.NotNil(t, result)
225
	require.Equal(t, "TestClient_PutWithCrc", result.Name)
226
}
227

228
func TestClient_Post(t *testing.T) {
229
	ctx := context.Background()
230
	result := &ret{}
231
	resp, err := simpleClient.Post(ctx, testServer.URL+"/json", &ret{Name: "TestClient_Post"})
232
	require.NoError(t, err)
233
	err = parseData(resp, result)
234
	require.NoError(t, err)
235
	require.NotNil(t, result)
236
	require.Equal(t, "TestClient_Post+Test", result.Name)
237
}
238

239
func TestClient_PostWithReadResponseTimeout(t *testing.T) {
240
	simpleCfg.BodyBaseTimeoutMs = 50
241
	simpleCfg.BodyBandwidthMBPs = 1
242
	simpleCli := NewClient(simpleCfg)
243
	ctx := context.Background()
244
	resp, err := simpleCli.Post(ctx, testServer.URL+"/timeout", &ret{Name: "TestClient_Post"})
245
	require.NoError(t, err)
246
	defer resp.Body.Close()
247
	cache := make([]byte, 128)
248
	_, err = resp.Body.Read(cache)
249
	for err == nil {
250
		_, err = resp.Body.Read(cache)
251
	}
252
	require.Equal(t, ErrBodyReadTimeout, err)
253
}
254

255
func TestTimeoutReadCloser_Read(t *testing.T) {
256
	// test for read data for input buffer timeout
257
	readCloser := timeoutReadCloser{
258
		timeoutMs: 1,
259
		body:      ioutil.NopCloser(&testTimeoutReader{}),
260
	}
261
	res := make([]byte, 30)
262
	_, err := readCloser.Read(res)
263
	require.Equal(t, ErrBodyReadTimeout, err)
264
}
265

266
func TestClient_Put(t *testing.T) {
267
	ctx := context.Background()
268
	result := &ret{}
269
	resp, err := simpleClient.Put(ctx, testServer.URL+"/json", &ret{Name: "TestClient_Put"})
270
	require.NoError(t, err)
271
	err = parseData(resp, result)
272
	require.NoError(t, err)
273
	require.NotNil(t, result)
274
	require.Equal(t, "TestClient_Put+Test", result.Name)
275
}
276

277
func TestClient_Close(t *testing.T) {
278
	simpleClient.Close()
279
}
280

281
func TestClient_Head(t *testing.T) {
282
	ctx := context.Background()
283
	resp, err := simpleClient.Head(ctx, testServer.URL)
284
	require.NoError(t, err)
285
	resp.Body.Close()
286
	require.Equal(t, http.StatusOK, resp.StatusCode)
287
}
288

289
func TestClient_DoWithNoCrc(t *testing.T) {
290
	ctx := context.Background()
291
	result := &ret{}
292
	request, err := http.NewRequest(http.MethodPost, testServer.URL, nil)
293
	require.NoError(t, err)
294
	err = simpleClient.DoWith(ctx, request, result)
295
	require.NoError(t, err)
296
	require.NotNil(t, result)
297
}
298

299
func TestClient_DoWithCrc(t *testing.T) {
300
	ctx := context.Background()
301
	result := &ret{}
302
	request, err := http.NewRequest(http.MethodPost, testServer.URL+"/crc", nil)
303
	require.NoError(t, err)
304
	err = simpleClient.DoWith(ctx, request, result, WithCrcEncode())
305
	require.NoError(t, err)
306
	require.NotNil(t, result)
307
}
308

309
func TestClient_Form(t *testing.T) {
310
	m := make(map[string][]string)
311
	m["test"] = []string{"test_lb_Form"}
312
	ctx := context.Background()
313
	resp, err := simpleClient.Form(ctx, http.MethodPost, testServer.URL, m)
314
	require.NoError(t, err)
315
	resp.Body.Close()
316
	require.Equal(t, http.StatusOK, resp.StatusCode)
317
}
318

319
func TestClient_ParseDataWithContentLengthZero(t *testing.T) {
320
	m := make(map[string][]string)
321
	m["test"] = []string{"yest_lb_Form"}
322
	ctx := context.Background()
323
	result := &ret{}
324
	resp, err := simpleClient.Form(ctx, http.MethodPost, testServer.URL, m)
325
	require.NoError(t, err)
326
	resp.StatusCode = 400
327
	resp.ContentLength = 0
328
	err = parseData(resp, result)
329
	require.Error(t, err)
330
}
331

332
func TestClient_ParseData(t *testing.T) {
333
	m := make(map[string][]string)
334
	m["test"] = []string{"yest_lb_Form"}
335
	ctx := context.Background()
336
	result := &ret{}
337
	resp, err := simpleClient.Form(ctx, http.MethodPost, testServer.URL, m)
338
	require.NoError(t, err)
339
	resp.StatusCode = 400
340
	resp.ContentLength = 12
341
	err = parseData(resp, result)
342
	require.Error(t, err)
343
}
344

345
func TestClient_DoContextCancel(t *testing.T) {
346
	ctx := context.Background()
347
	result := &ret{}
348
	cancel, cancelFunc := context.WithCancel(ctx)
349
	request, err := http.NewRequest(http.MethodPost, testServer.URL, nil)
350
	require.NoError(t, err)
351
	cancelFunc()
352
	err = simpleClient.DoWith(cancel, request, result)
353
	require.Error(t, err)
354
}
355

356
func TestClient_ResponseClose(t *testing.T) {
357
	ctx := context.Background()
358
	result := &ret{}
359
	resp, err := simpleClient.Post(ctx, testServer.URL+"/json",
360
		&ret{Name: "TestClient_ResponseClose"})
361
	require.NoError(t, err)
362
	resp.Body.Close()
363
	err = parseData(resp, result)
364
	require.Error(t, err)
365
}
366

367
func TestClient_PostWithOnServerNotAck(t *testing.T) {
368
	request, err := http.NewRequest(http.MethodPost, testServer.URL, nil)
369
	require.NoError(t, err)
370
	request.Header.Set(HeaderCrcEncoded, "1")
371
	response, err := simpleClient.Do(context.Background(), request)
372
	require.NoError(t, err)
373
	err = serverCrcEncodeCheck(context.Background(), request, response)
374
	response.Body.Close()
375
	require.NotNil(t, err)
376
}
377

378
func TestClient_PageNotFound(t *testing.T) {
379
	ctx := context.Background()
380
	result := &ret{}
381
	err := simpleClient.GetWith(ctx, testServer.URL+"/notfound", result)
382
	require.Error(t, err)
383
	require.Equal(t, "", result.Name)
384
}
385