gostride/stride_test.go at master · pipelinedb/gostride · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
package stride

import (
	"compress/gzip"
	"io/ioutil"
	"net/http"
	"net/http/httptest"
	"testing"

	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/suite"
)

type StrideTestSuite struct {
	suite.Suite
}

func createMockServer(t *testing.T) *httptest.Server {
	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		path := r.URL.Path

		w.Header().Set("Content-Type", "application/json")

		switch r.Method {
		case http.MethodGet:
			w.WriteHeader(http.StatusOK)

			if path == "/v1/collect" {
				w.Write([]byte(`["stream0", "stream1"]`))
			}
		case http.MethodPost:
			if r.Header.Get("Content-Encoding") == "gzip" {
				gz, _ := gzip.NewReader(r.Body)
				r.Body = gz
				defer gz.Close()
			}
			w.WriteHeader(http.StatusCreated)
			assert.NotNil(t, r.Body)
			body, _ := ioutil.ReadAll(r.Body)
			w.Write(body)
		case http.MethodDelete:
			w.WriteHeader(http.StatusOK)
		}
	}))

	return server
}

func (suite *StrideTestSuite) TestCollectCompression() {
	server := createMockServer(suite.T())
	config := NewConfig()
	config.Endpoint = server.URL + "/v1"

	s := NewStride("key", config)
	events := []map[string]string{
		{"x": "x", "y": "y", "z": "z"},
		{"x": "x", "y": "y", "z": "z"},
		{"x": "x", "y": "y", "z": "z"},
		{"x": "x", "y": "y", "z": "z"},
	}
	r := s.Post("/collect", events)

	assert.Equal(suite.T(), http.StatusCreated, r.StatusCode)
	assert.Equal(suite.T(), len(events), len(r.Data.([]interface{})))
}

func (suite *StrideTestSuite) TestPathValidation() {
	assert.True(suite.T(), isPathValid(http.MethodGet, "/collect"))
	assert.True(suite.T(), isPathValid(http.MethodPost, "/collect"))
	assert.True(suite.T(), isPathValid(http.MethodGet, "/collect/stream"))
	assert.True(suite.T(), isPathValid(http.MethodPost, "/collect/stream"))
	assert.True(suite.T(), isPathValid(http.MethodDelete, "/collect/stream"))
	assert.True(suite.T(), isPathValid("Subscribe", "/collect/stream"))
	assert.True(suite.T(), isPathValid(http.MethodGet, "/process"))
	assert.True(suite.T(), isPathValid(http.MethodGet, "/process/proc"))
	assert.True(suite.T(), isPathValid(http.MethodPost, "/process/proc"))
	assert.True(suite.T(), isPathValid(http.MethodDelete, "/process/proc"))
	assert.True(suite.T(), isPathValid("Subscribe", "/process/proc"))
	assert.True(suite.T(), isPathValid(http.MethodGet, "/analyze"))
	assert.True(suite.T(), isPathValid(http.MethodPost, "/analyze"))
	assert.True(suite.T(), isPathValid(http.MethodGet, "/analyze/query"))
	assert.True(suite.T(), isPathValid(http.MethodPost, "/analyze/query"))
	assert.True(suite.T(), isPathValid(http.MethodDelete, "/analyze/query"))
	assert.True(suite.T(), isPathValid(http.MethodGet, "/analyze/query/results"))
	assert.True(suite.T(), isPathValid(http.MethodGet, "/analyze/query/results"))
	assert.True(suite.T(), isPathValid(http.MethodPut, "/analyze/query"))
	assert.True(suite.T(), isPathValid(http.MethodPut, "/process/proc"))

	assert.False(suite.T(), isPathValid(http.MethodGet, "/collect/_stream"))
	assert.False(suite.T(), isPathValid(http.MethodGet, "/collect/1stream"))
	assert.False(suite.T(), isPathValid(http.MethodGet, "/collect/stream/results"))
	assert.False(suite.T(), isPathValid(http.MethodPost, "/process"))
	assert.False(suite.T(), isPathValid("Subscribe", "/analyze/query"))
	assert.False(suite.T(), isPathValid(http.MethodPut, "/collect/stream"))
	assert.False(suite.T(), isPathValid(http.MethodPut, "/collect"))
	assert.False(suite.T(), isPathValid(http.MethodPut, "/process"))
	assert.False(suite.T(), isPathValid(http.MethodPut, "/analyze"))
}

func (suite *StrideTestSuite) TestMethods() {
	server := createMockServer(suite.T())

	config := NewConfig()
	config.Endpoint = server.URL + "/v1"

	s := NewStride("key", config)

	r := s.Get("/collect")
	assert.Equal(suite.T(), http.StatusOK, r.StatusCode)
	assert.Nil(suite.T(), r.Error)
	streams, ok := r.Data.([]interface{})
	assert.True(suite.T(), ok)
	assert.Equal(suite.T(), []interface{}{"stream0", "stream1"}, streams)

	proc := map[string]interface{}{
		"query": "SELECT 1",
		"action": map[string]interface{}{
			"type": "MATERIALIZE",
		},
	}
	r = s.Post("/process/p1", proc)
	assert.Equal(suite.T(), http.StatusCreated, r.StatusCode)
	assert.Nil(suite.T(), r.Error)
	data, ok := r.Data.(map[string]interface{})
	assert.True(suite.T(), ok)
	assert.Equal(suite.T(), proc, data)

	r = s.Delete("/analyze/q1")
	assert.Equal(suite.T(), http.StatusOK, r.StatusCode)
	assert.Nil(suite.T(), r.Error)
	assert.Nil(suite.T(), r.Data)
}

func TestStrideTestSuite(t *testing.T) {
	suite.Run(t, new(StrideTestSuite))
}