Merge branch '6429-crunch2-api' closes #6429
[arvados.git] / services / keepstore / pull_worker_integration_test.go
index 05da98b43dc1e958fdc5318de5a5d09a82d80357..e0613a2cd12685195d841c2ef2ae31f7b90f60bb 100644 (file)
@@ -1,19 +1,19 @@
 package main
 
 import (
-       "crypto/tls"
-       "fmt"
+       "bytes"
+       "errors"
        "git.curoverse.com/arvados.git/sdk/go/arvadosclient"
        "git.curoverse.com/arvados.git/sdk/go/arvadostest"
        "git.curoverse.com/arvados.git/sdk/go/keepclient"
+       "io"
        "net/http"
        "os"
        "strings"
        "testing"
-       "encoding/json"
 )
 
-var keepClient keepclient.KeepClient
+var keepClient *keepclient.KeepClient
 
 type PullWorkIntegrationTestData struct {
        Name     string
@@ -25,108 +25,53 @@ type PullWorkIntegrationTestData struct {
 func SetupPullWorkerIntegrationTest(t *testing.T, testData PullWorkIntegrationTestData, wantData bool) PullRequest {
        os.Setenv("ARVADOS_API_HOST_INSECURE", "true")
 
+       // start api and keep servers
        arvadostest.StartAPI()
-       arvadostest.StartKeep()
+       arvadostest.StartKeep(2, false)
 
+       // make arvadosclient
        arv, err := arvadosclient.MakeArvadosClient()
        if err != nil {
                t.Error("Error creating arv")
        }
 
-       keepClient = keepclient.KeepClient{
+       // keep client
+       keepClient = &keepclient.KeepClient{
                Arvados:       &arv,
-               Want_replicas: 0,
-               Using_proxy:   true,
+               Want_replicas: 1,
                Client:        &http.Client{},
        }
 
-       random_token := GenerateRandomApiToken()
-       keepClient.Arvados.ApiToken = random_token
-       if err != nil {
-               t.Error("Error creating keepclient")
+       // discover keep services
+       var servers []string
+       if err := keepClient.DiscoverKeepServers(); err != nil {
+               t.Error("Error discovering keep services")
        }
-
-       servers := GetKeepServices(t)
-
-       pullRequest := PullRequest{
-               Locator: testData.Locator,
-               Servers: servers,
+       for _, host := range keepClient.LocalRoots() {
+               servers = append(servers, host)
        }
 
+       // Put content if the test needs it
        if wantData {
-               service_roots := make(map[string]string)
-               for _, addr := range pullRequest.Servers {
-                       service_roots[addr] = addr
-               }
-               keepClient.SetServiceRoots(service_roots)
-
                locator, _, err := keepClient.PutB([]byte(testData.Content))
                if err != nil {
                        t.Errorf("Error putting test data in setup for %s %s %v", testData.Content, locator, err)
                }
-       }
-
-       return pullRequest
-}
-
-func GetKeepServices(t *testing.T) []string {
-       client := &http.Client{Transport: &http.Transport{
-               TLSClientConfig: &tls.Config{InsecureSkipVerify: true}}}
-
-       req, err := http.NewRequest("GET", fmt.Sprintf("https://%s/arvados/v1/keep_services", os.Getenv("ARVADOS_API_HOST")), nil)
-       if err != nil {
-               t.Errorf("Error getting keep services: ", err)
-       }
-       req.Header.Set("Authorization", fmt.Sprintf("OAuth2 %s", os.Getenv("ARVADOS_API_TOKEN")))
-
-       resp, err := client.Do(req)
-       if err != nil {
-               t.Errorf("Error getting keep services: ", err)
-       }
-       if resp.StatusCode != 200 {
-               t.Errorf("Error status code getting keep services", resp.StatusCode)
-       }
-
-       defer resp.Body.Close()
-       var servers []string
-
-       decoder := json.NewDecoder(resp.Body)
-
-       var respJSON map[string]interface{}
-       err = decoder.Decode(&respJSON)
-       if err != nil {
-               t.Errorf("Error decoding response for keep services: ", err)
-       }
-
-       var service_names []string
-       var service_ports []string
-       for _, v1 := range respJSON {
-               switch v1_type := v1.(type) {
-               case []interface{}:
-                       for _, v2 := range v1_type {
-                               switch v2_type := v2.(type) {
-                               case map[string]interface{}:
-                                       for name, value := range v2_type {
-                                               if name == "service_host" {
-                                                       service_names = append(service_names, fmt.Sprintf("%s", value))
-                                               } else if name == "service_port" {
-                                                       service_ports = append(service_ports, strings.Split(fmt.Sprintf("%f", value), ".")[0])
-                                               }
-                                       }
-                               default:
-                               }
-                       }
-               default:
+               if locator == "" {
+                       t.Errorf("No locator found after putting test data")
                }
        }
 
-       for i, port := range service_ports {
-               servers = append(servers, "https://"+service_names[i]+":"+port)
+       // Create pullRequest for the test
+       pullRequest := PullRequest{
+               Locator: testData.Locator,
+               Servers: servers,
        }
-
-       return servers
+       return pullRequest
 }
 
+// Do a get on a block that is not existing in any of the keep servers.
+// Expect "block not found" error.
 func TestPullWorkerIntegration_GetNonExistingLocator(t *testing.T) {
        testData := PullWorkIntegrationTestData{
                Name:     "TestPullWorkerIntegration_GetLocator",
@@ -136,10 +81,14 @@ func TestPullWorkerIntegration_GetNonExistingLocator(t *testing.T) {
        }
 
        pullRequest := SetupPullWorkerIntegrationTest(t, testData, false)
+       defer arvadostest.StopAPI()
+       defer arvadostest.StopKeep(2)
 
        performPullWorkerIntegrationTest(testData, pullRequest, t)
 }
 
+// Do a get on a block that exists on one of the keep servers.
+// The setup method will create this block before doing the get.
 func TestPullWorkerIntegration_GetExistingLocator(t *testing.T) {
        testData := PullWorkIntegrationTestData{
                Name:     "TestPullWorkerIntegration_GetLocator",
@@ -149,23 +98,49 @@ func TestPullWorkerIntegration_GetExistingLocator(t *testing.T) {
        }
 
        pullRequest := SetupPullWorkerIntegrationTest(t, testData, true)
+       defer arvadostest.StopAPI()
+       defer arvadostest.StopKeep(2)
 
        performPullWorkerIntegrationTest(testData, pullRequest, t)
 }
 
+// Perform the test.
+// The test directly invokes the "PullItemAndProcess" rather than
+// putting an item on the pullq so that the errors can be verified.
 func performPullWorkerIntegrationTest(testData PullWorkIntegrationTestData, pullRequest PullRequest, t *testing.T) {
+
        // Override PutContent to mock PutBlock functionality
+       defer func(orig func([]byte, string) error) { PutContent = orig }(PutContent)
        PutContent = func(content []byte, locator string) (err error) {
+               if string(content) != testData.Content {
+                       t.Errorf("PutContent invoked with unexpected data. Expected: %s; Found: %s", testData.Content, content)
+               }
                return
        }
 
+       // Override GetContent to mock keepclient Get functionality
+       defer func(orig func(string, *keepclient.KeepClient) (io.ReadCloser, int64, string, error)) {
+               GetContent = orig
+       }(GetContent)
+       GetContent = func(signedLocator string, keepClient *keepclient.KeepClient) (
+               reader io.ReadCloser, contentLength int64, url string, err error) {
+               if testData.GetError != "" {
+                       return nil, 0, "", errors.New(testData.GetError)
+               }
+               rdr := &ClosingBuffer{bytes.NewBufferString(testData.Content)}
+               return rdr, int64(len(testData.Content)), "", nil
+       }
+
+       keepClient.Arvados.ApiToken = GenerateRandomAPIToken()
        err := PullItemAndProcess(pullRequest, keepClient.Arvados.ApiToken, keepClient)
 
        if len(testData.GetError) > 0 {
                if (err == nil) || (!strings.Contains(err.Error(), testData.GetError)) {
-                       t.Fail()
+                       t.Errorf("Got error %v, expected %v", err, testData.GetError)
                }
        } else {
-               t.Fail()
+               if err != nil {
+                       t.Errorf("Got error %v, expected nil", err)
+               }
        }
 }