services/keepstore/pull_worker_integration_test.go

   1 package main
   2
   3 import (
   4         "crypto/tls"
   5         "fmt"
   6         "git.curoverse.com/arvados.git/sdk/go/arvadosclient"
   7         "git.curoverse.com/arvados.git/sdk/go/arvadostest"
   8         "git.curoverse.com/arvados.git/sdk/go/keepclient"
   9         "net/http"
  10         "os"
  11         "strings"
  12         "testing"
  13         "encoding/json"
  14 )
  15
  16 var keepClient keepclient.KeepClient
  17
  18 type PullWorkIntegrationTestData struct {
  19         Name     string
  20         Locator  string
  21         Content  string
  22         GetError string
  23 }
  24
  25 func SetupPullWorkerIntegrationTest(t *testing.T, testData PullWorkIntegrationTestData, wantData bool) PullRequest {
  26         os.Setenv("ARVADOS_API_HOST_INSECURE", "true")
  27
  28         arvadostest.StartAPI()
  29         arvadostest.StartKeep()
  30
  31         arv, err := arvadosclient.MakeArvadosClient()
  32         if err != nil {
  33                 t.Error("Error creating arv")
  34         }
  35
  36         keepClient = keepclient.KeepClient{
  37                 Arvados:       &arv,
  38                 Want_replicas: 0,
  39                 Using_proxy:   true,
  40                 Client:        &http.Client{},
  41         }
  42
  43         random_token := GenerateRandomApiToken()
  44         keepClient.Arvados.ApiToken = random_token
  45         if err != nil {
  46                 t.Error("Error creating keepclient")
  47         }
  48
  49         servers := GetKeepServices(t)
  50
  51         pullRequest := PullRequest{
  52                 Locator: testData.Locator,
  53                 Servers: servers,
  54         }
  55
  56         if wantData {
  57                 service_roots := make(map[string]string)
  58                 for _, addr := range pullRequest.Servers {
  59                         service_roots[addr] = addr
  60                 }
  61                 keepClient.SetServiceRoots(service_roots)
  62
  63                 locator, _, err := keepClient.PutB([]byte(testData.Content))
  64                 if err != nil {
  65                         t.Errorf("Error putting test data in setup for %s %s %v", testData.Content, locator, err)
  66                 }
  67         }
  68
  69         return pullRequest
  70 }
  71
  72 func GetKeepServices(t *testing.T) []string {
  73         client := &http.Client{Transport: &http.Transport{
  74                 TLSClientConfig: &tls.Config{InsecureSkipVerify: true}}}
  75
  76         req, err := http.NewRequest("GET", fmt.Sprintf("https://%s/arvados/v1/keep_services", os.Getenv("ARVADOS_API_HOST")), nil)
  77         if err != nil {
  78                 t.Errorf("Error getting keep services: ", err)
  79         }
  80         req.Header.Set("Authorization", fmt.Sprintf("OAuth2 %s", os.Getenv("ARVADOS_API_TOKEN")))
  81
  82         resp, err := client.Do(req)
  83         if err != nil {
  84                 t.Errorf("Error getting keep services: ", err)
  85         }
  86         if resp.StatusCode != 200 {
  87                 t.Errorf("Error status code getting keep services", resp.StatusCode)
  88         }
  89
  90         defer resp.Body.Close()
  91         var servers []string
  92
  93         decoder := json.NewDecoder(resp.Body)
  94
  95         var respJSON map[string]interface{}
  96         err = decoder.Decode(&respJSON)
  97         if err != nil {
  98                 t.Errorf("Error decoding response for keep services: ", err)
  99         }
 100
 101         var service_names []string
 102         var service_ports []string
 103         for _, v1 := range respJSON {
 104                 switch v1_type := v1.(type) {
 105                 case []interface{}:
 106                         for _, v2 := range v1_type {
 107                                 switch v2_type := v2.(type) {
 108                                 case map[string]interface{}:
 109                                         for name, value := range v2_type {
 110                                                 if name == "service_host" {
 111                                                         service_names = append(service_names, fmt.Sprintf("%s", value))
 112                                                 } else if name == "service_port" {
 113                                                         service_ports = append(service_ports, strings.Split(fmt.Sprintf("%f", value), ".")[0])
 114                                                 }
 115                                         }
 116                                 default:
 117                                 }
 118                         }
 119                 default:
 120                 }
 121         }
 122
 123         for i, port := range service_ports {
 124                 servers = append(servers, "https://"+service_names[i]+":"+port)
 125         }
 126
 127         return servers
 128 }
 129
 130 func TestPullWorkerIntegration_GetNonExistingLocator(t *testing.T) {
 131         testData := PullWorkIntegrationTestData{
 132                 Name:     "TestPullWorkerIntegration_GetLocator",
 133                 Locator:  "5d41402abc4b2a76b9719d911017c592",
 134                 Content:  "hello",
 135                 GetError: "Block not found",
 136         }
 137
 138         pullRequest := SetupPullWorkerIntegrationTest(t, testData, false)
 139
 140         performPullWorkerIntegrationTest(testData, pullRequest, t)
 141 }
 142
 143 func TestPullWorkerIntegration_GetExistingLocator(t *testing.T) {
 144         testData := PullWorkIntegrationTestData{
 145                 Name:     "TestPullWorkerIntegration_GetLocator",
 146                 Locator:  "5d41402abc4b2a76b9719d911017c592",
 147                 Content:  "hello",
 148                 GetError: "",
 149         }
 150
 151         pullRequest := SetupPullWorkerIntegrationTest(t, testData, true)
 152
 153         performPullWorkerIntegrationTest(testData, pullRequest, t)
 154 }
 155
 156 func performPullWorkerIntegrationTest(testData PullWorkIntegrationTestData, pullRequest PullRequest, t *testing.T) {
 157         // Override PutContent to mock PutBlock functionality
 158         PutContent = func(content []byte, locator string) (err error) {
 159                 return
 160         }
 161
 162         err := PullItemAndProcess(pullRequest, keepClient.Arvados.ApiToken, keepClient)
 163
 164         if len(testData.GetError) > 0 {
 165                 if (err == nil) || (!strings.Contains(err.Error(), testData.GetError)) {
 166                         t.Fail()
 167                 }
 168         } else {
 169                 t.Fail()
 170         }
 171 }