GitVita
New repositoryImport repositoryNew issue
Your repositories Your dashboard Sign in

cloudflare/cloudflared

Public

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

Watch0Fork0Star0
CodeCommitsIssuesPull requestsActionsInsightsSecurity
2020.11.7

Branches

Tags

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

Clone

HTTPS

Download ZIP

cloudflared/connection

connection/http2_test.go

303lines · 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
15"github.com/cloudflare/cloudflared/logger"
16"github.com/cloudflare/cloudflared/tunnelrpc/pogs"
17tunnelpogs "github.com/cloudflare/cloudflared/tunnelrpc/pogs"
18"github.com/gobwas/ws/wsutil"
19"github.com/stretchr/testify/require"
20"golang.org/x/net/http2"
21)
22
23var (
24testTransport = http2.Transport{}
25)
26
27func newTestHTTP2Connection() (*http2Connection, net.Conn) {
28edgeConn, originConn := net.Pipe()
29var connIndex = uint8(0)
30return NewHTTP2Connection(
31originConn,
32testConfig,
33&NamedTunnelConfig{},
34&pogs.ConnectionOptions{},
35testObserver,
36connIndex,
37mockConnectedFuse{},
38), edgeConn
39}
40
41func TestServeHTTP(t *testing.T) {
42tests := []testRequest{
43{
44name: "ok",
45endpoint: "ok",
46expectedStatus: http.StatusOK,
47expectedBody: []byte(http.StatusText(http.StatusOK)),
48},
49{
50name: "large_file",
51endpoint: "large_file",
52expectedStatus: http.StatusOK,
53expectedBody: testLargeResp,
54},
55{
56name: "Bad request",
57endpoint: "400",
58expectedStatus: http.StatusBadRequest,
59expectedBody: []byte(http.StatusText(http.StatusBadRequest)),
60},
61{
62name: "Internal server error",
63endpoint: "500",
64expectedStatus: http.StatusInternalServerError,
65expectedBody: []byte(http.StatusText(http.StatusInternalServerError)),
66},
67{
68name: "Proxy error",
69endpoint: "error",
70expectedStatus: http.StatusBadGateway,
71expectedBody: nil,
72isProxyError: true,
73},
74}
75
76http2Conn, edgeConn := newTestHTTP2Connection()
77
78ctx, cancel := context.WithCancel(context.Background())
79var wg sync.WaitGroup
80wg.Add(1)
81go func() {
82defer wg.Done()
83http2Conn.Serve(ctx)
84}()
85
86edgeHTTP2Conn, err := testTransport.NewClientConn(edgeConn)
87require.NoError(t, err)
88
89for _, test := range tests {
90endpoint := fmt.Sprintf("http://localhost:8080/%s", test.endpoint)
91req, err := http.NewRequestWithContext(ctx, http.MethodGet, endpoint, nil)
92require.NoError(t, err)
93
94resp, err := edgeHTTP2Conn.RoundTrip(req)
95require.NoError(t, err)
96require.Equal(t, test.expectedStatus, resp.StatusCode)
97if test.expectedBody != nil {
98respBody, err := ioutil.ReadAll(resp.Body)
99require.NoError(t, err)
100require.Equal(t, test.expectedBody, respBody)
101}
102if test.isProxyError {
103require.Equal(t, responseMetaHeaderCfd, resp.Header.Get(responseMetaHeaderField))
104} else {
105require.Equal(t, responseMetaHeaderOrigin, resp.Header.Get(responseMetaHeaderField))
106}
107}
108cancel()
109wg.Wait()
110}
111
112type mockNamedTunnelRPCClient struct {
113registered chan struct{}
114unregistered chan struct{}
115}
116
117func (mc mockNamedTunnelRPCClient) RegisterConnection(
118c context.Context,
119config *NamedTunnelConfig,
120options *tunnelpogs.ConnectionOptions,
121connIndex uint8,
122observer *Observer,
123) error {
124close(mc.registered)
125return nil
126}
127
128func (mc mockNamedTunnelRPCClient) GracefulShutdown(ctx context.Context, gracePeriod time.Duration) {
129close(mc.unregistered)
130}
131
132func (mockNamedTunnelRPCClient) Close() {}
133
134type mockRPCClientFactory struct {
135registered chan struct{}
136unregistered chan struct{}
137}
138
139func (mf *mockRPCClientFactory) newMockRPCClient(context.Context, io.ReadWriteCloser, logger.Service) NamedTunnelRPCClient {
140return mockNamedTunnelRPCClient{
141registered: mf.registered,
142unregistered: mf.unregistered,
143}
144}
145
146type wsRespWriter struct {
147*httptest.ResponseRecorder
148readPipe *io.PipeReader
149writePipe *io.PipeWriter
150}
151
152func newWSRespWriter() *wsRespWriter {
153readPipe, writePipe := io.Pipe()
154return &wsRespWriter{
155httptest.NewRecorder(),
156readPipe,
157writePipe,
158}
159}
160
161func (w *wsRespWriter) RespBody() io.ReadWriter {
162return nowriter{w.readPipe}
163}
164
165func (w *wsRespWriter) Write(data []byte) (n int, err error) {
166return w.writePipe.Write(data)
167}
168
169func TestServeWS(t *testing.T) {
170http2Conn, _ := newTestHTTP2Connection()
171
172ctx, cancel := context.WithCancel(context.Background())
173var wg sync.WaitGroup
174wg.Add(1)
175go func() {
176defer wg.Done()
177http2Conn.Serve(ctx)
178}()
179
180respWriter := newWSRespWriter()
181readPipe, writePipe := io.Pipe()
182
183req, err := http.NewRequestWithContext(ctx, http.MethodGet, "http://localhost:8080/ws", readPipe)
184require.NoError(t, err)
185req.Header.Set(internalUpgradeHeader, websocketUpgrade)
186
187wg.Add(1)
188go func() {
189defer wg.Done()
190http2Conn.ServeHTTP(respWriter, req)
191}()
192
193data := []byte("test websocket")
194err = wsutil.WriteClientText(writePipe, data)
195require.NoError(t, err)
196
197respBody, err := wsutil.ReadServerText(respWriter.RespBody())
198require.NoError(t, err)
199require.Equal(t, data, respBody, fmt.Sprintf("Expect %s, got %s", string(data), string(respBody)))
200
201cancel()
202resp := respWriter.Result()
203// http2RespWriter should rewrite status 101 to 200
204require.Equal(t, http.StatusOK, resp.StatusCode)
205require.Equal(t, responseMetaHeaderOrigin, resp.Header.Get(responseMetaHeaderField))
206
207wg.Wait()
208}
209
210func TestServeControlStream(t *testing.T) {
211http2Conn, edgeConn := newTestHTTP2Connection()
212
213rpcClientFactory := mockRPCClientFactory{
214registered: make(chan struct{}),
215unregistered: make(chan struct{}),
216}
217http2Conn.newRPCClientFunc = rpcClientFactory.newMockRPCClient
218
219ctx, cancel := context.WithCancel(context.Background())
220var wg sync.WaitGroup
221wg.Add(1)
222go func() {
223defer wg.Done()
224http2Conn.Serve(ctx)
225}()
226
227req, err := http.NewRequestWithContext(ctx, http.MethodGet, "http://localhost:8080/", nil)
228require.NoError(t, err)
229req.Header.Set(internalUpgradeHeader, controlStreamUpgrade)
230
231edgeHTTP2Conn, err := testTransport.NewClientConn(edgeConn)
232require.NoError(t, err)
233
234wg.Add(1)
235go func() {
236defer wg.Done()
237edgeHTTP2Conn.RoundTrip(req)
238}()
239
240<-rpcClientFactory.registered
241cancel()
242<-rpcClientFactory.unregistered
243
244wg.Wait()
245}
246
247func benchmarkServeHTTP(b *testing.B, test testRequest) {
248http2Conn, edgeConn := newTestHTTP2Connection()
249
250ctx, cancel := context.WithCancel(context.Background())
251var wg sync.WaitGroup
252wg.Add(1)
253go func() {
254defer wg.Done()
255http2Conn.Serve(ctx)
256}()
257
258endpoint := fmt.Sprintf("http://localhost:8080/%s", test.endpoint)
259req, err := http.NewRequestWithContext(ctx, http.MethodGet, endpoint, nil)
260require.NoError(b, err)
261
262edgeHTTP2Conn, err := testTransport.NewClientConn(edgeConn)
263require.NoError(b, err)
264
265b.ResetTimer()
266for i := 0; i < b.N; i++ {
267b.StartTimer()
268resp, err := edgeHTTP2Conn.RoundTrip(req)
269b.StopTimer()
270require.NoError(b, err)
271require.Equal(b, test.expectedStatus, resp.StatusCode)
272if test.expectedBody != nil {
273respBody, err := ioutil.ReadAll(resp.Body)
274require.NoError(b, err)
275require.Equal(b, test.expectedBody, respBody)
276}
277resp.Body.Close()
278}
279
280cancel()
281wg.Wait()
282}
283func BenchmarkServeHTTPSimple(b *testing.B) {
284test := testRequest{
285name: "ok",
286endpoint: "ok",
287expectedStatus: http.StatusOK,
288expectedBody: []byte(http.StatusText(http.StatusOK)),
289}
290
291benchmarkServeHTTP(b, test)
292}
293
294func BenchmarkServeHTTPLargeFile(b *testing.B) {
295test := testRequest{
296name: "large_file",
297endpoint: "large_file",
298expectedStatus: http.StatusOK,
299expectedBody: testLargeResp,
300}
301
302benchmarkServeHTTP(b, test)
303}