services/crunch-dispatch-local/crunch-dispatch-local_test.go

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