services/keepstore/pull_worker_integration_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         "errors"
  11         "io"
  12         "io/ioutil"
  13         "strings"
  14
  15         "git.curoverse.com/arvados.git/sdk/go/arvadostest"
  16         "git.curoverse.com/arvados.git/sdk/go/keepclient"
  17         "github.com/prometheus/client_golang/prometheus"
  18         check "gopkg.in/check.v1"
  19 )
  20
  21 type PullWorkIntegrationTestData struct {
  22         Name     string
  23         Locator  string
  24         Content  string
  25         GetError string
  26 }
  27
  28 func (s *HandlerSuite) setupPullWorkerIntegrationTest(c *check.C, testData PullWorkIntegrationTestData, wantData bool) PullRequest {
  29         arvadostest.StartKeep(2, false)
  30         c.Assert(s.handler.setup(context.Background(), s.cluster, "", prometheus.NewRegistry(), testServiceURL), check.IsNil)
  31         // Put content if the test needs it
  32         if wantData {
  33                 locator, _, err := s.handler.keepClient.PutB([]byte(testData.Content))
  34                 if err != nil {
  35                         c.Errorf("Error putting test data in setup for %s %s %v", testData.Content, locator, err)
  36                 }
  37                 if locator == "" {
  38                         c.Errorf("No locator found after putting test data")
  39                 }
  40         }
  41
  42         // Create pullRequest for the test
  43         pullRequest := PullRequest{
  44                 Locator: testData.Locator,
  45         }
  46         return pullRequest
  47 }
  48
  49 // Do a get on a block that is not existing in any of the keep servers.
  50 // Expect "block not found" error.
  51 func (s *HandlerSuite) TestPullWorkerIntegration_GetNonExistingLocator(c *check.C) {
  52         c.Assert(s.handler.setup(context.Background(), s.cluster, "", prometheus.NewRegistry(), testServiceURL), check.IsNil)
  53         testData := PullWorkIntegrationTestData{
  54                 Name:     "TestPullWorkerIntegration_GetLocator",
  55                 Locator:  "5d41402abc4b2a76b9719d911017c592",
  56                 Content:  "hello",
  57                 GetError: "Block not found",
  58         }
  59
  60         pullRequest := s.setupPullWorkerIntegrationTest(c, testData, false)
  61         defer arvadostest.StopAPI()
  62         defer arvadostest.StopKeep(2)
  63
  64         s.performPullWorkerIntegrationTest(testData, pullRequest, c)
  65 }
  66
  67 // Do a get on a block that exists on one of the keep servers.
  68 // The setup method will create this block before doing the get.
  69 func (s *HandlerSuite) TestPullWorkerIntegration_GetExistingLocator(c *check.C) {
  70         c.Assert(s.handler.setup(context.Background(), s.cluster, "", prometheus.NewRegistry(), testServiceURL), check.IsNil)
  71         testData := PullWorkIntegrationTestData{
  72                 Name:     "TestPullWorkerIntegration_GetLocator",
  73                 Locator:  "5d41402abc4b2a76b9719d911017c592",
  74                 Content:  "hello",
  75                 GetError: "",
  76         }
  77
  78         pullRequest := s.setupPullWorkerIntegrationTest(c, testData, true)
  79         defer arvadostest.StopAPI()
  80         defer arvadostest.StopKeep(2)
  81
  82         s.performPullWorkerIntegrationTest(testData, pullRequest, c)
  83 }
  84
  85 // Perform the test.
  86 // The test directly invokes the "PullItemAndProcess" rather than
  87 // putting an item on the pullq so that the errors can be verified.
  88 func (s *HandlerSuite) performPullWorkerIntegrationTest(testData PullWorkIntegrationTestData, pullRequest PullRequest, c *check.C) {
  89
  90         // Override writePulledBlock to mock PutBlock functionality
  91         defer func(orig func(*RRVolumeManager, Volume, []byte, string) error) { writePulledBlock = orig }(writePulledBlock)
  92         writePulledBlock = func(_ *RRVolumeManager, _ Volume, content []byte, _ string) error {
  93                 c.Check(string(content), check.Equals, testData.Content)
  94                 return nil
  95         }
  96
  97         // Override GetContent to mock keepclient Get functionality
  98         defer func(orig func(string, *keepclient.KeepClient) (io.ReadCloser, int64, string, error)) {
  99                 GetContent = orig
 100         }(GetContent)
 101         GetContent = func(signedLocator string, keepClient *keepclient.KeepClient) (reader io.ReadCloser, contentLength int64, url string, err error) {
 102                 if testData.GetError != "" {
 103                         return nil, 0, "", errors.New(testData.GetError)
 104                 }
 105                 rdr := ioutil.NopCloser(bytes.NewBufferString(testData.Content))
 106                 return rdr, int64(len(testData.Content)), "", nil
 107         }
 108
 109         err := s.handler.pullItemAndProcess(pullRequest)
 110
 111         if len(testData.GetError) > 0 {
 112                 if (err == nil) || (!strings.Contains(err.Error(), testData.GetError)) {
 113                         c.Errorf("Got error %v, expected %v", err, testData.GetError)
 114                 }
 115         } else {
 116                 if err != nil {
 117                         c.Errorf("Got error %v, expected nil", err)
 118                 }
 119         }
 120 }