amadan

carlos / platform Public

Clone 
git clone https://amadan.net/carlos/platform

Plain git — no account needed to clone.

Download

Download this file

1package console
2
3import (
4 "context"
5 "encoding/json"
6 "errors"
7 "fmt"
8 "net/http"
9 "slices"
10 "strings"
11 "testing"
12 "time"
13
14 "github.com/carlosframework/platform/internal/provision"
15 "github.com/carlosframework/platform/internal/release"
16 "github.com/carlosframework/platform/internal/runs"
17 "github.com/carlosframework/platform/internal/svcrole"
18)
19
20// startRun admits one run through the door and returns its id.
21func (fx *runFx) startRun(t *testing.T) string {
22 t.Helper()
23 status, body := postRun(t, fx.runsURL(runTestHost), fx.member, okMeta(), nil)
24 if status != http.StatusAccepted {
25 t.Fatalf("create: %d %s", status, body)
26 }
27 var c runs.Created
28 if err := json.Unmarshal([]byte(body), &c); err != nil {
29 t.Fatal(err)
30 }
31 return c.ID
32}
33
34func (fx *runFx) transition(t *testing.T, mut func(*provision.Record) error) {
35 t.Helper()
36 if _, err := provision.SetRecordRun(context.Background(), fx.rel, release.OpsAccount, "shop", runTestHost, mut); err != nil {
37 t.Fatal(err)
38 }
39}
40
41func (fx *runFx) chunk(t *testing.T, id string, seq int64, stream, data string) {
42 t.Helper()
43 if _, err := fx.rel.Put(context.Background(), runs.ChunkKey(release.OpsAccount, "shop", id, seq, stream), []byte(data), ""); err != nil {
44 t.Fatal(err)
45 }
46}
47
48func (fx *runFx) page(t *testing.T, id string, after int64) runs.Page {
49 t.Helper()
50 status, _, body := rawReq(t, http.MethodGet, fmt.Sprintf("%s/%s?after=%d", fx.runsURL(runTestHost), id, after), fx.member, "")
51 if status != http.StatusOK {
52 t.Fatalf("GET page: %d %s", status, body)
53 }
54 var p runs.Page
55 if err := json.Unmarshal([]byte(body), &p); err != nil {
56 t.Fatal(err)
57 }
58 return p
59}
60
61func (fx *runFx) post(t *testing.T, id, verb, body string) (int, string) {
62 t.Helper()
63 status, _, out := rawReq(t, http.MethodPost, fx.runsURL(runTestHost)+"/"+id+"/"+verb, fx.member, body)
64 return status, out
65}
66
67func TestAPIRunGetPagesChunksAndTheResult(t *testing.T) {
68 fx := runFixture(t)
69 id := fx.startRun(t)
70 fx.transition(t, provision.RunPick(id, "box-1", "boot-1", time.Now().Add(time.Hour)))
71 fx.chunk(t, id, 1, runs.StreamOut, "hello ")
72 fx.chunk(t, id, 2, runs.StreamErr, "warn")
73 fx.chunk(t, id, 3, runs.StreamOut, "world")
74 p := fx.page(t, id, 0)
75 if p.State != provision.RunPicked || len(p.Chunks) != 3 || p.Result != nil || string(p.Chunks[1].Data) != "warn" || p.Chunks[1].Stream != runs.StreamErr {
76 t.Fatalf("page %+v", p)
77 }
78 if p := fx.page(t, id, 2); len(p.Chunks) != 1 || p.Chunks[0].Seq != 3 {
79 t.Fatalf("after 2: %+v", p)
80 }
81}
82
83func TestAPIRunGetPagesSeveralPagesAfterTheResult(t *testing.T) {
84 fx := runFixture(t)
85 id := fx.startRun(t)
86 fx.transition(t, provision.RunPick(id, "box-1", "boot-1", time.Now().Add(time.Hour)))
87 for seq := int64(1); seq <= 3*runs.PageChunks; seq++ {
88 fx.chunk(t, id, seq, runs.StreamOut, "x")
89 }
90 if err := runs.WriteResult(context.Background(), fx.rel, release.OpsAccount, "shop", id, runs.Result{Exit: 0, LastSeq: 3 * runs.PageChunks}); err != nil {
91 t.Fatal(err)
92 }
93 var after int64
94 pages := 0
95 for {
96 p := fx.page(t, id, after)
97 pages++
98 for _, c := range p.Chunks {
99 if c.Seq != after+1 {
100 t.Fatalf("seq %d after %d", c.Seq, after)
101 }
102 after = c.Seq
103 }
104 if p.Result == nil {
105 t.Fatal("every page must carry the result once it exists")
106 }
107 if !p.More {
108 break
109 }
110 }
111 if after != 3*runs.PageChunks || pages != 3 {
112 t.Fatalf("read to %d in %d pages", after, pages)
113 }
114}
115
116func TestAPIRunGetReadsTheResultBeforeTheChunks(t *testing.T) {
117 fx := runFixture(t)
118 id := fx.startRun(t)
119 fx.transition(t, provision.RunPick(id, "box-1", "boot-1", time.Now().Add(time.Hour)))
120 fx.chunk(t, id, 1, runs.StreamOut, "a")
121 // The box writes the last chunk, then the result. A store that shows
122 // the result to the console's read must also show every chunk to the
123 // listing after it.
124 order := []string{}
125 fx.srv.cfg.Releases = &orderedStore{Store: fx.rel, seen: &order}
126 fx.page(t, id, 0)
127 result, list := -1, -1
128 for i, call := range order {
129 if result < 0 && strings.HasSuffix(call, "/result.json") {
130 result = i
131 }
132 if list < 0 && strings.HasPrefix(call, "list ") && strings.HasSuffix(call, "/out/") {
133 list = i
134 }
135 }
136 if result < 0 || list < 0 || result > list {
137 t.Fatalf("store calls %q: the result must be read before the chunks are listed", order)
138 }
139}
140
141// orderedStore records Get and List calls in order.
142type orderedStore struct {
143 release.Store
144 seen *[]string
145}
146
147func (o *orderedStore) Get(ctx context.Context, key string) ([]byte, string, error) {
148 *o.seen = append(*o.seen, "get "+key)
149 return o.Store.Get(ctx, key)
150}
151func (o *orderedStore) List(ctx context.Context, prefix string) ([]string, error) {
152 *o.seen = append(*o.seen, "list "+prefix)
153 return o.Store.List(ctx, prefix)
154}
155
156func TestAPIRunAckChecksLastSeqThenCleansUp(t *testing.T) {
157 fx := runFixture(t)
158 id := fx.startRun(t)
159 if status, body := fx.post(t, id, "ack", `{"last_seq":0}`); status != http.StatusConflict || errorOf(t, body) != "run "+id+" has not finished yet" {
160 t.Fatalf("ack before the result: %d %s", status, body)
161 }
162 fx.transition(t, provision.RunPick(id, "box-1", "boot-1", time.Now().Add(time.Hour)))
163 fx.chunk(t, id, 1, runs.StreamOut, "a")
164 if err := runs.WriteResult(context.Background(), fx.rel, release.OpsAccount, "shop", id, runs.Result{Exit: 0, LastSeq: 1}); err != nil {
165 t.Fatal(err)
166 }
167 fx.transition(t, provision.RunFinish(id))
168 if status, body := fx.post(t, id, "ack", `{"last_seq":0}`); status != http.StatusConflict || errorOf(t, body) != "run "+id+" ended at chunk 1, not 0" {
169 t.Fatalf("wrong last_seq: %d %s", status, body)
170 }
171 if left, _ := fx.rel.List(context.Background(), runs.Prefix(release.OpsAccount, "shop", id)); len(left) == 0 {
172 t.Fatal("a refused ack deleted the run")
173 }
174 if status, body := fx.post(t, id, "ack", `{"last_seq":1}`); status != http.StatusOK {
175 t.Fatalf("ack: %d %s", status, body)
176 }
177 if left, _ := fx.rel.List(context.Background(), runs.Prefix(release.OpsAccount, "shop", id)); len(left) != 0 {
178 t.Fatalf("ack left %q", left)
179 }
180 if rec, _ := provision.GetRecord(context.Background(), fx.rel, release.OpsAccount, "shop", runTestHost); rec.ActiveRun != nil {
181 t.Fatalf("ack left the run on the record: %+v", rec.ActiveRun)
182 }
183}
184
185func TestAPIRunCancelBeforeAndAfterPickup(t *testing.T) {
186 fx := runFixture(t)
187 id := fx.startRun(t)
188 if status, body := fx.post(t, id, "cancel", ""); status != http.StatusOK || !strings.Contains(body, `"cancelled"`) {
189 t.Fatalf("queued cancel: %d %s", status, body)
190 }
191 if status, body := fx.post(t, id, "cancel", ""); status != http.StatusConflict || errorOf(t, body) != "run "+id+" has already ended" {
192 t.Fatalf("cancel a cancelled run: %d %s", status, body)
193 }
194 id2 := fx.startRun(t)
195 fx.transition(t, provision.RunPick(id2, "box-1", "boot-1", time.Now().Add(time.Hour)))
196 if status, body := fx.post(t, id2, "cancel", ""); status != http.StatusOK || !strings.Contains(body, `"cancel"`) {
197 t.Fatalf("picked cancel: %d %s", status, body)
198 }
199 events, _ := fx.st.Events(OpsAccountID, "shop", 20)
200 ends := 0
201 for _, e := range events {
202 if e.Action == "run-end" && strings.Contains(e.Detail, id) && strings.Contains(e.Detail, `"cancelled"`) {
203 ends++
204 }
205 }
206 if ends != 1 {
207 t.Errorf("a run cancelled before pickup must write one run-end row, got %d", ends)
208 }
209}
210
211func TestAPIRunExpireRacesAPickup(t *testing.T) {
212 fx := runFixture(t)
213 id := fx.startRun(t)
214 if status, body := fx.post(t, id, "expire", ""); status != http.StatusOK || !strings.Contains(body, `"expired"`) {
215 t.Fatalf("expire: %d %s", status, body)
216 }
217 id2 := fx.startRun(t)
218 fx.transition(t, provision.RunPick(id2, "box-1", "boot-1", time.Now().Add(time.Hour)))
219 if status, body := fx.post(t, id2, "expire", ""); status != http.StatusConflict || errorOf(t, body) != "run "+id2+" has already started" {
220 t.Fatalf("expire after pickup: %d %s", status, body)
221 }
222}
223
224func TestAPIRunRoutesNameAMissingRun(t *testing.T) {
225 fx := runFixture(t)
226 missing := "run_bbbbbbbbbbbbbbbbbbbbbbbbbb"
227 status, _, body := rawReq(t, http.MethodGet, fx.runsURL(runTestHost)+"/"+missing, fx.member, "")
228 if status != http.StatusNotFound || errorOf(t, body) != runTestHost+" has no run "+missing {
229 t.Fatalf("GET: %d %s", status, body)
230 }
231 for _, verb := range []string{"cancel", "expire", "ack"} {
232 if status, body := fx.post(t, missing, verb, `{"last_seq":0}`); status != http.StatusNotFound || errorOf(t, body) != runTestHost+" has no run "+missing {
233 t.Errorf("%s: %d %s", verb, status, body)
234 }
235 }
236}
237
238func TestAPIRunGetWritesRunEndOnce(t *testing.T) {
239 fx := runFixture(t)
240 id := fx.startRun(t)
241 fx.transition(t, provision.RunPick(id, "box-1", "boot-1", time.Now().Add(time.Hour)))
242 if err := runs.WriteResult(context.Background(), fx.rel, release.OpsAccount, "shop", id, runs.Result{Exit: 7, Box: "box-1"}); err != nil {
243 t.Fatal(err)
244 }
245 fx.page(t, id, 0)
246 fx.page(t, id, 0)
247 events, _ := fx.st.Events(OpsAccountID, "shop", 20)
248 n := 0
249 for _, e := range events {
250 if e.Action == "run-end" && strings.Contains(e.Detail, id) {
251 n++
252 if !strings.Contains(e.Detail, `"exit":7`) || e.Address != "alice@keymail.dev" {
253 t.Errorf("run-end %+v", e)
254 }
255 }
256 }
257 if n != 1 {
258 t.Fatalf("run-end rows %d, want 1", n)
259 }
260}
261
262func TestAPIRunAckKeepsTheObjectsOfALiveRun(t *testing.T) {
263 fx := runFixture(t)
264 id := fx.startRun(t)
265 fx.transition(t, provision.RunPick(id, "box-1", "boot-1", time.Now().Add(time.Hour)))
266 fx.chunk(t, id, 1, runs.StreamOut, "a")
267 // The box has written the result and not yet moved the run to done.
268 if err := runs.WriteResult(context.Background(), fx.rel, release.OpsAccount, "shop", id, runs.Result{Exit: 0, LastSeq: 1}); err != nil {
269 t.Fatal(err)
270 }
271 if status, body := fx.post(t, id, "ack", `{"last_seq":1}`); status != http.StatusConflict || errorOf(t, body) != "run "+id+" has not finished yet" {
272 t.Fatalf("ack of a picked run: %d %s", status, body)
273 }
274 if left, _ := fx.rel.List(context.Background(), runs.Prefix(release.OpsAccount, "shop", id)); len(left) == 0 {
275 t.Fatal("an ack deleted the objects of a run the record holds as picked")
276 }
277 if rec, _ := provision.GetRecord(context.Background(), fx.rel, release.OpsAccount, "shop", runTestHost); rec.ActiveRun == nil || rec.ActiveRun.State != provision.RunPicked {
278 t.Fatalf("record %+v", rec.ActiveRun)
279 }
280}
281
282func TestAPIRunAckAfterTheRecordMovedOnDeletesTheObjects(t *testing.T) {
283 fx := runFixture(t)
284 id := fx.startRun(t)
285 fx.transition(t, provision.RunPick(id, "box-1", "boot-1", time.Now().Add(time.Hour)))
286 if err := runs.WriteResult(context.Background(), fx.rel, release.OpsAccount, "shop", id, runs.Result{Exit: 0}); err != nil {
287 t.Fatal(err)
288 }
289 fx.transition(t, provision.RunFinish(id))
290 fx.transition(t, provision.RunClear(id))
291 // A repeated ack, the record already clear: the objects go, the answer stands.
292 if status, body := fx.post(t, id, "ack", `{"last_seq":0}`); status != http.StatusOK {
293 t.Fatalf("ack: %d %s", status, body)
294 }
295 if left, _ := fx.rel.List(context.Background(), runs.Prefix(release.OpsAccount, "shop", id)); len(left) != 0 {
296 t.Fatalf("left %q", left)
297 }
298}
299
300func TestAPIRunRoutesRefuseAnotherHostsRun(t *testing.T) {
301 fx := runFixture(t)
302 id := fx.startRun(t)
303 other := "other.example.com"
304 putShopRecord(t, fx.consoleFixture, provision.Record{Host: other, Channel: "stable", CreatedAt: time.Now().UTC()})
305 if err := runs.WriteResult(context.Background(), fx.rel, release.OpsAccount, "shop", id, runs.Result{Exit: 0}); err != nil {
306 t.Fatal(err)
307 }
308 want := other + " has no run " + id
309 status, _, body := rawReq(t, http.MethodGet, fx.runsURL(other)+"/"+id, fx.member, "")
310 if status != http.StatusNotFound || errorOf(t, body) != want {
311 t.Errorf("GET: %d %s", status, body)
312 }
313 for _, verb := range []string{"cancel", "expire", "ack"} {
314 _, _, body := rawReq(t, http.MethodPost, fx.runsURL(other)+"/"+id+"/"+verb, fx.member, `{"last_seq":0}`)
315 if errorOf(t, body) != want {
316 t.Errorf("%s: %s", verb, body)
317 }
318 }
319 if left, _ := fx.rel.List(context.Background(), runs.Prefix(release.OpsAccount, "shop", id)); len(left) == 0 {
320 t.Fatal("another host's ack deleted the run")
321 }
322}
323
324func TestAPIRunRoutesRefuseAServiceCredentialAndABadID(t *testing.T) {
325 fx := runFixture(t)
326 svc := serviceToken(t, fx.st, OpsAccountID, "ci", "shop", svcrole.Instance)
327 id := "run_aaaaaaaaaaaaaaaaaaaaaaaaaa"
328 if status, _, _ := rawReq(t, http.MethodGet, fx.runsURL(runTestHost)+"/"+id, svc, ""); status != http.StatusForbidden {
329 t.Errorf("service credential: %d, want 403", status)
330 }
331 status, _, body := rawReq(t, http.MethodGet, fx.runsURL(runTestHost)+"/not-a-run", fx.member, "")
332 if status != http.StatusNotFound || errorOf(t, body) != runTestHost+" has no run not-a-run" {
333 t.Errorf("bad id: %d %s", status, body)
334 }
335}
336
337// faultStore injects the failures a bucket has: a delete that fails, a Put
338// whose response is lost (landed or not), a Get that fails.
339type faultStore struct {
340 release.Store
341 delete func(key string) error
342 put func(key string) (landed bool, err error)
343 get func(key string) error
344}
345
346func (f *faultStore) Delete(ctx context.Context, key string) error {
347 if f.delete != nil {
348 if err := f.delete(key); err != nil {
349 return err
350 }
351 }
352 return f.Store.Delete(ctx, key)
353}
354
355func (f *faultStore) Put(ctx context.Context, key string, body []byte, ifMatch string) (string, error) {
356 if f.put != nil {
357 if landed, err := f.put(key); err != nil {
358 if landed {
359 _, _ = f.Store.Put(ctx, key, body, ifMatch)
360 }
361 return "", err
362 }
363 }
364 return f.Store.Put(ctx, key, body, ifMatch)
365}
366
367func (f *faultStore) Get(ctx context.Context, key string) ([]byte, string, error) {
368 if f.get != nil {
369 if err := f.get(key); err != nil {
370 return nil, "", err
371 }
372 }
373 return f.Store.Get(ctx, key)
374}
375
376// runEnds is the details of the run-end rows written for id.
377func (fx *runFx) runEnds(t *testing.T, id string) []string {
378 t.Helper()
379 events, err := fx.st.Events(OpsAccountID, "shop", 50)
380 if err != nil {
381 t.Fatal(err)
382 }
383 var out []string
384 for _, e := range events {
385 if e.Action == "run-end" && strings.Contains(e.Detail, id) {
386 out = append(out, e.Detail)
387 }
388 }
389 return out
390}
391
392func (fx *runFx) left(id string) []string {
393 left, _ := fx.rel.List(context.Background(), runs.Prefix(release.OpsAccount, "shop", id))
394 return left
395}
396
397// finishedRun is a picked run with one chunk and a result, finished on the record.
398func (fx *runFx) finishedRun(t *testing.T) string {
399 t.Helper()
400 id := fx.startRun(t)
401 fx.transition(t, provision.RunPick(id, "box-1", "boot-1", time.Now().Add(time.Hour)))
402 fx.chunk(t, id, 1, runs.StreamOut, "a")
403 if err := runs.WriteResult(context.Background(), fx.rel, release.OpsAccount, "shop", id, runs.Result{Exit: 0, LastSeq: 1}); err != nil {
404 t.Fatal(err)
405 }
406 fx.transition(t, provision.RunFinish(id))
407 return id
408}
409
410func TestAPIRunAckThatCannotDeleteKeepsTheMarkerSoTheRowIsWrittenOnce(t *testing.T) {
411 fx := runFixture(t)
412 id := fx.finishedRun(t)
413 failing := true
414 fx.srv.cfg.Releases = &faultStore{Store: fx.rel, delete: func(key string) error {
415 if failing && key == runs.ResultKey(release.OpsAccount, "shop", id) {
416 return errors.New("delete refused")
417 }
418 return nil
419 }}
420 if status, body := fx.post(t, id, "ack", `{"last_seq":1}`); status != http.StatusOK {
421 t.Fatalf("ack: %d %s", status, body)
422 }
423 left := fx.left(id)
424 for _, want := range []string{runs.ResultKey(release.OpsAccount, "shop", id), runs.RequestKey(release.OpsAccount, "shop", id), runs.AuditedKey(release.OpsAccount, "shop", id)} {
425 if !slices.Contains(left, want) {
426 t.Fatalf("after a failed delete %q is gone; left %q", want, left)
427 }
428 }
429 // A follower and the retried ack both find the result; neither writes a second row.
430 failing = false
431 if p := fx.page(t, id, 1); p.Result == nil {
432 t.Fatalf("page %+v", p)
433 }
434 if status, body := fx.post(t, id, "ack", `{"last_seq":1}`); status != http.StatusOK {
435 t.Fatalf("retried ack: %d %s", status, body)
436 }
437 if left := fx.left(id); len(left) != 0 {
438 t.Fatalf("retried ack left %q", left)
439 }
440 if rows := fx.runEnds(t, id); len(rows) != 1 {
441 t.Fatalf("run-end rows %q, want 1", rows)
442 }
443}
444
445func TestAPIRunEndIsNotWrittenForARunAnAckHasDeleted(t *testing.T) {
446 fx := runFixture(t)
447 id := fx.finishedRun(t)
448 if status, body := fx.post(t, id, "ack", `{"last_seq":1}`); status != http.StatusOK {
449 t.Fatalf("ack: %d %s", status, body)
450 }
451 // A follower that read the result before the ack's delete reaches the
452 // audit after it.
453 fx.srv.auditRunEnd(context.Background(), &Account{ID: OpsAccountID, Sqid: release.OpsAccount}, "shop", runTestHost, id, "result", &runs.Result{LastSeq: 1})
454 if left := fx.left(id); len(left) != 0 {
455 t.Fatalf("a late audit left %q", left)
456 }
457 if rows := fx.runEnds(t, id); len(rows) != 1 {
458 t.Fatalf("run-end rows %q, want 1", rows)
459 }
460}
461
462func TestAPIRunCancelAndExpireBeforePickupLeaveAResult(t *testing.T) {
463 fx := runFixture(t)
464 cancelled := fx.startRun(t)
465 if status, body := fx.post(t, cancelled, "cancel", ""); status != http.StatusOK {
466 t.Fatalf("cancel: %d %s", status, body)
467 }
468 fx.transition(t, provision.RunClear(cancelled))
469 expiredID := fx.startRun(t)
470 if status, body := fx.post(t, expiredID, "expire", ""); status != http.StatusOK {
471 t.Fatalf("expire: %d %s", status, body)
472 }
473 fx.transition(t, provision.RunClear(expiredID))
474 // The record no longer names either run; a follower still learns how it ended.
475 if p := fx.page(t, cancelled, 0); p.State != "" || p.Result == nil || !p.Result.Cancelled || p.More || len(p.Chunks) != 0 {
476 t.Errorf("cancelled run page %+v", p)
477 }
478 if p := fx.page(t, expiredID, 0); p.State != "" || p.Result == nil || p.Result.Refused != runs.RefusedExpired {
479 t.Errorf("expired run page %+v", p)
480 }
481 if status, body := fx.post(t, cancelled, "ack", `{"last_seq":0}`); status != http.StatusOK {
482 t.Errorf("ack of a cancelled run: %d %s", status, body)
483 }
484 if left := fx.left(cancelled); len(left) != 0 {
485 t.Errorf("left %q", left)
486 }
487}
488
489func TestAPIRunExpireWritesTheRunEndRow(t *testing.T) {
490 fx := runFixture(t)
491 id := fx.startRun(t)
492 if status, body := fx.post(t, id, "expire", ""); status != http.StatusOK {
493 t.Fatalf("expire: %d %s", status, body)
494 }
495 rows := fx.runEnds(t, id)
496 if len(rows) != 1 || !strings.Contains(rows[0], `"outcome":"expired"`) {
497 t.Fatalf("run-end rows %q", rows)
498 }
499}
500
501// A cancel or expire whose CAS landed but whose response was lost is retried
502// and finds its own terminal state; the result and the row are still written.
503func TestAPIRunRetriedCancelAndExpireStillEndTheRun(t *testing.T) {
504 fx := runFixture(t)
505 for _, c := range []struct {
506 verb, outcome string
507 mut func(string) func(*provision.Record) error
508 }{
509 {"cancel", "cancelled", provision.RunCancelTransition},
510 {"expire", "expired", provision.RunExpire},
511 } {
512 id := fx.startRun(t)
513 fx.transition(t, c.mut(id))
514 if status, body := fx.post(t, id, c.verb, ""); status != http.StatusConflict || errorOf(t, body) != "run "+id+" has already ended" {
515 t.Fatalf("%s retry: %d %s", c.verb, status, body)
516 }
517 if res, err := runs.ReadResult(context.Background(), fx.rel, release.OpsAccount, "shop", id); err != nil || res == nil {
518 t.Errorf("%s retry left no result: %+v, %v", c.verb, res, err)
519 }
520 if rows := fx.runEnds(t, id); len(rows) != 1 || !strings.Contains(rows[0], `"outcome":"`+c.outcome+`"`) {
521 t.Errorf("%s retry rows %q", c.verb, rows)
522 }
523 fx.post(t, id, c.verb, "")
524 if rows := fx.runEnds(t, id); len(rows) != 1 {
525 t.Errorf("%s retried twice: rows %q", c.verb, rows)
526 }
527 fx.transition(t, provision.RunClear(id))
528 }
529}
530
531func TestAPIRunAckKeepsTheObjectsWhenTheClearIsNotSettled(t *testing.T) {
532 recordKey := provision.RecordKey(release.OpsAccount, "shop", runTestHost)
533 for _, c := range []struct {
534 name string
535 setup func(*faultStore, *bool)
536 wantStatus int
537 wantKept bool
538 }{
539 {"clear refused", func(f *faultStore, _ *bool) {
540 f.put = func(key string) (bool, error) {
541 if key == recordKey {
542 return false, errors.New("put refused")
543 }
544 return false, nil
545 }
546 }, http.StatusBadGateway, true},
547 {"clear landed, response lost", func(f *faultStore, _ *bool) {
548 f.put = func(key string) (bool, error) {
549 if key == recordKey {
550 return true, errors.New("response lost")
551 }
552 return false, nil
553 }
554 }, http.StatusOK, false},
555 {"clear landed, re-read fails", func(f *faultStore, afterPut *bool) {
556 f.put = func(key string) (bool, error) {
557 if key == recordKey {
558 *afterPut = true
559 }
560 return false, nil
561 }
562 f.get = func(key string) error {
563 if key == recordKey && *afterPut {
564 return errors.New("get refused")
565 }
566 return nil
567 }
568 }, http.StatusBadGateway, true},
569 } {
570 t.Run(c.name, func(t *testing.T) {
571 fx := runFixture(t)
572 id := fx.finishedRun(t)
573 f := &faultStore{Store: fx.rel}
574 var afterPut bool
575 c.setup(f, &afterPut)
576 fx.srv.cfg.Releases = f
577 status, body := fx.post(t, id, "ack", `{"last_seq":1}`)
578 if status != c.wantStatus {
579 t.Fatalf("ack: %d %s, want %d", status, body, c.wantStatus)
580 }
581 if kept := len(fx.left(id)) != 0; kept != c.wantKept {
582 t.Fatalf("objects kept = %v, want %v (left %q)", kept, c.wantKept, fx.left(id))
583 }
584 })
585 }
586}
587