services/datamanager/keep/keep_test.go

   1 package keep
   2
   3 import (
   4         "encoding/json"
   5         "fmt"
   6         "net/http"
   7         "net/http/httptest"
   8         "net/url"
   9         "strconv"
  10         "strings"
  11         "testing"
  12
  13         "git.curoverse.com/arvados.git/sdk/go/arvadosclient"
  14         "git.curoverse.com/arvados.git/sdk/go/blockdigest"
  15         "git.curoverse.com/arvados.git/sdk/go/keepclient"
  16
  17         . "gopkg.in/check.v1"
  18 )
  19
  20 // Gocheck boilerplate
  21 func Test(t *testing.T) {
  22         TestingT(t)
  23 }
  24
  25 type KeepSuite struct{}
  26
  27 var _ = Suite(&KeepSuite{})
  28
  29 type TestHandler struct {
  30         request TrashList
  31 }
  32
  33 func (ts *TestHandler) ServeHTTP(writer http.ResponseWriter, req *http.Request) {
  34         r := json.NewDecoder(req.Body)
  35         r.Decode(&ts.request)
  36 }
  37
  38 func (s *KeepSuite) TestSendTrashLists(c *C) {
  39         th := TestHandler{}
  40         server := httptest.NewServer(&th)
  41         defer server.Close()
  42
  43         tl := map[string]TrashList{
  44                 server.URL: TrashList{TrashRequest{"000000000000000000000000deadbeef", 99}}}
  45
  46         arv := arvadosclient.ArvadosClient{ApiToken: "abc123"}
  47         kc := keepclient.KeepClient{Arvados: &arv, Client: &http.Client{}}
  48         kc.SetServiceRoots(map[string]string{"xxxx": server.URL},
  49                 map[string]string{"xxxx": server.URL},
  50                 map[string]string{})
  51
  52         err := SendTrashLists(&kc, tl)
  53
  54         c.Check(err, IsNil)
  55
  56         c.Check(th.request,
  57                 DeepEquals,
  58                 tl[server.URL])
  59
  60 }
  61
  62 type TestHandlerError struct {
  63 }
  64
  65 func (tse *TestHandlerError) ServeHTTP(writer http.ResponseWriter, req *http.Request) {
  66         http.Error(writer, "I'm a teapot", 418)
  67 }
  68
  69 func sendTrashListError(c *C, server *httptest.Server) {
  70         tl := map[string]TrashList{
  71                 server.URL: TrashList{TrashRequest{"000000000000000000000000deadbeef", 99}}}
  72
  73         arv := arvadosclient.ArvadosClient{ApiToken: "abc123"}
  74         kc := keepclient.KeepClient{Arvados: &arv, Client: &http.Client{}}
  75         kc.SetServiceRoots(map[string]string{"xxxx": server.URL},
  76                 map[string]string{"xxxx": server.URL},
  77                 map[string]string{})
  78
  79         err := SendTrashLists(&kc, tl)
  80
  81         c.Check(err, NotNil)
  82         c.Check(err[0], NotNil)
  83 }
  84
  85 func (s *KeepSuite) TestSendTrashListErrorResponse(c *C) {
  86         server := httptest.NewServer(&TestHandlerError{})
  87         sendTrashListError(c, server)
  88         defer server.Close()
  89 }
  90
  91 func (s *KeepSuite) TestSendTrashListUnreachable(c *C) {
  92         sendTrashListError(c, httptest.NewUnstartedServer(&TestHandler{}))
  93 }
  94
  95 type StatusAndBody struct {
  96         respStatus   int
  97         responseBody string
  98 }
  99
 100 type APIStub struct {
 101         data map[string]StatusAndBody
 102 }
 103
 104 func (stub *APIStub) ServeHTTP(resp http.ResponseWriter, req *http.Request) {
 105         if req.URL.Path == "/redirect-loop" {
 106                 http.Redirect(resp, req, "/redirect-loop", http.StatusFound)
 107                 return
 108         }
 109
 110         pathResponse := stub.data[req.URL.Path]
 111         if pathResponse.responseBody != "" {
 112                 if pathResponse.respStatus == -1 {
 113                         http.Redirect(resp, req, "/redirect-loop", http.StatusFound)
 114                 } else {
 115                         resp.WriteHeader(pathResponse.respStatus)
 116                         resp.Write([]byte(pathResponse.responseBody))
 117                 }
 118         } else {
 119                 resp.WriteHeader(500)
 120                 resp.Write([]byte(``))
 121         }
 122 }
 123
 124 type KeepServerStub struct {
 125         data map[string]StatusAndBody
 126 }
 127
 128 func (stub *KeepServerStub) ServeHTTP(resp http.ResponseWriter, req *http.Request) {
 129         if req.URL.Path == "/redirect-loop" {
 130                 http.Redirect(resp, req, "/redirect-loop", http.StatusFound)
 131                 return
 132         }
 133
 134         pathResponse := stub.data[req.URL.Path]
 135         if pathResponse.responseBody != "" {
 136                 if pathResponse.respStatus == -1 {
 137                         http.Redirect(resp, req, "/redirect-loop", http.StatusFound)
 138                 } else {
 139                         resp.WriteHeader(pathResponse.respStatus)
 140                         resp.Write([]byte(pathResponse.responseBody))
 141                 }
 142         } else {
 143                 resp.WriteHeader(500)
 144                 resp.Write([]byte(``))
 145         }
 146 }
 147
 148 type APITestData struct {
 149         numServers int
 150         serverType string
 151         statusCode int
 152 }
 153
 154 func (s *KeepSuite) TestGetKeepServers_UnsupportedServiceType(c *C) {
 155         testGetKeepServersFromAPI(c, APITestData{1, "notadisk", 200})
 156 }
 157
 158 func (s *KeepSuite) TestGetKeepServers_ReceivedTooFewServers(c *C) {
 159         testGetKeepServersFromAPI(c, APITestData{2, "disk", 200})
 160 }
 161
 162 func (s *KeepSuite) TestGetKeepServers_ServerError(c *C) {
 163         testGetKeepServersFromAPI(c, APITestData{-1, "disk", -1})
 164 }
 165
 166 func testGetKeepServersFromAPI(c *C, testData APITestData) {
 167         keepServers := ServiceList{
 168                 ItemsAvailable: testData.numServers,
 169                 KeepServers: []ServerAddress{{
 170                         SSL:         false,
 171                         Host:        "example.com",
 172                         Port:        12345,
 173                         UUID:        "abcdefg",
 174                         ServiceType: testData.serverType,
 175                 }},
 176         }
 177
 178         ksJSON, _ := json.Marshal(keepServers)
 179         apiData := make(map[string]StatusAndBody)
 180         apiData["/arvados/v1/keep_services"] = StatusAndBody{testData.statusCode, string(ksJSON)}
 181         apiStub := APIStub{apiData}
 182
 183         api := httptest.NewServer(&apiStub)
 184         defer api.Close()
 185
 186         arv := arvadosclient.ArvadosClient{
 187                 Scheme:    "http",
 188                 ApiServer: api.URL[7:],
 189                 ApiToken:  "abc123",
 190                 Client:    &http.Client{Transport: &http.Transport{}},
 191         }
 192
 193         kc := keepclient.KeepClient{Arvados: &arv, Client: &http.Client{}}
 194         kc.SetServiceRoots(map[string]string{"xxxx": "http://example.com:23456"},
 195                 map[string]string{"xxxx": "http://example.com:23456"},
 196                 map[string]string{})
 197
 198         params := GetKeepServersParams{
 199                 Client: arv,
 200                 Logger: nil,
 201                 Limit:  10,
 202         }
 203
 204         _, err := GetKeepServersAndSummarize(params)
 205         if testData.numServers > 1 {
 206                 c.Assert(err, ErrorMatches, ".*Did not receive all available keep servers.*")
 207         } else if testData.serverType != "disk" {
 208                 c.Assert(err, ErrorMatches, ".*Unsupported service type.*")
 209         }
 210 }
 211
 212 type KeepServerTestData struct {
 213         // handle /status.json
 214         statusStatusCode int
 215
 216         // handle /index
 217         indexStatusCode   int
 218         indexResponseBody string
 219
 220         // expected error, if any
 221         expectedError string
 222 }
 223
 224 func (s *KeepSuite) TestGetKeepServers_ErrorGettingKeepServerStatus(c *C) {
 225         testGetKeepServersAndSummarize(c, KeepServerTestData{500, 200, "ok",
 226                 "500 Internal Server Error"})
 227 }
 228
 229 func (s *KeepSuite) TestGetKeepServers_GettingIndex(c *C) {
 230         testGetKeepServersAndSummarize(c, KeepServerTestData{200, -1, "notok",
 231                 "redirect-loop"})
 232 }
 233
 234 func (s *KeepSuite) TestGetKeepServers_ErrorReadServerResponse(c *C) {
 235         testGetKeepServersAndSummarize(c, KeepServerTestData{200, 500, "notok",
 236                 "500 Internal Server Error"})
 237 }
 238
 239 func (s *KeepSuite) TestGetKeepServers_ReadServerResponseTuncatedAtLineOne(c *C) {
 240         testGetKeepServersAndSummarize(c, KeepServerTestData{200, 200,
 241                 "notterminatedwithnewline", "truncated at line 1"})
 242 }
 243
 244 func (s *KeepSuite) TestGetKeepServers_InvalidBlockLocatorPattern(c *C) {
 245         testGetKeepServersAndSummarize(c, KeepServerTestData{200, 200, "testing\n",
 246                 "Error parsing BlockInfo from index line"})
 247 }
 248
 249 func (s *KeepSuite) TestGetKeepServers_ReadServerResponseEmpty(c *C) {
 250         testGetKeepServersAndSummarize(c, KeepServerTestData{200, 200, "\n", ""})
 251 }
 252
 253 func (s *KeepSuite) TestGetKeepServers_ReadServerResponseWithTwoBlocks(c *C) {
 254         testGetKeepServersAndSummarize(c, KeepServerTestData{200, 200,
 255                 "51752ba076e461ec9ec1d27400a08548+20 1447526361\na048cc05c02ba1ee43ad071274b9e547+52 1447526362\n\n", ""})
 256 }
 257
 258 func testGetKeepServersAndSummarize(c *C, testData KeepServerTestData) {
 259         ksData := make(map[string]StatusAndBody)
 260         ksData["/status.json"] = StatusAndBody{testData.statusStatusCode, string(`{}`)}
 261         ksData["/index"] = StatusAndBody{testData.indexStatusCode, testData.indexResponseBody}
 262         ksStub := KeepServerStub{ksData}
 263         ks := httptest.NewServer(&ksStub)
 264         defer ks.Close()
 265
 266         ksURL, err := url.Parse(ks.URL)
 267         c.Check(err, IsNil)
 268         ksURLParts := strings.Split(ksURL.Host, ":")
 269         ksPort, err := strconv.Atoi(ksURLParts[1])
 270         c.Check(err, IsNil)
 271
 272         servers_list := ServiceList{
 273                 ItemsAvailable: 1,
 274                 KeepServers: []ServerAddress{{
 275                         SSL:         false,
 276                         Host:        strings.Split(ksURL.Host, ":")[0],
 277                         Port:        ksPort,
 278                         UUID:        "abcdefg",
 279                         ServiceType: "disk",
 280                 }},
 281         }
 282         ksJSON, _ := json.Marshal(servers_list)
 283         apiData := make(map[string]StatusAndBody)
 284         apiData["/arvados/v1/keep_services"] = StatusAndBody{200, string(ksJSON)}
 285         apiStub := APIStub{apiData}
 286
 287         api := httptest.NewServer(&apiStub)
 288         defer api.Close()
 289
 290         arv := arvadosclient.ArvadosClient{
 291                 Scheme:    "http",
 292                 ApiServer: api.URL[7:],
 293                 ApiToken:  "abc123",
 294                 Client:    &http.Client{Transport: &http.Transport{}},
 295         }
 296
 297         kc := keepclient.KeepClient{Arvados: &arv, Client: &http.Client{}}
 298         kc.SetServiceRoots(map[string]string{"xxxx": ks.URL},
 299                 map[string]string{"xxxx": ks.URL},
 300                 map[string]string{})
 301
 302         params := GetKeepServersParams{
 303                 Client: arv,
 304                 Logger: nil,
 305                 Limit:  10,
 306         }
 307
 308         // GetKeepServersAndSummarize
 309         results, err := GetKeepServersAndSummarize(params)
 310
 311         if testData.expectedError == "" {
 312                 c.Assert(err, IsNil)
 313                 c.Assert(results, NotNil)
 314
 315                 blockToServers := results.BlockToServers
 316
 317                 blockLocators := strings.Split(testData.indexResponseBody, "\n")
 318                 for _, loc := range blockLocators {
 319                         locator := strings.Split(loc, " ")[0]
 320                         if locator != "" {
 321                                 blockLocator, err := blockdigest.ParseBlockLocator(locator)
 322                                 c.Assert(err, IsNil)
 323
 324                                 blockDigestWithSize := blockdigest.DigestWithSize{blockLocator.Digest, uint32(blockLocator.Size)}
 325                                 blockServerInfo := blockToServers[blockDigestWithSize]
 326                                 c.Assert(blockServerInfo[0].Mtime, NotNil)
 327                         }
 328                 }
 329         } else {
 330                 if testData.expectedError == "Error parsing BlockInfo from index line" {
 331                         // In this case ErrorMatches does not work because the error message contains regexp match characters
 332                         strings.Contains(err.Error(), testData.expectedError)
 333                 } else {
 334                         c.Assert(err, ErrorMatches, fmt.Sprintf(".*%s.*", testData.expectedError))
 335                 }
 336         }
 337 }