test(http): add flux query service test

http/query_handler.go

@@ -157,14 +157,14 @@ type FluxQueryService struct {
 }
 
 // Query runs a flux query against a influx server and decodes the result
-func (s *FluxQueryService) Query(ctx context.Context, req *query.Request) (flux.ResultIterator, error) {
+func (s *FluxQueryService) Query(ctx context.Context, r *query.Request) (flux.ResultIterator, error) {
 	u, err := newURL(s.URL, fluxPath)
 	if err != nil {
 		return nil, err
 	}
 
 	preq := &query.ProxyRequest{
-		Request: *req,
+		Request: *r,
 		Dialect: csv.DefaultDialect(),
 	}
 	var body bytes.Buffer
@@ -192,7 +192,6 @@ func (s *FluxQueryService) Query(ctx context.Context, req *query.Request) (flux.
 	if err != nil {
 		return nil, err
 	}
-	defer resp.Body.Close()
 
 	if err := CheckError(resp); err != nil {
 		return nil, err

http/query_handler_test.go

@@ -4,12 +4,15 @@ import (
 	"bytes"
 	"context"
 	"fmt"
+	"net/http"
+	"net/http/httptest"
+	"reflect"
+	"regexp"
+	"testing"
+
 	"github.com/influxdata/flux/csv"
 	"github.com/influxdata/flux/lang"
 	"github.com/influxdata/platform/query"
-	"net/http"
-	"net/http/httptest"
-	"testing"
 )
 
 func TestFluxService_Query(t *testing.T) {
@@ -82,3 +85,99 @@ func TestFluxService_Query(t *testing.T) {
 		})
 	}
 }
+
+func TestFluxQueryService_Query(t *testing.T) {
+	tests := []struct {
+		name    string
+		token   string
+		ctx     context.Context
+		r       *query.Request
+		csv     string
+		status  int
+		want    string
+		wantErr bool
+	}{
+		{
+			name:  "error status",
+			token: "mytoken",
+			ctx:   context.Background(),
+			r: &query.Request{
+				Compiler: lang.FluxCompiler{
+					Query: "from()",
+				},
+			},
+			status:  http.StatusUnauthorized,
+			wantErr: true,
+		},
+		{
+			name:  "returns csv",
+			token: "mytoken",
+			ctx:   context.Background(),
+			r: &query.Request{
+				Compiler: lang.FluxCompiler{
+					Query: "from()",
+				},
+			},
+			status: http.StatusOK,
+			csv: `#datatype,string,long,dateTime:RFC3339,double,long,string,boolean,string,string,string
+#group,false,false,false,false,false,false,false,true,true,true
+#default,0,,,,,,,,,
+,result,table,_time,usage_user,test,mystr,this,cpu,host,_measurement
+,,0,2018-08-29T13:08:47Z,10.2,10,yay,true,cpu-total,a,cpui
+`,
+			want: toCRLF(`,,,2018-08-29T13:08:47Z,10.2,10,yay,true,cpu-total,a,cpui
+
+`),
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+				w.WriteHeader(tt.status)
+				fmt.Fprintln(w, tt.csv)
+			}))
+			s := &FluxQueryService{
+				URL:   ts.URL,
+				Token: tt.token,
+			}
+			res, err := s.Query(tt.ctx, tt.r)
+			if (err != nil) != tt.wantErr {
+				t.Errorf("FluxQueryService.Query() error = %v, wantErr %v", err, tt.wantErr)
+				return
+			}
+			if res != nil && res.Err() != nil {
+				t.Errorf("FluxQueryService.Query() result error = %v", res.Err())
+				return
+			}
+			if tt.wantErr {
+				return
+			}
+			defer res.Cancel()
+
+			enc := csv.NewMultiResultEncoder(csv.ResultEncoderConfig{
+				NoHeader:  true,
+				Delimiter: ',',
+			})
+			b := bytes.Buffer{}
+			n, err := enc.Encode(&b, res)
+			if err != nil {
+				t.Errorf("FluxQueryService.Query() encode error = %v", err)
+				return
+			}
+			if n != int64(len(tt.want)) {
+				t.Errorf("FluxQueryService.Query() encode result = %d, want %d", n, len(tt.want))
+			}
+
+			got := b.String()
+			if !reflect.DeepEqual(got, tt.want) {
+				t.Errorf("FluxQueryService.Query() =\n%s\n%s", got, tt.want)
+			}
+		})
+	}
+}
+
+var crlfPattern = regexp.MustCompile(`\r?\n`)
+
+func toCRLF(data string) string {
+	return crlfPattern.ReplaceAllString(data, "\r\n")
+}