OLD | NEW |
| (Empty) |
1 // Copyright 2015 The Chromium Authors. All rights reserved. | |
2 // Use of this source code is governed by a BSD-style license that can be | |
3 // found in the LICENSE file. | |
4 | |
5 package memory | |
6 | |
7 import ( | |
8 "fmt" | |
9 "math/rand" | |
10 "net/http" | |
11 "testing" | |
12 "time" | |
13 | |
14 "github.com/luci/gae" | |
15 "github.com/luci/luci-go/common/clock" | |
16 "github.com/luci/luci-go/common/clock/testclock" | |
17 . "github.com/smartystreets/goconvey/convey" | |
18 "golang.org/x/net/context" | |
19 ) | |
20 | |
21 func TestTaskQueue(t *testing.T) { | |
22 t.Parallel() | |
23 | |
24 Convey("TaskQueue", t, func() { | |
25 now := time.Date(2000, time.January, 1, 1, 1, 1, 1, time.UTC) | |
26 c, tc := testclock.UseTime(context.Background(), now) | |
27 c = gae.SetMathRand(c, rand.New(rand.NewSource(clock.Now(c).Unix
Nano()))) | |
28 c = Use(c) | |
29 | |
30 tq := gae.GetTQ(c).(interface { | |
31 gae.TaskQueue | |
32 gae.TQTestable | |
33 }) | |
34 | |
35 So(tq, ShouldNotBeNil) | |
36 | |
37 Convey("implements TQMultiReadWriter", func() { | |
38 Convey("Add", func() { | |
39 t := &gae.TQTask{Path: "/hello/world"} | |
40 | |
41 Convey("works", func() { | |
42 t.Delay = 4 * time.Second | |
43 t.Header = http.Header{} | |
44 t.Header.Add("Cat", "tabby") | |
45 t.Payload = []byte("watwatwat") | |
46 t.RetryOptions = &gae.TQRetryOptions{Age
Limit: 7 * time.Second} | |
47 _, err := tq.Add(t, "") | |
48 So(err, ShouldBeNil) | |
49 name := "Z_UjshxM9ecyMQfGbZmUGOEcgxWU0_5
CGLl_-RntudwAw2DqQ5-58bzJiWQN4OKzeuUb9O4JrPkUw2rOvk2Ax46THojnQ6avBQgZdrKcJmrwQ6o
4qKfJdiyUbGXvy691yRfzLeQhs6cBhWrgf3wH-VPMcA4SC-zlbJ2U8An7I0zJQA5nBFnMNoMgT-2peGo
ay3rCSbj4z9VFFm9kS_i6JCaQH518ujLDSNCYdjTq6B6lcWrZAh0U_q3a1S2nXEwrKiw_t9MTNQFgAQZ
WyGBbvZQPmeRYtu8SPaWzTfd25v_YWgBuVL2rRSPSMvlDwE04nNdtvVzE8vNNiA1zRimmdzKeqATQF9_
ReUvj4D7U8dcS703DZWfKMBLgBffY9jqCassOOOw77V72Oq5EVauUw3Qw0L6bBsfM9FtahTKUdabzRZj
XUoze3EK4KXPt3-wdidau-8JrVf2XFocjjZbwHoxcGvbtT3b4nGLDlgwdC00bwaFBZWff" | |
50 So(*tq.GetScheduledTasks()["default"][na
me], ShouldResemble, gae.TQTask{ | |
51 ETA: now.Add(4 * time.S
econd), | |
52 Header: http.Header{"Cat":
[]string{"tabby"}}, | |
53 Method: "POST", | |
54 Name: name, | |
55 Path: "/hello/world", | |
56 Payload: []byte("watwatwat"
), | |
57 RetryOptions: &gae.TQRetryOption
s{AgeLimit: 7 * time.Second}, | |
58 }) | |
59 }) | |
60 | |
61 Convey("cannot add to bad queues", func() { | |
62 _, err := tq.Add(nil, "waaat") | |
63 So(err.Error(), ShouldContainSubstring,
"UNKNOWN_QUEUE") | |
64 | |
65 Convey("but you can add Queues when test
ing", func() { | |
66 tq.CreateQueue("waaat") | |
67 _, err := tq.Add(t, "waaat") | |
68 So(err, ShouldBeNil) | |
69 | |
70 Convey("you just can't add them
twice", func() { | |
71 So(func() { tq.CreateQue
ue("waaat") }, ShouldPanic) | |
72 }) | |
73 }) | |
74 }) | |
75 | |
76 Convey("supplies a URL if it's missing", func()
{ | |
77 t.Path = "" | |
78 tr, err := tq.Add(t, "") | |
79 So(err, ShouldBeNil) | |
80 So(tr.Path, ShouldEqual, "/_ah/queue/def
ault") | |
81 }) | |
82 | |
83 Convey("cannot add twice", func() { | |
84 t.Name = "bob" | |
85 _, err := tq.Add(t, "") | |
86 So(err, ShouldBeNil) | |
87 | |
88 // can't add the same one twice! | |
89 _, err = tq.Add(t, "") | |
90 So(err, ShouldEqual, gae.ErrTQTaskAlread
yAdded) | |
91 }) | |
92 | |
93 Convey("cannot add deleted task", func() { | |
94 t.Name = "bob" | |
95 _, err := tq.Add(t, "") | |
96 So(err, ShouldBeNil) | |
97 | |
98 err = tq.Delete(t, "") | |
99 So(err, ShouldBeNil) | |
100 | |
101 // can't add a deleted task! | |
102 _, err = tq.Add(t, "") | |
103 So(err, ShouldEqual, gae.ErrTQTaskAlread
yAdded) | |
104 }) | |
105 | |
106 Convey("cannot set ETA+Delay", func() { | |
107 t.ETA = clock.Now(c).Add(time.Hour) | |
108 tc.Add(time.Second) | |
109 t.Delay = time.Hour | |
110 So(func() { tq.Add(t, "") }, ShouldPanic
) | |
111 }) | |
112 | |
113 Convey("must use a reasonable method", func() { | |
114 t.Method = "Crystal" | |
115 _, err := tq.Add(t, "") | |
116 So(err.Error(), ShouldContainSubstring,
"bad method") | |
117 }) | |
118 | |
119 Convey("payload gets dumped for non POST/PUT met
hods", func() { | |
120 t.Method = "HEAD" | |
121 t.Payload = []byte("coool") | |
122 tq, err := tq.Add(t, "") | |
123 So(err, ShouldBeNil) | |
124 So(tq.Payload, ShouldBeNil) | |
125 | |
126 // check that it didn't modify our origi
nal | |
127 So(t.Payload, ShouldResemble, []byte("co
ool")) | |
128 }) | |
129 | |
130 Convey("invalid names are rejected", func() { | |
131 t.Name = "happy times" | |
132 _, err := tq.Add(t, "") | |
133 So(err.Error(), ShouldContainSubstring,
"INVALID_TASK_NAME") | |
134 }) | |
135 | |
136 Convey("AddMulti also works", func() { | |
137 t2 := dupTask(t) | |
138 t2.Path = "/hi/city" | |
139 | |
140 expect := []*gae.TQTask{t, t2} | |
141 | |
142 tasks, err := tq.AddMulti(expect, "defau
lt") | |
143 So(err, ShouldBeNil) | |
144 So(len(tasks), ShouldEqual, 2) | |
145 So(len(tq.GetScheduledTasks()["default"]
), ShouldEqual, 2) | |
146 | |
147 for i := range expect { | |
148 Convey(fmt.Sprintf("task %d: %s"
, i, expect[i].Path), func() { | |
149 expect[i].Method = "POST
" | |
150 expect[i].ETA = now | |
151 So(expect[i].Name, Shoul
dEqual, "") | |
152 So(len(tasks[i].Name), S
houldEqual, 500) | |
153 tasks[i].Name = "" | |
154 So(tasks[i], ShouldResem
ble, expect[i]) | |
155 }) | |
156 } | |
157 }) | |
158 }) | |
159 | |
160 Convey("Delete", func() { | |
161 t := &gae.TQTask{Path: "/hello/world"} | |
162 tEnQ, err := tq.Add(t, "") | |
163 So(err, ShouldBeNil) | |
164 | |
165 Convey("works", func() { | |
166 t.Name = tEnQ.Name | |
167 err := tq.Delete(t, "") | |
168 So(err, ShouldBeNil) | |
169 So(len(tq.GetScheduledTasks()["default"]
), ShouldEqual, 0) | |
170 So(len(tq.GetTombstonedTasks()["default"
]), ShouldEqual, 1) | |
171 So(tq.GetTombstonedTasks()["default"][tE
nQ.Name], ShouldResemble, tEnQ) | |
172 }) | |
173 | |
174 Convey("cannot delete a task twice", func() { | |
175 err := tq.Delete(tEnQ, "") | |
176 So(err, ShouldBeNil) | |
177 | |
178 err = tq.Delete(tEnQ, "") | |
179 So(err.Error(), ShouldContainSubstring,
"TOMBSTONED_TASK") | |
180 | |
181 Convey("but you can if you do a reset",
func() { | |
182 tq.ResetTasks() | |
183 | |
184 tEnQ, err := tq.Add(t, "") | |
185 So(err, ShouldBeNil) | |
186 err = tq.Delete(tEnQ, "") | |
187 So(err, ShouldBeNil) | |
188 }) | |
189 }) | |
190 | |
191 Convey("cannot delete from bogus queues", func()
{ | |
192 err := tq.Delete(t, "wat") | |
193 So(err.Error(), ShouldContainSubstring,
"UNKNOWN_QUEUE") | |
194 }) | |
195 | |
196 Convey("cannot delete a missing task", func() { | |
197 t.Name = "tarntioarenstyw" | |
198 err := tq.Delete(t, "") | |
199 So(err.Error(), ShouldContainSubstring,
"UNKNOWN_TASK") | |
200 }) | |
201 | |
202 Convey("DeleteMulti also works", func() { | |
203 t2 := dupTask(t) | |
204 t2.Path = "/hi/city" | |
205 tEnQ2, err := tq.Add(t2, "") | |
206 So(err, ShouldBeNil) | |
207 | |
208 Convey("usually works", func() { | |
209 err = tq.DeleteMulti([]*gae.TQTa
sk{tEnQ, tEnQ2}, "") | |
210 So(err, ShouldBeNil) | |
211 So(len(tq.GetScheduledTasks()["d
efault"]), ShouldEqual, 0) | |
212 So(len(tq.GetTombstonedTasks()["
default"]), ShouldEqual, 2) | |
213 }) | |
214 }) | |
215 }) | |
216 }) | |
217 | |
218 Convey("works with transactions", func() { | |
219 t := &gae.TQTask{Path: "/hello/world"} | |
220 tEnQ, err := tq.Add(t, "") | |
221 So(err, ShouldBeNil) | |
222 | |
223 t2 := &gae.TQTask{Path: "/hi/city"} | |
224 tEnQ2, err := tq.Add(t2, "") | |
225 So(err, ShouldBeNil) | |
226 | |
227 err = tq.Delete(tEnQ2, "") | |
228 So(err, ShouldBeNil) | |
229 | |
230 Convey("can view regular tasks", func() { | |
231 gae.GetRDS(c).RunInTransaction(func(c context.Co
ntext) error { | |
232 tq := gae.GetTQ(c).(interface { | |
233 gae.TQTestable | |
234 gae.TaskQueue | |
235 }) | |
236 | |
237 So(tq.GetScheduledTasks()["default"][tEn
Q.Name], ShouldResemble, tEnQ) | |
238 So(tq.GetTombstonedTasks()["default"][tE
nQ2.Name], ShouldResemble, tEnQ2) | |
239 So(tq.GetTransactionTasks()["default"],
ShouldBeNil) | |
240 return nil | |
241 }, nil) | |
242 }) | |
243 | |
244 Convey("can add a new task", func() { | |
245 tEnQ3 := (*gae.TQTask)(nil) | |
246 | |
247 gae.GetRDS(c).RunInTransaction(func(c context.Co
ntext) error { | |
248 tq := gae.GetTQ(c).(interface { | |
249 gae.TQTestable | |
250 gae.TaskQueue | |
251 }) | |
252 | |
253 t3 := &gae.TQTask{Path: "/sandwitch/vict
ory"} | |
254 tEnQ3, err = tq.Add(t3, "") | |
255 So(err, ShouldBeNil) | |
256 | |
257 So(tq.GetScheduledTasks()["default"][tEn
Q.Name], ShouldResemble, tEnQ) | |
258 So(tq.GetTombstonedTasks()["default"][tE
nQ2.Name], ShouldResemble, tEnQ2) | |
259 So(tq.GetTransactionTasks()["default"][0
], ShouldResemble, tEnQ3) | |
260 return nil | |
261 }, nil) | |
262 | |
263 // name gets generated at transaction-commit-tim
e | |
264 for name := range tq.GetScheduledTasks()["defaul
t"] { | |
265 if name == tEnQ.Name { | |
266 continue | |
267 } | |
268 tEnQ3.Name = name | |
269 break | |
270 } | |
271 | |
272 So(tq.GetScheduledTasks()["default"][tEnQ.Name],
ShouldResemble, tEnQ) | |
273 So(tq.GetScheduledTasks()["default"][tEnQ3.Name]
, ShouldResemble, tEnQ3) | |
274 So(tq.GetTombstonedTasks()["default"][tEnQ2.Name
], ShouldResemble, tEnQ2) | |
275 So(tq.GetTransactionTasks()["default"], ShouldBe
Nil) | |
276 }) | |
277 | |
278 Convey("can a new task (but reset the state in a test)",
func() { | |
279 tEnQ3 := (*gae.TQTask)(nil) | |
280 | |
281 ttq := interface { | |
282 gae.TQTestable | |
283 gae.TaskQueue | |
284 }(nil) | |
285 | |
286 gae.GetRDS(c).RunInTransaction(func(c context.Co
ntext) error { | |
287 ttq = gae.GetTQ(c).(interface { | |
288 gae.TQTestable | |
289 gae.TaskQueue | |
290 }) | |
291 | |
292 t3 := &gae.TQTask{Path: "/sandwitch/vict
ory"} | |
293 tEnQ3, err = ttq.Add(t3, "") | |
294 So(err, ShouldBeNil) | |
295 | |
296 So(ttq.GetScheduledTasks()["default"][tE
nQ.Name], ShouldResemble, tEnQ) | |
297 So(ttq.GetTombstonedTasks()["default"][t
EnQ2.Name], ShouldResemble, tEnQ2) | |
298 So(ttq.GetTransactionTasks()["default"][
0], ShouldResemble, tEnQ3) | |
299 | |
300 ttq.ResetTasks() | |
301 | |
302 So(len(ttq.GetScheduledTasks()["default"
]), ShouldEqual, 0) | |
303 So(len(ttq.GetTombstonedTasks()["default
"]), ShouldEqual, 0) | |
304 So(len(ttq.GetTransactionTasks()["defaul
t"]), ShouldEqual, 0) | |
305 | |
306 return nil | |
307 }, nil) | |
308 | |
309 So(len(tq.GetScheduledTasks()["default"]), Shoul
dEqual, 0) | |
310 So(len(tq.GetTombstonedTasks()["default"]), Shou
ldEqual, 0) | |
311 So(len(tq.GetTransactionTasks()["default"]), Sho
uldEqual, 0) | |
312 | |
313 Convey("and reusing a closed context is bad time
s", func() { | |
314 _, err := ttq.Add(nil, "") | |
315 So(err.Error(), ShouldContainSubstring,
"expired") | |
316 }) | |
317 }) | |
318 | |
319 Convey("you can AddMulti as well", func() { | |
320 gae.GetRDS(c).RunInTransaction(func(c context.Co
ntext) error { | |
321 tq := gae.GetTQ(c).(interface { | |
322 gae.TQTestable | |
323 gae.TaskQueue | |
324 }) | |
325 _, err := tq.AddMulti([]*gae.TQTask{t, t
, t}, "") | |
326 So(err, ShouldBeNil) | |
327 So(len(tq.GetScheduledTasks()["default"]
), ShouldEqual, 1) | |
328 So(len(tq.GetTransactionTasks()["default
"]), ShouldEqual, 3) | |
329 return nil | |
330 }, nil) | |
331 So(len(tq.GetScheduledTasks()["default"]), Shoul
dEqual, 4) | |
332 So(len(tq.GetTransactionTasks()["default"]), Sho
uldEqual, 0) | |
333 }) | |
334 | |
335 Convey("unless you add too many things", func() { | |
336 gae.GetRDS(c).RunInTransaction(func(c context.Co
ntext) error { | |
337 for i := 0; i < 5; i++ { | |
338 _, err = gae.GetTQ(c).Add(t, "") | |
339 So(err, ShouldBeNil) | |
340 } | |
341 _, err = gae.GetTQ(c).Add(t, "") | |
342 So(err.Error(), ShouldContainSubstring,
"BAD_REQUEST") | |
343 return nil | |
344 }, nil) | |
345 }) | |
346 | |
347 Convey("unless you Add to a bad queue", func() { | |
348 gae.GetRDS(c).RunInTransaction(func(c context.Co
ntext) error { | |
349 _, err = gae.GetTQ(c).Add(t, "meat") | |
350 So(err.Error(), ShouldContainSubstring,
"UNKNOWN_QUEUE") | |
351 | |
352 Convey("unless you add it!", func() { | |
353 gae.GetTQ(c).(gae.TQTestable).Cr
eateQueue("meat") | |
354 _, err = gae.GetTQ(c).Add(t, "me
at") | |
355 So(err, ShouldBeNil) | |
356 }) | |
357 | |
358 return nil | |
359 }, nil) | |
360 }) | |
361 | |
362 Convey("No other features are available, however", func(
) { | |
363 err := error(nil) | |
364 func() { | |
365 defer func() { err = recover().(error) }
() | |
366 gae.GetRDS(c).RunInTransaction(func(c co
ntext.Context) error { | |
367 gae.GetTQ(c).Delete(t, "") | |
368 return nil | |
369 }, nil) | |
370 }() | |
371 So(err.Error(), ShouldContainSubstring, "TaskQue
ue.Delete") | |
372 }) | |
373 | |
374 Convey("adding a new task only happens if we don't errou
t", func() { | |
375 gae.GetRDS(c).RunInTransaction(func(c context.Co
ntext) error { | |
376 t3 := &gae.TQTask{Path: "/sandwitch/vict
ory"} | |
377 _, err = gae.GetTQ(c).Add(t3, "") | |
378 So(err, ShouldBeNil) | |
379 return fmt.Errorf("nooooo") | |
380 }, nil) | |
381 | |
382 So(tq.GetScheduledTasks()["default"][tEnQ.Name],
ShouldResemble, tEnQ) | |
383 So(tq.GetTombstonedTasks()["default"][tEnQ2.Name
], ShouldResemble, tEnQ2) | |
384 So(tq.GetTransactionTasks()["default"], ShouldBe
Nil) | |
385 }) | |
386 | |
387 Convey("likewise, a panic doesn't schedule anything", fu
nc() { | |
388 func() { | |
389 defer func() { recover() }() | |
390 gae.GetRDS(c).RunInTransaction(func(c co
ntext.Context) error { | |
391 tq := gae.GetTQ(c).(interface { | |
392 gae.TQTestable | |
393 gae.TaskQueue | |
394 }) | |
395 | |
396 t3 := &gae.TQTask{Path: "/sandwi
tch/victory"} | |
397 _, err = tq.Add(t3, "") | |
398 So(err, ShouldBeNil) | |
399 | |
400 panic(fmt.Errorf("nooooo")) | |
401 }, nil) | |
402 }() | |
403 | |
404 So(tq.GetScheduledTasks()["default"][tEnQ.Name],
ShouldResemble, tEnQ) | |
405 So(tq.GetTombstonedTasks()["default"][tEnQ2.Name
], ShouldResemble, tEnQ2) | |
406 So(tq.GetTransactionTasks()["default"], ShouldBe
Nil) | |
407 }) | |
408 | |
409 }) | |
410 }) | |
411 } | |
OLD | NEW |