GitVita
New repositoryImport repositoryNew issue
Your repositories Your dashboard Sign in

cloudflare/cloudflared

Public

mirrored from https://github.com/cloudflare/cloudflaredAvailable

Watch0Fork0Star0
CodeCommitsIssuesPull requestsActionsInsightsSecurity
2021.3.2

Branches

Tags

  • No tags available.
0Branches0Tags
Go to file
Add file
Code

Clone

HTTPS

Download ZIP

cloudflared/connection

connection/http2_test.go

409lines · modeblame

RawDownload
Latest commit unavailable.
d5769519cthuang5 years ago1package connection
2
3import (
4"context"
5"fmt"
6"io"
7"io/ioutil"
8"net"
9"net/http"
10"net/http/httptest"
11"sync"
12"testing"
13"time"
14
d503aeafIgor Postelnik5 years ago15"github.com/stretchr/testify/assert"
16
d5769519cthuang5 years ago17"github.com/cloudflare/cloudflared/tunnelrpc/pogs"
18tunnelpogs "github.com/cloudflare/cloudflared/tunnelrpc/pogs"
870f5fa9Areg Harutyunyan5 years ago19
d5769519cthuang5 years ago20"github.com/gobwas/ws/wsutil"
870f5fa9Areg Harutyunyan5 years ago21"github.com/rs/zerolog"
d5769519cthuang5 years ago22"github.com/stretchr/testify/require"
23"golang.org/x/net/http2"
24)
25
26var (
27testTransport = http2.Transport{}
28)
29
30func newTestHTTP2Connection() (*http2Connection, net.Conn) {
31edgeConn, originConn := net.Pipe()
32var connIndex = uint8(0)
33return NewHTTP2Connection(
34originConn,
35testConfig,
36&NamedTunnelConfig{},
37&pogs.ConnectionOptions{},
cf562ef8Igor Postelnik5 years ago38NewObserver(&log, &log, false),
d5769519cthuang5 years ago39connIndex,
40mockConnectedFuse{},
d503aeafIgor Postelnik5 years ago41nil,
d5769519cthuang5 years ago42), edgeConn
43}
44
45func TestServeHTTP(t *testing.T) {
46tests := []testRequest{
47{
48name: "ok",
49endpoint: "ok",
50expectedStatus: http.StatusOK,
51expectedBody: []byte(http.StatusText(http.StatusOK)),
52},
53{
54name: "large_file",
55endpoint: "large_file",
56expectedStatus: http.StatusOK,
57expectedBody: testLargeResp,
58},
59{
60name: "Bad request",
61endpoint: "400",
62expectedStatus: http.StatusBadRequest,
63expectedBody: []byte(http.StatusText(http.StatusBadRequest)),
64},
65{
66name: "Internal server error",
67endpoint: "500",
68expectedStatus: http.StatusInternalServerError,
69expectedBody: []byte(http.StatusText(http.StatusInternalServerError)),
70},
71{
72name: "Proxy error",
73endpoint: "error",
74expectedStatus: http.StatusBadGateway,
75expectedBody: nil,
76isProxyError: true,
77},
78}
79
80http2Conn, edgeConn := newTestHTTP2Connection()
81
82ctx, cancel := context.WithCancel(context.Background())
83var wg sync.WaitGroup
84wg.Add(1)
85go func() {
86defer wg.Done()
87http2Conn.Serve(ctx)
88}()
89
90edgeHTTP2Conn, err := testTransport.NewClientConn(edgeConn)
91require.NoError(t, err)
92
93for _, test := range tests {
94endpoint := fmt.Sprintf("http://localhost:8080/%s", test.endpoint)
95req, err := http.NewRequestWithContext(ctx, http.MethodGet, endpoint, nil)
96require.NoError(t, err)
97
98resp, err := edgeHTTP2Conn.RoundTrip(req)
99require.NoError(t, err)
100require.Equal(t, test.expectedStatus, resp.StatusCode)
101if test.expectedBody != nil {
102respBody, err := ioutil.ReadAll(resp.Body)
103require.NoError(t, err)
104require.Equal(t, test.expectedBody, respBody)
105}
106if test.isProxyError {
a1a554a2cthuang5 years ago107require.Equal(t, responseMetaHeaderCfd, resp.Header.Get(ResponseMetaHeaderField))
d5769519cthuang5 years ago108} else {
a1a554a2cthuang5 years ago109require.Equal(t, responseMetaHeaderOrigin, resp.Header.Get(ResponseMetaHeaderField))
d5769519cthuang5 years ago110}
111}
112cancel()
113wg.Wait()
114}
115
116type mockNamedTunnelRPCClient struct {
89b738f8Nuno Diegues5 years ago117shouldFail error
d5769519cthuang5 years ago118registered chan struct{}
119unregistered chan struct{}
120}
121
122func (mc mockNamedTunnelRPCClient) RegisterConnection(
123c context.Context,
124config *NamedTunnelConfig,
125options *tunnelpogs.ConnectionOptions,
126connIndex uint8,
127observer *Observer,
128) error {
89b738f8Nuno Diegues5 years ago129if mc.shouldFail != nil {
130return mc.shouldFail
131}
d5769519cthuang5 years ago132close(mc.registered)
133return nil
134}
135
136func (mc mockNamedTunnelRPCClient) GracefulShutdown(ctx context.Context, gracePeriod time.Duration) {
137close(mc.unregistered)
138}
139
140func (mockNamedTunnelRPCClient) Close() {}
141
142type mockRPCClientFactory struct {
89b738f8Nuno Diegues5 years ago143shouldFail error
d5769519cthuang5 years ago144registered chan struct{}
145unregistered chan struct{}
146}
147
870f5fa9Areg Harutyunyan5 years ago148func (mf *mockRPCClientFactory) newMockRPCClient(context.Context, io.ReadWriteCloser, *zerolog.Logger) NamedTunnelRPCClient {
d5769519cthuang5 years ago149return mockNamedTunnelRPCClient{
89b738f8Nuno Diegues5 years ago150shouldFail: mf.shouldFail,
d5769519cthuang5 years ago151registered: mf.registered,
152unregistered: mf.unregistered,
153}
154}
155
156type wsRespWriter struct {
157*httptest.ResponseRecorder
158readPipe *io.PipeReader
159writePipe *io.PipeWriter
160}
161
162func newWSRespWriter() *wsRespWriter {
163readPipe, writePipe := io.Pipe()
164return &wsRespWriter{
165httptest.NewRecorder(),
166readPipe,
167writePipe,
168}
169}
170
171func (w *wsRespWriter) RespBody() io.ReadWriter {
172return nowriter{w.readPipe}
173}
174
175func (w *wsRespWriter) Write(data []byte) (n int, err error) {
176return w.writePipe.Write(data)
177}
178
179func TestServeWS(t *testing.T) {
180http2Conn, _ := newTestHTTP2Connection()
181
182ctx, cancel := context.WithCancel(context.Background())
183var wg sync.WaitGroup
184wg.Add(1)
185go func() {
186defer wg.Done()
187http2Conn.Serve(ctx)
188}()
189
190respWriter := newWSRespWriter()
191readPipe, writePipe := io.Pipe()
192
193req, err := http.NewRequestWithContext(ctx, http.MethodGet, "http://localhost:8080/ws", readPipe)
194require.NoError(t, err)
195req.Header.Set(internalUpgradeHeader, websocketUpgrade)
196
197wg.Add(1)
198go func() {
199defer wg.Done()
200http2Conn.ServeHTTP(respWriter, req)
201}()
202
203data := []byte("test websocket")
204err = wsutil.WriteClientText(writePipe, data)
205require.NoError(t, err)
206
207respBody, err := wsutil.ReadServerText(respWriter.RespBody())
208require.NoError(t, err)
209require.Equal(t, data, respBody, fmt.Sprintf("Expect %s, got %s", string(data), string(respBody)))
210
211cancel()
212resp := respWriter.Result()
213// http2RespWriter should rewrite status 101 to 200
214require.Equal(t, http.StatusOK, resp.StatusCode)
a1a554a2cthuang5 years ago215require.Equal(t, responseMetaHeaderOrigin, resp.Header.Get(ResponseMetaHeaderField))
d5769519cthuang5 years ago216
217wg.Wait()
218}
219
220func TestServeControlStream(t *testing.T) {
221http2Conn, edgeConn := newTestHTTP2Connection()
222
223rpcClientFactory := mockRPCClientFactory{
224registered: make(chan struct{}),
225unregistered: make(chan struct{}),
226}
227http2Conn.newRPCClientFunc = rpcClientFactory.newMockRPCClient
228
229ctx, cancel := context.WithCancel(context.Background())
230var wg sync.WaitGroup
231wg.Add(1)
232go func() {
233defer wg.Done()
234http2Conn.Serve(ctx)
235}()
236
237req, err := http.NewRequestWithContext(ctx, http.MethodGet, "http://localhost:8080/", nil)
238require.NoError(t, err)
239req.Header.Set(internalUpgradeHeader, controlStreamUpgrade)
240
241edgeHTTP2Conn, err := testTransport.NewClientConn(edgeConn)
242require.NoError(t, err)
243
244wg.Add(1)
245go func() {
246defer wg.Done()
247edgeHTTP2Conn.RoundTrip(req)
248}()
249
250<-rpcClientFactory.registered
251cancel()
252<-rpcClientFactory.unregistered
d503aeafIgor Postelnik5 years ago253assert.False(t, http2Conn.stoppedGracefully)
d5769519cthuang5 years ago254
255wg.Wait()
256}
257
89b738f8Nuno Diegues5 years ago258func TestFailRegistration(t *testing.T) {
259http2Conn, edgeConn := newTestHTTP2Connection()
260
261rpcClientFactory := mockRPCClientFactory{
262shouldFail: errDuplicationConnection,
263registered: make(chan struct{}),
264unregistered: make(chan struct{}),
265}
266http2Conn.newRPCClientFunc = rpcClientFactory.newMockRPCClient
267
268ctx, cancel := context.WithCancel(context.Background())
269var wg sync.WaitGroup
270wg.Add(1)
271go func() {
272defer wg.Done()
273http2Conn.Serve(ctx)
274}()
275
276req, err := http.NewRequestWithContext(ctx, http.MethodGet, "http://localhost:8080/", nil)
277require.NoError(t, err)
278req.Header.Set(internalUpgradeHeader, controlStreamUpgrade)
279
280edgeHTTP2Conn, err := testTransport.NewClientConn(edgeConn)
281require.NoError(t, err)
282resp, err := edgeHTTP2Conn.RoundTrip(req)
283require.NoError(t, err)
284require.Equal(t, http.StatusBadGateway, resp.StatusCode)
285
286assert.NotNil(t, http2Conn.controlStreamErr)
287cancel()
288wg.Wait()
289}
290
d503aeafIgor Postelnik5 years ago291func TestGracefulShutdownHTTP2(t *testing.T) {
292http2Conn, edgeConn := newTestHTTP2Connection()
293
294rpcClientFactory := mockRPCClientFactory{
295registered: make(chan struct{}),
296unregistered: make(chan struct{}),
297}
cf562ef8Igor Postelnik5 years ago298events := &eventCollectorSink{}
d503aeafIgor Postelnik5 years ago299http2Conn.newRPCClientFunc = rpcClientFactory.newMockRPCClient
cf562ef8Igor Postelnik5 years ago300http2Conn.observer.RegisterSink(events)
0b16a473Igor Postelnik5 years ago301shutdownC := make(chan struct{})
302http2Conn.gracefulShutdownC = shutdownC
d503aeafIgor Postelnik5 years ago303
304ctx, cancel := context.WithCancel(context.Background())
305var wg sync.WaitGroup
306wg.Add(1)
307go func() {
308defer wg.Done()
309http2Conn.Serve(ctx)
310}()
311
312req, err := http.NewRequestWithContext(ctx, http.MethodGet, "http://localhost:8080/", nil)
313require.NoError(t, err)
314req.Header.Set(internalUpgradeHeader, controlStreamUpgrade)
315
316edgeHTTP2Conn, err := testTransport.NewClientConn(edgeConn)
317require.NoError(t, err)
318
319wg.Add(1)
320go func() {
321defer wg.Done()
322_, _ = edgeHTTP2Conn.RoundTrip(req)
323}()
324
325select {
326case <-rpcClientFactory.registered:
327break //ok
328case <-time.Tick(time.Second):
329t.Fatal("timeout out waiting for registration")
330}
331
332// signal graceful shutdown
0b16a473Igor Postelnik5 years ago333close(shutdownC)
d503aeafIgor Postelnik5 years ago334
335select {
336case <-rpcClientFactory.unregistered:
337break //ok
338case <-time.Tick(time.Second):
339t.Fatal("timeout out waiting for unregistered signal")
340}
341assert.True(t, http2Conn.stoppedGracefully)
342
343cancel()
344wg.Wait()
cf562ef8Igor Postelnik5 years ago345
346events.assertSawEvent(t, Event{
347Index: http2Conn.connIndex,
348EventType: Unregistering,
349})
d503aeafIgor Postelnik5 years ago350}
351
d5769519cthuang5 years ago352func benchmarkServeHTTP(b *testing.B, test testRequest) {
353http2Conn, edgeConn := newTestHTTP2Connection()
354
355ctx, cancel := context.WithCancel(context.Background())
356var wg sync.WaitGroup
357wg.Add(1)
358go func() {
359defer wg.Done()
360http2Conn.Serve(ctx)
361}()
362
363endpoint := fmt.Sprintf("http://localhost:8080/%s", test.endpoint)
364req, err := http.NewRequestWithContext(ctx, http.MethodGet, endpoint, nil)
365require.NoError(b, err)
366
367edgeHTTP2Conn, err := testTransport.NewClientConn(edgeConn)
368require.NoError(b, err)
369
370b.ResetTimer()
371for i := 0; i < b.N; i++ {
372b.StartTimer()
373resp, err := edgeHTTP2Conn.RoundTrip(req)
374b.StopTimer()
375require.NoError(b, err)
376require.Equal(b, test.expectedStatus, resp.StatusCode)
377if test.expectedBody != nil {
378respBody, err := ioutil.ReadAll(resp.Body)
379require.NoError(b, err)
380require.Equal(b, test.expectedBody, respBody)
381}
382resp.Body.Close()
383}
384
385cancel()
386wg.Wait()
387}
d503aeafIgor Postelnik5 years ago388
d5769519cthuang5 years ago389func BenchmarkServeHTTPSimple(b *testing.B) {
390test := testRequest{
391name: "ok",
392endpoint: "ok",
393expectedStatus: http.StatusOK,
394expectedBody: []byte(http.StatusText(http.StatusOK)),
395}
396
397benchmarkServeHTTP(b, test)
398}
399
400func BenchmarkServeHTTPLargeFile(b *testing.B) {
401test := testRequest{
402name: "large_file",
403endpoint: "large_file",
404expectedStatus: http.StatusOK,
405expectedBody: testLargeResp,
406}
407
408benchmarkServeHTTP(b, test)
409}