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 keepstore
   6
   7 import (
   8         "bytes"
   9         "context"
  10         "errors"
  11         "io"
  12         "io/ioutil"
  13         "strings"
  14
  15         "git.arvados.org/arvados.git/sdk/go/arvadostest"
  16         "git.arvados.org/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.StopKeep(2)
  62
  63         s.performPullWorkerIntegrationTest(testData, pullRequest, c)
  64 }
  65
  66 // Do a get on a block that exists on one of the keep servers.
  67 // The setup method will create this block before doing the get.
  68 func (s *HandlerSuite) TestPullWorkerIntegration_GetExistingLocator(c *check.C) {
  69         c.Assert(s.handler.setup(context.Background(), s.cluster, "", prometheus.NewRegistry(), testServiceURL), check.IsNil)
  70         testData := PullWorkIntegrationTestData{
  71                 Name:     "TestPullWorkerIntegration_GetLocator",
  72                 Locator:  "5d41402abc4b2a76b9719d911017c592",
  73                 Content:  "hello",
  74                 GetError: "",
  75         }
  76
  77         pullRequest := s.setupPullWorkerIntegrationTest(c, testData, true)
  78         defer arvadostest.StopKeep(2)
  79
  80         s.performPullWorkerIntegrationTest(testData, pullRequest, c)
  81 }
  82
  83 // Perform the test.
  84 // The test directly invokes the "PullItemAndProcess" rather than
  85 // putting an item on the pullq so that the errors can be verified.
  86 func (s *HandlerSuite) performPullWorkerIntegrationTest(testData PullWorkIntegrationTestData, pullRequest PullRequest, c *check.C) {
  87
  88         // Override writePulledBlock to mock PutBlock functionality
  89         defer func(orig func(*RRVolumeManager, Volume, []byte, string) error) { writePulledBlock = orig }(writePulledBlock)
  90         writePulledBlock = func(_ *RRVolumeManager, _ Volume, content []byte, _ string) error {
  91                 c.Check(string(content), check.Equals, testData.Content)
  92                 return nil
  93         }
  94
  95         // Override GetContent to mock keepclient Get functionality
  96         defer func(orig func(string, *keepclient.KeepClient) (io.ReadCloser, int64, string, error)) {
  97                 GetContent = orig
  98         }(GetContent)
  99         GetContent = func(signedLocator string, keepClient *keepclient.KeepClient) (reader io.ReadCloser, contentLength int64, url string, err error) {
 100                 if testData.GetError != "" {
 101                         return nil, 0, "", errors.New(testData.GetError)
 102                 }
 103                 rdr := ioutil.NopCloser(bytes.NewBufferString(testData.Content))
 104                 return rdr, int64(len(testData.Content)), "", nil
 105         }
 106
 107         err := s.handler.pullItemAndProcess(pullRequest)
 108
 109         if len(testData.GetError) > 0 {
 110                 if (err == nil) || (!strings.Contains(err.Error(), testData.GetError)) {
 111                         c.Errorf("Got error %v, expected %v", err, testData.GetError)
 112                 }
 113         } else {
 114                 if err != nil {
 115                         c.Errorf("Got error %v, expected nil", err)
 116                 }
 117         }
 118 }