Merge branch 'master' into 9766-register-workflow
[arvados.git] / services / keepstore / pull_worker_test.go
1 package main
2
3 import (
4         "bytes"
5         "errors"
6         "git.curoverse.com/arvados.git/sdk/go/arvadosclient"
7         "git.curoverse.com/arvados.git/sdk/go/keepclient"
8         . "gopkg.in/check.v1"
9         "io"
10         "net/http"
11         "time"
12 )
13
14 var _ = Suite(&PullWorkerTestSuite{})
15
16 type PullWorkerTestSuite struct{}
17
18 var testPullLists map[string]string
19 var readContent string
20 var readError error
21 var putContent []byte
22 var putError error
23 var currentTestData PullWorkerTestData
24
25 func (s *PullWorkerTestSuite) SetUpTest(c *C) {
26         readContent = ""
27         readError = nil
28         putContent = []byte("")
29         putError = nil
30
31         // When a new pull request arrives, the old one will be overwritten.
32         // This behavior is verified using these two maps in the
33         // "TestPullWorkerPullList_with_two_items_latest_replacing_old"
34         testPullLists = make(map[string]string)
35 }
36
37 // Since keepstore does not come into picture in tests,
38 // we need to explicitly start the goroutine in tests.
39 func RunTestPullWorker(c *C) {
40         arv, err := arvadosclient.MakeArvadosClient()
41         c.Assert(err, Equals, nil)
42         keepClient, err := keepclient.MakeKeepClient(&arv)
43         c.Assert(err, Equals, nil)
44
45         pullq = NewWorkQueue()
46         go RunPullWorker(pullq, keepClient)
47 }
48
49 var firstPullList = []byte(`[
50                 {
51                         "locator":"acbd18db4cc2f85cedef654fccc4a4d8+3",
52                         "servers":[
53                                 "server_1",
54                                 "server_2"
55                         ]
56                 },{
57                         "locator":"37b51d194a7513e45b56f6524f2d51f2+3",
58                         "servers":[
59                                 "server_3"
60                         ]
61                 }
62         ]`)
63
64 var secondPullList = []byte(`[
65                 {
66                         "locator":"73feffa4b7f6bb68e44cf984c85f6e88+3",
67                         "servers":[
68                                 "server_1",
69                                 "server_2"
70                         ]
71                 }
72         ]`)
73
74 type PullWorkerTestData struct {
75         name         string
76         req          RequestTester
77         responseCode int
78         responseBody string
79         readContent  string
80         readError    bool
81         putError     bool
82 }
83
84 func (s *PullWorkerTestSuite) TestPullWorkerPullList_with_two_locators(c *C) {
85         defer teardown()
86
87         dataManagerToken = "DATA MANAGER TOKEN"
88         testData := PullWorkerTestData{
89                 name:         "TestPullWorkerPullList_with_two_locators",
90                 req:          RequestTester{"/pull", dataManagerToken, "PUT", firstPullList},
91                 responseCode: http.StatusOK,
92                 responseBody: "Received 2 pull requests\n",
93                 readContent:  "hello",
94                 readError:    false,
95                 putError:     false,
96         }
97
98         performTest(testData, c)
99 }
100
101 func (s *PullWorkerTestSuite) TestPullWorkerPullList_with_one_locator(c *C) {
102         defer teardown()
103
104         dataManagerToken = "DATA MANAGER TOKEN"
105         testData := PullWorkerTestData{
106                 name:         "TestPullWorkerPullList_with_one_locator",
107                 req:          RequestTester{"/pull", dataManagerToken, "PUT", secondPullList},
108                 responseCode: http.StatusOK,
109                 responseBody: "Received 1 pull requests\n",
110                 readContent:  "hola",
111                 readError:    false,
112                 putError:     false,
113         }
114
115         performTest(testData, c)
116 }
117
118 func (s *PullWorkerTestSuite) TestPullWorker_error_on_get_one_locator(c *C) {
119         defer teardown()
120
121         dataManagerToken = "DATA MANAGER TOKEN"
122         testData := PullWorkerTestData{
123                 name:         "TestPullWorker_error_on_get_one_locator",
124                 req:          RequestTester{"/pull", dataManagerToken, "PUT", secondPullList},
125                 responseCode: http.StatusOK,
126                 responseBody: "Received 1 pull requests\n",
127                 readContent:  "unused",
128                 readError:    true,
129                 putError:     false,
130         }
131
132         performTest(testData, c)
133 }
134
135 func (s *PullWorkerTestSuite) TestPullWorker_error_on_get_two_locators(c *C) {
136         defer teardown()
137
138         dataManagerToken = "DATA MANAGER TOKEN"
139         testData := PullWorkerTestData{
140                 name:         "TestPullWorker_error_on_get_two_locators",
141                 req:          RequestTester{"/pull", dataManagerToken, "PUT", firstPullList},
142                 responseCode: http.StatusOK,
143                 responseBody: "Received 2 pull requests\n",
144                 readContent:  "unused",
145                 readError:    true,
146                 putError:     false,
147         }
148
149         performTest(testData, c)
150 }
151
152 func (s *PullWorkerTestSuite) TestPullWorker_error_on_put_one_locator(c *C) {
153         defer teardown()
154
155         dataManagerToken = "DATA MANAGER TOKEN"
156         testData := PullWorkerTestData{
157                 name:         "TestPullWorker_error_on_put_one_locator",
158                 req:          RequestTester{"/pull", dataManagerToken, "PUT", secondPullList},
159                 responseCode: http.StatusOK,
160                 responseBody: "Received 1 pull requests\n",
161                 readContent:  "hello hello",
162                 readError:    false,
163                 putError:     true,
164         }
165
166         performTest(testData, c)
167 }
168
169 func (s *PullWorkerTestSuite) TestPullWorker_error_on_put_two_locators(c *C) {
170         defer teardown()
171
172         dataManagerToken = "DATA MANAGER TOKEN"
173         testData := PullWorkerTestData{
174                 name:         "TestPullWorker_error_on_put_two_locators",
175                 req:          RequestTester{"/pull", dataManagerToken, "PUT", firstPullList},
176                 responseCode: http.StatusOK,
177                 responseBody: "Received 2 pull requests\n",
178                 readContent:  "hello again",
179                 readError:    false,
180                 putError:     true,
181         }
182
183         performTest(testData, c)
184 }
185
186 // When a new pull request arrives, the old one is replaced. This test
187 // is used to check that behavior by first putting an item on the queue,
188 // and then performing the test. Thus the "testPullLists" has two entries;
189 // however, processedPullLists will see only the newest item in the list.
190 func (s *PullWorkerTestSuite) TestPullWorkerPullList_with_two_items_latest_replacing_old(c *C) {
191         defer teardown()
192
193         var firstInput = []int{1}
194         pullq = NewWorkQueue()
195         pullq.ReplaceQueue(makeTestWorkList(firstInput))
196         testPullLists["Added_before_actual_test_item"] = string(1)
197
198         dataManagerToken = "DATA MANAGER TOKEN"
199         testData := PullWorkerTestData{
200                 name:         "TestPullWorkerPullList_with_two_items_latest_replacing_old",
201                 req:          RequestTester{"/pull", dataManagerToken, "PUT", secondPullList},
202                 responseCode: http.StatusOK,
203                 responseBody: "Received 1 pull requests\n",
204                 readContent:  "hola de nuevo",
205                 readError:    false,
206                 putError:     false,
207         }
208
209         performTest(testData, c)
210 }
211
212 // In this case, the item will not be placed on pullq
213 func (s *PullWorkerTestSuite) TestPullWorker_invalid_dataManagerToken(c *C) {
214         defer teardown()
215
216         dataManagerToken = "DATA MANAGER TOKEN"
217
218         testData := PullWorkerTestData{
219                 name:         "TestPullWorkerPullList_with_two_locators",
220                 req:          RequestTester{"/pull", "invalid_dataManagerToken", "PUT", firstPullList},
221                 responseCode: http.StatusUnauthorized,
222                 responseBody: "Unauthorized\n",
223                 readContent:  "hello",
224                 readError:    false,
225                 putError:     false,
226         }
227
228         performTest(testData, c)
229 }
230
231 func performTest(testData PullWorkerTestData, c *C) {
232         KeepVM = MakeTestVolumeManager(2)
233         defer KeepVM.Close()
234
235         RunTestPullWorker(c)
236         defer pullq.Close()
237
238         currentTestData = testData
239         testPullLists[testData.name] = testData.responseBody
240
241         processedPullLists := make(map[string]string)
242
243         // Override GetContent to mock keepclient Get functionality
244         defer func(orig func(string, *keepclient.KeepClient) (io.ReadCloser, int64, string, error)) {
245                 GetContent = orig
246         }(GetContent)
247         GetContent = func(signedLocator string, keepClient *keepclient.KeepClient) (reader io.ReadCloser, contentLength int64, url string, err error) {
248                 c.Assert(getStatusItem("PullQueue", "InProgress"), Equals, float64(1))
249                 processedPullLists[testData.name] = testData.responseBody
250                 if testData.readError {
251                         err = errors.New("Error getting data")
252                         readError = err
253                         return nil, 0, "", err
254                 }
255                 readContent = testData.readContent
256                 cb := &ClosingBuffer{bytes.NewBufferString(testData.readContent)}
257                 var rc io.ReadCloser
258                 rc = cb
259                 return rc, int64(len(testData.readContent)), "", nil
260         }
261
262         // Override PutContent to mock PutBlock functionality
263         defer func(orig func([]byte, string) error) { PutContent = orig }(PutContent)
264         PutContent = func(content []byte, locator string) (err error) {
265                 if testData.putError {
266                         err = errors.New("Error putting data")
267                         putError = err
268                         return err
269                 }
270                 putContent = content
271                 return nil
272         }
273
274         c.Assert(getStatusItem("PullQueue", "InProgress"), Equals, float64(0))
275         c.Assert(getStatusItem("PullQueue", "Queued"), Equals, float64(0))
276
277         response := IssueRequest(&testData.req)
278         c.Assert(response.Code, Equals, testData.responseCode)
279         c.Assert(response.Body.String(), Equals, testData.responseBody)
280
281         expectEqualWithin(c, time.Second, 0, func() interface{} {
282                 st := pullq.Status()
283                 return st.InProgress + st.Queued
284         })
285
286         if testData.name == "TestPullWorkerPullList_with_two_items_latest_replacing_old" {
287                 c.Assert(len(testPullLists), Equals, 2)
288                 c.Assert(len(processedPullLists), Equals, 1)
289                 c.Assert(testPullLists["Added_before_actual_test_item"], NotNil)
290                 c.Assert(testPullLists["TestPullWorkerPullList_with_two_items_latest_replacing_old"], NotNil)
291                 c.Assert(processedPullLists["TestPullWorkerPullList_with_two_items_latest_replacing_old"], NotNil)
292         } else {
293                 if testData.responseCode == http.StatusOK {
294                         c.Assert(len(testPullLists), Equals, 1)
295                         c.Assert(len(processedPullLists), Equals, 1)
296                         c.Assert(testPullLists[testData.name], NotNil)
297                 } else {
298                         c.Assert(len(testPullLists), Equals, 1)
299                         c.Assert(len(processedPullLists), Equals, 0)
300                 }
301         }
302
303         if testData.readError {
304                 c.Assert(readError, NotNil)
305         } else if testData.responseCode == http.StatusOK {
306                 c.Assert(readError, IsNil)
307                 c.Assert(readContent, Equals, testData.readContent)
308                 if testData.putError {
309                         c.Assert(putError, NotNil)
310                 } else {
311                         c.Assert(putError, IsNil)
312                         c.Assert(string(putContent), Equals, testData.readContent)
313                 }
314         }
315
316         expectChannelEmpty(c, pullq.NextItem)
317 }
318
319 type ClosingBuffer struct {
320         *bytes.Buffer
321 }
322
323 func (cb *ClosingBuffer) Close() (err error) {
324         return
325 }