| index b6b4479a1..6fd238cbb 100644 |
| --- a/run_test.go |
| +++ b/run_test.go |
| @@ -12,6 +12,7 @@ import ( |
| "net/url" |
| "os" |
| "path/filepath" |
| + "slices" |
| "strings" |
| "sync/atomic" |
| "testing" |
| @@ -328,3 +329,356 @@ func TestRunAPICreateRefusesFilesThatDisagreeWithMeta(t *testing.T) { |
| t.Fatal("create accepted more paths than meta files") |
| } |
| } |
| + |
| +// scriptedRun is a runClient that answers pages from a script. |
| +type scriptedRun struct { |
| + pages []runs.Page |
| + pageErr error |
| + expireTo string |
| + expireErr error |
| + cancelErr error |
| + createErr error |
| + ackErrs []error // one per ack call, in order; nil once spent |
| + acked []int64 |
| + cancels int |
| + expires int |
| + created runs.Meta |
| +} |
| + |
| +func (s *scriptedRun) create(_ context.Context, m runs.Meta, _ []string) (runs.Created, error) { |
| + s.created = m |
| + if s.createErr != nil { |
| + return runs.Created{}, s.createErr |
| + } |
| + return runs.Created{ID: "run_a", State: "queued"}, nil |
| +} |
| +func (s *scriptedRun) page(context.Context, string, int64) (runs.Page, error) { |
| + if s.pageErr != nil { |
| + return runs.Page{}, s.pageErr |
| + } |
| + p := s.pages[0] |
| + if len(s.pages) > 1 { |
| + s.pages = s.pages[1:] |
| + } |
| + return p, nil |
| +} |
| +func (s *scriptedRun) cancel(context.Context, string) (string, error) { |
| + s.cancels++ |
| + return "cancel", s.cancelErr |
| +} |
| +func (s *scriptedRun) expire(context.Context, string) (string, error) { |
| + s.expires++ |
| + return s.expireTo, s.expireErr |
| +} |
| +func (s *scriptedRun) ack(_ context.Context, _ string, last int64) error { |
| + s.acked = append(s.acked, last) |
| + if len(s.ackErrs) == 0 { |
| + return nil |
| + } |
| + err := s.ackErrs[0] |
| + s.ackErrs = s.ackErrs[1:] |
| + return err |
| +} |
| + |
| +type termFx struct { |
| + out, err strings.Builder |
| + sig chan os.Signal |
| + now time.Time |
| +} |
| + |
| +func newTermFx() *termFx { |
| + return &termFx{sig: make(chan os.Signal, 2), now: time.Unix(1_700_000_000, 0)} |
| +} |
| + |
| +func (f *termFx) term() runTerm { |
| + return runTerm{stdout: &f.out, stderr: &f.err, interrupts: f.sig, |
| + after: func(d time.Duration) <-chan time.Time { |
| + f.now = f.now.Add(d) |
| + c := make(chan time.Time, 1) |
| + c <- f.now |
| + return c |
| + }, |
| + now: func() time.Time { return f.now }} |
| +} |
| + |
| +func chunk(seq int64, stream, data string) runs.Chunk { |
| + return runs.Chunk{Seq: seq, Stream: stream, Data: []byte(data)} |
| +} |
| + |
| +func TestFollowRunInterleavesAndAcks(t *testing.T) { |
| + f := newTermFx() |
| + s := &scriptedRun{pages: []runs.Page{ |
| + {State: "picked", Chunks: []runs.Chunk{chunk(1, "o", "one\n"), chunk(2, "e", "warn\n")}}, |
| + {State: "picked", Chunks: []runs.Chunk{chunk(3, "o", "two\n")}, Result: &runs.Result{Exit: 0, LastSeq: 3}}, |
| + }} |
| + if code := followRun(context.Background(), s, f.term(), "shop.example.com", "run_a", time.Time{}); code != 0 { |
| + t.Fatalf("exit %d", code) |
| + } |
| + if f.out.String() != "one\ntwo\n" || f.err.String() != "warn\n" { |
| + t.Fatalf("stdout %q stderr %q", f.out.String(), f.err.String()) |
| + } |
| + if len(s.acked) != 1 || s.acked[0] != 3 { |
| + t.Fatalf("acked %v", s.acked) |
| + } |
| +} |
| + |
| +func TestFollowRunPagesUntilTheResult(t *testing.T) { |
| + f := newTermFx() |
| + res := &runs.Result{Exit: 0, LastSeq: 3} |
| + s := &scriptedRun{pages: []runs.Page{ |
| + {State: "done", Chunks: []runs.Chunk{chunk(1, "o", "a")}, Result: res, More: true}, |
| + {State: "done", Chunks: []runs.Chunk{chunk(2, "o", "b")}, Result: res, More: true}, |
| + {State: "done", Chunks: []runs.Chunk{chunk(3, "o", "c")}, Result: res}, |
| + }} |
| + followRun(context.Background(), s, f.term(), "shop.example.com", "run_a", time.Time{}) |
| + if f.out.String() != "abc" || len(s.acked) != 1 || s.acked[0] != 3 { |
| + t.Fatalf("stdout %q acked %v (every page before the ack)", f.out.String(), s.acked) |
| + } |
| +} |
| + |
| +func TestFollowRunExitCodes(t *testing.T) { |
| + started := time.Unix(1_700_000_000, 0) |
| + for _, c := range []struct { |
| + res runs.Result |
| + code int |
| + says string |
| + }{ |
| + {runs.Result{Exit: 7}, 7, ""}, |
| + {runs.Result{Signal: "9"}, 137, ""}, |
| + {runs.Result{TimedOut: true, Started: started, Ended: started.Add(30 * time.Minute)}, 124, "shop.example.com: the command timed out after 30m0s"}, |
| + {runs.Result{Interrupted: "the instance was restarted"}, 125, "shop.example.com: the run was interrupted: the instance was restarted. it may have done part of its work"}, |
| + {runs.Result{Interrupted: runs.InterruptedLost}, 125, "shop.example.com: the run was interrupted: its box stopped reporting. it may have done part of its work"}, |
| + {runs.Result{Cancelled: true, Started: started}, 130, "shop.example.com: the command was cancelled"}, |
| + {runs.Result{Refused: runs.RefusedExpired}, 3, "shop.example.com: not picked up within 60 seconds. the run did not start. try again"}, |
| + {runs.Result{Refused: runs.RefusedFlat}, 3, "shop.example.com: the box is not on the nested layout yet, so carlos run cannot run there"}, |
| + {runs.Result{Refused: runs.RefusedClaim}, 3, "shop.example.com: the box has host claims armed. carlos run cannot run there yet"}, |
| + {runs.Result{Refused: runs.RefusedWake, Detail: "wake failed: restore"}, 3, "shop.example.com: the instance did not wake. wake failed: restore"}, |
| + {runs.Result{Refused: runs.RefusedIncarnation}, 3, "shop.example.com was deleted and created again after this run was requested. the run did not start"}, |
| + {runs.Result{Refused: runs.RefusedImport}, 3, "shop.example.com is an import that is not committed yet. the run did not start"}, |
| + {runs.Result{Exit: 0, Truncated: true}, 0, "shop.example.com: the output passed 64 MB and was cut off there. the command ran on to its end"}, |
| + {runs.Result{Exit: 5, Truncated: true}, 5, "shop.example.com: the output passed 64 MB and was cut off there. the command ran on to its end"}, |
| + } { |
| + code, lines := runExit("shop.example.com", c.res) |
| + if code != c.code || (c.says != "" && !slices.Contains(lines, c.says)) || (c.says == "" && len(lines) != 0) { |
| + t.Errorf("%+v: %d %q, want %d %q", c.res, code, lines, c.code, c.says) |
| + } |
| + } |
| +} |
| + |
| +func TestFollowRunTellsCancelledBeforeStartFromCancelledAfter(t *testing.T) { |
| + f := newTermFx() |
| + s := &scriptedRun{pages: []runs.Page{{State: "cancelled", Result: &runs.Result{Cancelled: true, LastSeq: 0}}}} |
| + if code := followRun(context.Background(), s, f.term(), "shop.example.com", "run_a", time.Time{}); code != 130 { |
| + t.Fatalf("exit %d", code) |
| + } |
| + if got := f.err.String(); got != "shop.example.com: run run_a was cancelled before it started\n" { |
| + t.Fatalf("stderr %q", got) |
| + } |
| + if len(s.acked) != 1 { |
| + t.Fatalf("acked %v", s.acked) |
| + } |
| +} |
| + |
| +func TestFollowRunExpiresAQueuedRunAfterSixtySeconds(t *testing.T) { |
| + f := newTermFx() |
| + s := &scriptedRun{pages: []runs.Page{{State: "queued"}}, expireTo: "expired"} |
| + code := followRun(context.Background(), s, f.term(), "shop.example.com", "run_a", f.now.Add(60*time.Second)) |
| + if code != 3 || s.expires != 1 || !strings.Contains(f.err.String(), "shop.example.com: not picked up within 60 seconds. the run did not start. try again") { |
| + t.Fatalf("exit %d expires %d stderr %q", code, s.expires, f.err.String()) |
| + } |
| + if f.now.Sub(time.Unix(1_700_000_000, 0)) < 60*time.Second { |
| + t.Fatal("expired before 60 seconds") |
| + } |
| +} |
| + |
| +func TestFollowRunKeepsFollowingWhenExpiryLoses(t *testing.T) { |
| + f := newTermFx() |
| + queued := runs.Page{State: "queued"} |
| + pages := make([]runs.Page, 0, 62) |
| + for range 61 { |
| + pages = append(pages, queued) |
| + } |
| + pages = append(pages, runs.Page{State: "done", Result: &runs.Result{Exit: 4}}) |
| + s := &scriptedRun{pages: pages, expireErr: &runConsoleError{status: http.StatusConflict, msg: "run run_a has already started"}} |
| + if code := followRun(context.Background(), s, f.term(), "shop.example.com", "run_a", f.now.Add(60*time.Second)); code != 4 || s.expires != 1 { |
| + t.Fatalf("exit %d expires %d (a late pickup wins over the expiry)", code, s.expires) |
| + } |
| +} |
| + |
| +func TestFollowRunAsksAgainWhenTheExpiryCallFailsToArrive(t *testing.T) { |
| + f := newTermFx() |
| + s := &scriptedRun{pages: []runs.Page{{State: "queued"}}, expireTo: "expired"} |
| + calls := 0 |
| + c := &countingExpire{scriptedRun: s, fail: 2, calls: &calls} |
| + if code := followRun(context.Background(), c, f.term(), "shop.example.com", "run_a", f.now.Add(time.Second)); code != 3 || calls != 3 { |
| + t.Fatalf("exit %d, expire calls %d", code, calls) |
| + } |
| +} |
| + |
| +// countingExpire fails its first `fail` expire calls as a lost connection. |
| +type countingExpire struct { |
| + *scriptedRun |
| + fail int |
| + calls *int |
| +} |
| + |
| +func (c *countingExpire) expire(ctx context.Context, id string) (string, error) { |
| + *c.calls++ |
| + if *c.calls <= c.fail { |
| + return "", errors.New("connection reset") |
| + } |
| + return c.scriptedRun.expire(ctx, id) |
| +} |
| + |
| +func TestFollowRunStatesWithoutAResult(t *testing.T) { |
| + for state, want := range map[string]struct { |
| + code int |
| + says string |
| + }{ |
| + "expired": {3, "shop.example.com: not picked up within 60 seconds. the run did not start. try again"}, |
| + "cancelled": {130, "shop.example.com: run run_a was cancelled before it started"}, |
| + } { |
| + f := newTermFx() |
| + s := &scriptedRun{pages: []runs.Page{{State: state}}} |
| + if code := followRun(context.Background(), s, f.term(), "shop.example.com", "run_a", time.Time{}); code != want.code || !strings.Contains(f.err.String(), want.says) { |
| + t.Errorf("%s: exit %d stderr %q", state, code, f.err.String()) |
| + } |
| + } |
| +} |
| + |
| +func TestFollowRunPageErrors(t *testing.T) { |
| + for name, c := range map[string]struct { |
| + err error |
| + code int |
| + says string |
| + }{ |
| + "no run": {&runConsoleError{status: http.StatusNotFound, msg: "shop.example.com has no run run_a"}, 2, "shop.example.com has no run run_a, or this console has no carlos run yet"}, |
| + "console 4xx": {&runConsoleError{status: http.StatusConflict, msg: "carlos run is not available here"}, 2, "carlos run: carlos run is not available here"}, |
| + "network": {errors.New("connection refused"), 1, "carlos run: connection refused"}, |
| + } { |
| + f := newTermFx() |
| + s := &scriptedRun{pageErr: c.err} |
| + if code := followRun(context.Background(), s, f.term(), "shop.example.com", "run_a", time.Time{}); code != c.code || f.err.String() != c.says+"\n" { |
| + t.Errorf("%s: exit %d stderr %q", name, code, f.err.String()) |
| + } |
| + } |
| +} |
| + |
| +func TestFollowRunAck(t *testing.T) { |
| + notDone := &runConsoleError{status: http.StatusConflict, msg: "run run_a has not finished yet"} |
| + gone := &runConsoleError{status: http.StatusNotFound, msg: "shop.example.com has no run run_a"} |
| + page := runs.Page{State: "done", Result: &runs.Result{Exit: 6, LastSeq: 2}} |
| + t.Run("a lost 200 retried finds the objects gone and is success", func(t *testing.T) { |
| + f := newTermFx() |
| + s := &scriptedRun{pages: []runs.Page{page}, ackErrs: []error{gone}} |
| + if code := followRun(context.Background(), s, f.term(), "shop.example.com", "run_a", time.Time{}); code != 6 || f.err.String() != "" || len(s.acked) != 1 { |
| + t.Fatalf("exit %d stderr %q acked %v", code, f.err.String(), s.acked) |
| + } |
| + }) |
| + t.Run("an ack before the box's done is retried", func(t *testing.T) { |
| + f := newTermFx() |
| + s := &scriptedRun{pages: []runs.Page{page}, ackErrs: []error{notDone, notDone}} |
| + if code := followRun(context.Background(), s, f.term(), "shop.example.com", "run_a", time.Time{}); code != 6 || f.err.String() != "" || len(s.acked) != 3 { |
| + t.Fatalf("exit %d stderr %q acked %v", code, f.err.String(), s.acked) |
| + } |
| + }) |
| + t.Run("a box that never finishes is given up on, the exit code stands", func(t *testing.T) { |
| + f := newTermFx() |
| + errs := make([]error, 100) |
| + for i := range errs { |
| + errs[i] = notDone |
| + } |
| + s := &scriptedRun{pages: []runs.Page{page}, ackErrs: errs} |
| + code := followRun(context.Background(), s, f.term(), "shop.example.com", "run_a", time.Time{}) |
| + if code != 6 || len(s.acked) < 2 || len(s.acked) > 20 || !strings.Contains(f.err.String(), "carlos run: run run_a has not finished yet") { |
| + t.Fatalf("exit %d acked %d stderr %q", code, len(s.acked), f.err.String()) |
| + } |
| + }) |
| + t.Run("a refused ack is said and the exit code stands", func(t *testing.T) { |
| + f := newTermFx() |
| + s := &scriptedRun{pages: []runs.Page{page}, ackErrs: []error{&runConsoleError{status: http.StatusConflict, msg: "run run_a ended at chunk 2, not 3"}}} |
| + if code := followRun(context.Background(), s, f.term(), "shop.example.com", "run_a", time.Time{}); code != 6 || len(s.acked) != 1 || f.err.String() != "carlos run: run run_a ended at chunk 2, not 3\n" { |
| + t.Fatalf("exit %d stderr %q acked %v", code, f.err.String(), s.acked) |
| + } |
| + }) |
| +} |
| + |
| +func TestFollowRunInterrupts(t *testing.T) { |
| + queued := runs.Page{State: "picked"} |
| + done := runs.Page{State: "done", Result: &runs.Result{Cancelled: true, Started: time.Unix(1_700_000_000, 0)}} |
| + t.Run("the first cancels and keeps following", func(t *testing.T) { |
| + f := newTermFx() |
| + f.sig <- os.Interrupt |
| + s := &scriptedRun{pages: []runs.Page{queued, done}} |
| + if code := followRun(context.Background(), s, f.term(), "shop.example.com", "run_a", time.Time{}); code != 130 || s.cancels != 1 { |
| + t.Fatalf("exit %d cancels %d", code, s.cancels) |
| + } |
| + if got := f.err.String(); got != "cancelling run run_a. press Ctrl-C again to stop waiting\nshop.example.com: the command was cancelled\n" { |
| + t.Fatalf("stderr %q", got) |
| + } |
| + }) |
| + t.Run("the second stops waiting", func(t *testing.T) { |
| + f := newTermFx() |
| + f.sig <- os.Interrupt |
| + f.sig <- os.Interrupt |
| + s := &scriptedRun{pages: []runs.Page{queued}} |
| + if code := followRun(context.Background(), s, f.term(), "shop.example.com", "run_a", time.Time{}); code != 130 || s.cancels != 1 || len(s.acked) != 0 { |
| + t.Fatalf("exit %d cancels %d acked %v", code, s.cancels, s.acked) |
| + } |
| + }) |
| + t.Run("a cancel retried after a lost 200 is success", func(t *testing.T) { |
| + f := newTermFx() |
| + f.sig <- os.Interrupt |
| + s := &scriptedRun{pages: []runs.Page{queued, done}, cancelErr: &runConsoleError{status: http.StatusConflict, msg: "run run_a has already ended"}} |
| + if code := followRun(context.Background(), s, f.term(), "shop.example.com", "run_a", time.Time{}); code != 130 { |
| + t.Fatalf("exit %d", code) |
| + } |
| + if got := f.err.String(); strings.Contains(got, "already ended") { |
| + t.Fatalf("stderr %q", got) |
| + } |
| + }) |
| + t.Run("a cancel that failed is said", func(t *testing.T) { |
| + f := newTermFx() |
| + f.sig <- os.Interrupt |
| + s := &scriptedRun{pages: []runs.Page{queued, done}, cancelErr: errors.New("connection reset")} |
| + followRun(context.Background(), s, f.term(), "shop.example.com", "run_a", time.Time{}) |
| + if !strings.Contains(f.err.String(), "carlos run: connection reset") { |
| + t.Fatalf("stderr %q", f.err.String()) |
| + } |
| + }) |
| +} |
| + |
| +func TestStartRunPrintsTheFollowCommand(t *testing.T) { |
| + f := newTermFx() |
| + s := &scriptedRun{pages: []runs.Page{{State: "done", Result: &runs.Result{Exit: 0}}}} |
| + o := runOpts{instance: "shop.example.com", argv: []string{"import"}, timeout: time.Minute} |
| + if code := startRun(context.Background(), s, f.term(), o, "shop"); code != 0 { |
| + t.Fatalf("exit %d", code) |
| + } |
| + if !strings.Contains(f.err.String(), "run run_a on shop.example.com. to follow it again: carlos run --app shop --instance shop.example.com --follow run_a") { |
| + t.Fatalf("stderr %q", f.err.String()) |
| + } |
| + if s.created.TimeoutS != 60 { |
| + t.Fatalf("meta %+v", s.created) |
| + } |
| +} |
| + |
| +func TestStartRunFailures(t *testing.T) { |
| + missing := filepath.Join(t.TempDir(), "absent.csv") |
| + for name, c := range map[string]struct { |
| + o runOpts |
| + createErr error |
| + code int |
| + says string |
| + }{ |
| + "a file that cannot be read": {runOpts{instance: "shop.example.com", argv: []string{"x"}, timeout: time.Minute, files: []string{missing}}, nil, 2, "could not read " + missing}, |
| + "a console refusal": {runOpts{instance: "shop.example.com", argv: []string{"x"}, timeout: time.Minute}, &runConsoleError{status: http.StatusConflict, msg: "an instance can run one command at a time"}, 2, "carlos run: an instance can run one command at a time"}, |
| + "a lost connection": {runOpts{instance: "shop.example.com", argv: []string{"x"}, timeout: time.Minute}, errors.New("connection refused"), 1, "carlos run: connection refused"}, |
| + } { |
| + f := newTermFx() |
| + s := &scriptedRun{createErr: c.createErr} |
| + if code := startRun(context.Background(), s, f.term(), c.o, "shop"); code != c.code || !strings.Contains(f.err.String(), c.says) || strings.Contains(f.err.String(), "to follow it again") { |
| + t.Errorf("%s: exit %d stderr %q", name, code, f.err.String()) |
| + } |
| + } |
| +} |
amadan