Pin version of launchy gem while we still support Ruby 2.3
[arvados.git] / services / crunch-dispatch-local / crunch-dispatch-local_test.go
1 // Copyright (C) The Arvados Authors. All rights reserved.
2 //
3 // SPDX-License-Identifier: AGPL-3.0
4
5 package main
6
7 import (
8         "bytes"
9         "context"
10         "io"
11         "net/http"
12         "net/http/httptest"
13         "os"
14         "os/exec"
15         "regexp"
16         "testing"
17         "time"
18
19         "git.arvados.org/arvados.git/sdk/go/arvados"
20         "git.arvados.org/arvados.git/sdk/go/arvadosclient"
21         "git.arvados.org/arvados.git/sdk/go/arvadostest"
22         "git.arvados.org/arvados.git/sdk/go/dispatch"
23         "github.com/sirupsen/logrus"
24         . "gopkg.in/check.v1"
25 )
26
27 // Gocheck boilerplate
28 func Test(t *testing.T) {
29         TestingT(t)
30 }
31
32 var _ = Suite(&TestSuite{})
33 var _ = Suite(&MockArvadosServerSuite{})
34
35 type TestSuite struct{}
36 type MockArvadosServerSuite struct{}
37
38 var initialArgs []string
39
40 func (s *TestSuite) SetUpSuite(c *C) {
41         initialArgs = os.Args
42         arvadostest.StartAPI()
43         runningCmds = make(map[string]*exec.Cmd)
44         logrus.SetFormatter(&logrus.TextFormatter{DisableColors: true})
45 }
46
47 func (s *TestSuite) TearDownSuite(c *C) {
48         arvadostest.StopAPI()
49 }
50
51 func (s *TestSuite) SetUpTest(c *C) {
52         args := []string{"crunch-dispatch-local"}
53         os.Args = args
54 }
55
56 func (s *TestSuite) TearDownTest(c *C) {
57         arvadostest.ResetEnv()
58         os.Args = initialArgs
59 }
60
61 func (s *MockArvadosServerSuite) TearDownTest(c *C) {
62         arvadostest.ResetEnv()
63 }
64
65 func (s *TestSuite) TestIntegration(c *C) {
66         arv, err := arvadosclient.MakeArvadosClient()
67         c.Assert(err, IsNil)
68
69         echo := "echo"
70         crunchRunCommand = &echo
71
72         ctx, cancel := context.WithCancel(context.Background())
73         dispatcher := dispatch.Dispatcher{
74                 Arv:        arv,
75                 PollPeriod: time.Second,
76         }
77
78         startCmd := func(container arvados.Container, cmd *exec.Cmd) error {
79                 dispatcher.UpdateState(container.UUID, "Running")
80                 dispatcher.UpdateState(container.UUID, "Complete")
81                 return cmd.Start()
82         }
83
84         dispatcher.RunContainer = func(d *dispatch.Dispatcher, c arvados.Container, s <-chan arvados.Container) {
85                 (&LocalRun{startCmd, make(chan bool, 8), ctx}).run(d, c, s)
86                 cancel()
87         }
88
89         err = dispatcher.Run(ctx)
90         c.Assert(err, Equals, context.Canceled)
91
92         // Wait for all running crunch jobs to complete / terminate
93         waitGroup.Wait()
94
95         // There should be no queued containers now
96         params := arvadosclient.Dict{
97                 "filters": [][]string{{"state", "=", "Queued"}},
98         }
99         var containers arvados.ContainerList
100         err = arv.List("containers", params, &containers)
101         c.Check(err, IsNil)
102         c.Assert(len(containers.Items), Equals, 0)
103
104         // Previously "Queued" container should now be in "Complete" state
105         var container arvados.Container
106         err = arv.Get("containers", "zzzzz-dz642-queuedcontainer", nil, &container)
107         c.Check(err, IsNil)
108         c.Check(string(container.State), Equals, "Complete")
109 }
110
111 func (s *MockArvadosServerSuite) Test_APIErrorGettingContainers(c *C) {
112         apiStubResponses := make(map[string]arvadostest.StubResponse)
113         apiStubResponses["/arvados/v1/containers"] = arvadostest.StubResponse{500, string(`{}`)}
114
115         testWithServerStub(c, apiStubResponses, "echo", "error getting count of containers")
116 }
117
118 func (s *MockArvadosServerSuite) Test_APIErrorUpdatingContainerState(c *C) {
119         apiStubResponses := make(map[string]arvadostest.StubResponse)
120         apiStubResponses["/arvados/v1/containers"] =
121                 arvadostest.StubResponse{200, string(`{"items_available":1, "items":[{"uuid":"zzzzz-dz642-xxxxxxxxxxxxxx1","State":"Queued","Priority":1}]}`)}
122         apiStubResponses["/arvados/v1/containers/zzzzz-dz642-xxxxxxxxxxxxxx1"] =
123                 arvadostest.StubResponse{500, string(`{}`)}
124
125         testWithServerStub(c, apiStubResponses, "echo", "error locking container zzzzz-dz642-xxxxxxxxxxxxxx1")
126 }
127
128 func (s *MockArvadosServerSuite) Test_ContainerStillInRunningAfterRun(c *C) {
129         apiStubResponses := make(map[string]arvadostest.StubResponse)
130         apiStubResponses["/arvados/v1/containers"] =
131                 arvadostest.StubResponse{200, string(`{"items_available":1, "items":[{"uuid":"zzzzz-dz642-xxxxxxxxxxxxxx2","State":"Queued","Priority":1}]}`)}
132         apiStubResponses["/arvados/v1/containers/zzzzz-dz642-xxxxxxxxxxxxxx2/lock"] =
133                 arvadostest.StubResponse{200, string(`{"uuid":"zzzzz-dz642-xxxxxxxxxxxxxx2", "state":"Locked", "priority":1, "locked_by_uuid": "` + arvadostest.Dispatch1AuthUUID + `"}`)}
134         apiStubResponses["/arvados/v1/containers/zzzzz-dz642-xxxxxxxxxxxxxx2"] =
135                 arvadostest.StubResponse{200, string(`{"uuid":"zzzzz-dz642-xxxxxxxxxxxxxx2", "state":"Running", "priority":1, "locked_by_uuid": "` + arvadostest.Dispatch1AuthUUID + `"}`)}
136
137         testWithServerStub(c, apiStubResponses, "echo",
138                 `after \\"echo\\" process termination, container state for zzzzz-dz642-xxxxxxxxxxxxxx2 is \\"Running\\"; updating it to \\"Cancelled\\"`)
139 }
140
141 func (s *MockArvadosServerSuite) Test_ErrorRunningContainer(c *C) {
142         apiStubResponses := make(map[string]arvadostest.StubResponse)
143         apiStubResponses["/arvados/v1/containers"] =
144                 arvadostest.StubResponse{200, string(`{"items_available":1, "items":[{"uuid":"zzzzz-dz642-xxxxxxxxxxxxxx3","State":"Queued","Priority":1}]}`)}
145
146         apiStubResponses["/arvados/v1/containers/zzzzz-dz642-xxxxxxxxxxxxxx3/lock"] =
147                 arvadostest.StubResponse{200, string(`{"uuid":"zzzzz-dz642-xxxxxxxxxxxxxx3", "state":"Locked", "priority":1}`)}
148
149         testWithServerStub(c, apiStubResponses, "nosuchcommand", `error starting \\"nosuchcommand\\" for zzzzz-dz642-xxxxxxxxxxxxxx3`)
150 }
151
152 func testWithServerStub(c *C, apiStubResponses map[string]arvadostest.StubResponse, crunchCmd string, expected string) {
153         apiStubResponses["/arvados/v1/api_client_authorizations/current"] =
154                 arvadostest.StubResponse{200, string(`{"uuid": "` + arvadostest.Dispatch1AuthUUID + `", "api_token": "xyz"}`)}
155
156         apiStub := arvadostest.ServerStub{apiStubResponses}
157
158         api := httptest.NewServer(&apiStub)
159         defer api.Close()
160
161         arv := &arvadosclient.ArvadosClient{
162                 Scheme:    "http",
163                 ApiServer: api.URL[7:],
164                 ApiToken:  "abc123",
165                 Client:    &http.Client{Transport: &http.Transport{}},
166                 Retries:   0,
167         }
168
169         buf := bytes.NewBuffer(nil)
170         logrus.SetOutput(io.MultiWriter(buf, os.Stderr))
171         defer logrus.SetOutput(os.Stderr)
172
173         *crunchRunCommand = crunchCmd
174
175         ctx, cancel := context.WithCancel(context.Background())
176         dispatcher := dispatch.Dispatcher{
177                 Arv:        arv,
178                 PollPeriod: time.Second / 20,
179         }
180
181         startCmd := func(container arvados.Container, cmd *exec.Cmd) error {
182                 dispatcher.UpdateState(container.UUID, "Running")
183                 dispatcher.UpdateState(container.UUID, "Complete")
184                 return cmd.Start()
185         }
186
187         dispatcher.RunContainer = func(d *dispatch.Dispatcher, c arvados.Container, s <-chan arvados.Container) {
188                 (&LocalRun{startCmd, make(chan bool, 8), ctx}).run(d, c, s)
189                 cancel()
190         }
191
192         re := regexp.MustCompile(`(?ms).*` + expected + `.*`)
193         go func() {
194                 for i := 0; i < 80 && !re.MatchString(buf.String()); i++ {
195                         time.Sleep(100 * time.Millisecond)
196                 }
197                 cancel()
198         }()
199
200         err := dispatcher.Run(ctx)
201         c.Assert(err, Equals, context.Canceled)
202
203         // Wait for all running crunch jobs to complete / terminate
204         waitGroup.Wait()
205
206         c.Check(buf.String(), Matches, `(?ms).*`+expected+`.*`)
207 }