sdk/go/keepclient/keepclient_test.go

   1 // Copyright (C) The Arvados Authors. All rights reserved.
   2 //
   3 // SPDX-License-Identifier: Apache-2.0
   4
   5 package keepclient
   6
   7 import (
   8         "bytes"
   9         "context"
  10         "crypto/md5"
  11         "fmt"
  12         "io"
  13         "io/ioutil"
  14         "log"
  15         "net"
  16         "net/http"
  17         "os"
  18         "strings"
  19         "sync"
  20         "testing"
  21         "time"
  22
  23         "git.arvados.org/arvados.git/sdk/go/arvados"
  24         "git.arvados.org/arvados.git/sdk/go/arvadosclient"
  25         "git.arvados.org/arvados.git/sdk/go/arvadostest"
  26         . "gopkg.in/check.v1"
  27 )
  28
  29 // Gocheck boilerplate
  30 func Test(t *testing.T) {
  31         TestingT(t)
  32 }
  33
  34 // Gocheck boilerplate
  35 var _ = Suite(&ServerRequiredSuite{})
  36 var _ = Suite(&StandaloneSuite{})
  37
  38 // Tests that require the Keep server running
  39 type ServerRequiredSuite struct{}
  40
  41 // Standalone tests
  42 type StandaloneSuite struct{}
  43
  44 func (s *StandaloneSuite) SetUpTest(c *C) {
  45         RefreshServiceDiscovery()
  46 }
  47
  48 func pythonDir() string {
  49         cwd, _ := os.Getwd()
  50         return fmt.Sprintf("%s/../../python/tests", cwd)
  51 }
  52
  53 func (s *ServerRequiredSuite) SetUpSuite(c *C) {
  54         arvadostest.StartKeep(2, false)
  55 }
  56
  57 func (s *ServerRequiredSuite) TearDownSuite(c *C) {
  58         arvadostest.StopKeep(2)
  59 }
  60
  61 func (s *ServerRequiredSuite) SetUpTest(c *C) {
  62         RefreshServiceDiscovery()
  63 }
  64
  65 func (s *ServerRequiredSuite) TestMakeKeepClient(c *C) {
  66         arv, err := arvadosclient.MakeArvadosClient()
  67         c.Assert(err, Equals, nil)
  68
  69         kc, err := MakeKeepClient(arv)
  70
  71         c.Assert(err, Equals, nil)
  72         c.Check(len(kc.LocalRoots()), Equals, 2)
  73         for _, root := range kc.LocalRoots() {
  74                 c.Check(root, Matches, "http://localhost:\\d+")
  75         }
  76 }
  77
  78 func (s *ServerRequiredSuite) TestDefaultStorageClasses(c *C) {
  79         arv, err := arvadosclient.MakeArvadosClient()
  80         c.Assert(err, IsNil)
  81
  82         cc, err := arv.ClusterConfig("StorageClasses")
  83         c.Assert(err, IsNil)
  84         c.Assert(cc, NotNil)
  85         c.Assert(cc.(map[string]interface{})["default"], NotNil)
  86
  87         kc := New(arv)
  88         c.Assert(kc.DefaultStorageClasses, DeepEquals, []string{"default"})
  89 }
  90
  91 func (s *ServerRequiredSuite) TestDefaultReplications(c *C) {
  92         arv, err := arvadosclient.MakeArvadosClient()
  93         c.Assert(err, IsNil)
  94
  95         kc, err := MakeKeepClient(arv)
  96         c.Check(err, IsNil)
  97         c.Assert(kc.Want_replicas, Equals, 2)
  98
  99         arv.DiscoveryDoc["defaultCollectionReplication"] = 3.0
 100         kc, err = MakeKeepClient(arv)
 101         c.Check(err, IsNil)
 102         c.Assert(kc.Want_replicas, Equals, 3)
 103
 104         arv.DiscoveryDoc["defaultCollectionReplication"] = 1.0
 105         kc, err = MakeKeepClient(arv)
 106         c.Check(err, IsNil)
 107         c.Assert(kc.Want_replicas, Equals, 1)
 108 }
 109
 110 type StubPutHandler struct {
 111         c                    *C
 112         expectPath           string
 113         expectAPIToken       string
 114         expectBody           string
 115         expectStorageClass   string
 116         returnStorageClasses string
 117         handled              chan string
 118         requests             []*http.Request
 119         mtx                  sync.Mutex
 120 }
 121
 122 func (sph *StubPutHandler) ServeHTTP(resp http.ResponseWriter, req *http.Request) {
 123         sph.mtx.Lock()
 124         sph.requests = append(sph.requests, req)
 125         sph.mtx.Unlock()
 126         sph.c.Check(req.URL.Path, Equals, "/"+sph.expectPath)
 127         sph.c.Check(req.Header.Get("Authorization"), Equals, fmt.Sprintf("OAuth2 %s", sph.expectAPIToken))
 128         if sph.expectStorageClass != "*" {
 129                 sph.c.Check(req.Header.Get("X-Keep-Storage-Classes"), Equals, sph.expectStorageClass)
 130         }
 131         body, err := ioutil.ReadAll(req.Body)
 132         sph.c.Check(err, Equals, nil)
 133         sph.c.Check(body, DeepEquals, []byte(sph.expectBody))
 134         resp.Header().Set("X-Keep-Replicas-Stored", "1")
 135         if sph.returnStorageClasses != "" {
 136                 resp.Header().Set("X-Keep-Storage-Classes-Confirmed", sph.returnStorageClasses)
 137         }
 138         resp.WriteHeader(200)
 139         sph.handled <- fmt.Sprintf("http://%s", req.Host)
 140 }
 141
 142 func RunFakeKeepServer(st http.Handler) (ks KeepServer) {
 143         var err error
 144         // If we don't explicitly bind it to localhost, ks.listener.Addr() will
 145         // bind to 0.0.0.0 or [::] which is not a valid address for Dial()
 146         ks.listener, err = net.ListenTCP("tcp", &net.TCPAddr{IP: []byte{127, 0, 0, 1}, Port: 0})
 147         if err != nil {
 148                 panic("Could not listen on any port")
 149         }
 150         ks.url = fmt.Sprintf("http://%s", ks.listener.Addr().String())
 151         go http.Serve(ks.listener, st)
 152         return
 153 }
 154
 155 func UploadToStubHelper(c *C, st http.Handler, f func(*KeepClient, string,
 156         io.ReadCloser, io.WriteCloser, chan uploadStatus)) {
 157
 158         ks := RunFakeKeepServer(st)
 159         defer ks.listener.Close()
 160
 161         arv, _ := arvadosclient.MakeArvadosClient()
 162         arv.ApiToken = "abc123"
 163
 164         kc, _ := MakeKeepClient(arv)
 165
 166         reader, writer := io.Pipe()
 167         uploadStatusChan := make(chan uploadStatus)
 168
 169         f(kc, ks.url, reader, writer, uploadStatusChan)
 170 }
 171
 172 func (s *StandaloneSuite) TestUploadToStubKeepServer(c *C) {
 173         log.Printf("TestUploadToStubKeepServer")
 174
 175         st := &StubPutHandler{
 176                 c:                    c,
 177                 expectPath:           "acbd18db4cc2f85cedef654fccc4a4d8",
 178                 expectAPIToken:       "abc123",
 179                 expectBody:           "foo",
 180                 expectStorageClass:   "",
 181                 returnStorageClasses: "default=1",
 182                 handled:              make(chan string),
 183         }
 184
 185         UploadToStubHelper(c, st,
 186                 func(kc *KeepClient, url string, reader io.ReadCloser, writer io.WriteCloser, uploadStatusChan chan uploadStatus) {
 187                         go kc.uploadToKeepServer(url, st.expectPath, nil, reader, uploadStatusChan, len("foo"), kc.getRequestID())
 188
 189                         writer.Write([]byte("foo"))
 190                         writer.Close()
 191
 192                         <-st.handled
 193                         status := <-uploadStatusChan
 194                         c.Check(status, DeepEquals, uploadStatus{nil, fmt.Sprintf("%s/%s", url, st.expectPath), 200, 1, map[string]int{"default": 1}, ""})
 195                 })
 196 }
 197
 198 func (s *StandaloneSuite) TestUploadToStubKeepServerBufferReader(c *C) {
 199         st := &StubPutHandler{
 200                 c:                    c,
 201                 expectPath:           "acbd18db4cc2f85cedef654fccc4a4d8",
 202                 expectAPIToken:       "abc123",
 203                 expectBody:           "foo",
 204                 expectStorageClass:   "",
 205                 returnStorageClasses: "default=1",
 206                 handled:              make(chan string),
 207         }
 208
 209         UploadToStubHelper(c, st,
 210                 func(kc *KeepClient, url string, _ io.ReadCloser, _ io.WriteCloser, uploadStatusChan chan uploadStatus) {
 211                         go kc.uploadToKeepServer(url, st.expectPath, nil, bytes.NewBuffer([]byte("foo")), uploadStatusChan, 3, kc.getRequestID())
 212
 213                         <-st.handled
 214
 215                         status := <-uploadStatusChan
 216                         c.Check(status, DeepEquals, uploadStatus{nil, fmt.Sprintf("%s/%s", url, st.expectPath), 200, 1, map[string]int{"default": 1}, ""})
 217                 })
 218 }
 219
 220 func (s *StandaloneSuite) TestUploadWithStorageClasses(c *C) {
 221         for _, trial := range []struct {
 222                 respHeader string
 223                 expectMap  map[string]int
 224         }{
 225                 {"", nil},
 226                 {"foo=1", map[string]int{"foo": 1}},
 227                 {" foo=1 , bar=2 ", map[string]int{"foo": 1, "bar": 2}},
 228                 {" =foo=1 ", nil},
 229                 {"foo", nil},
 230         } {
 231                 st := &StubPutHandler{
 232                         c:                    c,
 233                         expectPath:           "acbd18db4cc2f85cedef654fccc4a4d8",
 234                         expectAPIToken:       "abc123",
 235                         expectBody:           "foo",
 236                         expectStorageClass:   "",
 237                         returnStorageClasses: trial.respHeader,
 238                         handled:              make(chan string),
 239                 }
 240
 241                 UploadToStubHelper(c, st,
 242                         func(kc *KeepClient, url string, reader io.ReadCloser, writer io.WriteCloser, uploadStatusChan chan uploadStatus) {
 243                                 go kc.uploadToKeepServer(url, st.expectPath, nil, reader, uploadStatusChan, len("foo"), kc.getRequestID())
 244
 245                                 writer.Write([]byte("foo"))
 246                                 writer.Close()
 247
 248                                 <-st.handled
 249                                 status := <-uploadStatusChan
 250                                 c.Check(status, DeepEquals, uploadStatus{nil, fmt.Sprintf("%s/%s", url, st.expectPath), 200, 1, trial.expectMap, ""})
 251                         })
 252         }
 253 }
 254
 255 func (s *StandaloneSuite) TestPutWithStorageClasses(c *C) {
 256         nServers := 5
 257         for _, trial := range []struct {
 258                 replicas       int
 259                 defaultClasses []string
 260                 clientClasses  []string // clientClasses takes precedence over defaultClasses
 261                 putClasses     []string // putClasses takes precedence over clientClasses
 262                 minRequests    int
 263                 maxRequests    int
 264                 success        bool
 265         }{
 266                 {1, []string{"class1"}, nil, nil, 1, 1, true},
 267                 {2, []string{"class1"}, nil, nil, 1, 2, true},
 268                 {3, []string{"class1"}, nil, nil, 2, 3, true},
 269                 {1, []string{"class1", "class2"}, nil, nil, 1, 1, true},
 270
 271                 // defaultClasses doesn't matter when any of the others is specified.
 272                 {1, []string{"class1"}, []string{"class1"}, nil, 1, 1, true},
 273                 {2, []string{"class1"}, []string{"class1"}, nil, 1, 2, true},
 274                 {3, []string{"class1"}, []string{"class1"}, nil, 2, 3, true},
 275                 {1, []string{"class1"}, []string{"class1", "class2"}, nil, 1, 1, true},
 276                 {3, []string{"class1"}, nil, []string{"class1"}, 2, 3, true},
 277                 {1, []string{"class1"}, nil, []string{"class1", "class2"}, 1, 1, true},
 278                 {1, []string{"class1"}, []string{"class404"}, []string{"class1", "class2"}, 1, 1, true},
 279                 {1, []string{"class1"}, []string{"class1"}, []string{"class404", "class2"}, nServers, nServers, false},
 280                 {nServers*2 + 1, []string{}, []string{"class1"}, nil, nServers, nServers, false},
 281                 {1, []string{"class1"}, []string{"class404"}, nil, nServers, nServers, false},
 282                 {1, []string{"class1"}, []string{"class1", "class404"}, nil, nServers, nServers, false},
 283                 {1, []string{"class1"}, nil, []string{"class1", "class404"}, nServers, nServers, false},
 284
 285                 // Talking to an older cluster (with no default storage classes advertising)
 286                 {1, nil, []string{"class1"}, nil, 1, 1, true},
 287                 {2, nil, []string{"class1"}, nil, 1, 2, true},
 288                 {3, nil, []string{"class1"}, nil, 2, 3, true},
 289                 {1, nil, []string{"class1", "class2"}, nil, 1, 1, true},
 290                 {3, nil, nil, []string{"class1"}, 2, 3, true},
 291                 {1, nil, nil, []string{"class1", "class2"}, 1, 1, true},
 292                 {1, nil, []string{"class404"}, []string{"class1", "class2"}, 1, 1, true},
 293                 {1, nil, []string{"class1"}, []string{"class404", "class2"}, nServers, nServers, false},
 294                 {nServers*2 + 1, []string{}, []string{"class1"}, nil, nServers, nServers, false},
 295                 {1, nil, []string{"class404"}, nil, nServers, nServers, false},
 296                 {1, nil, []string{"class1", "class404"}, nil, nServers, nServers, false},
 297                 {1, nil, nil, []string{"class1", "class404"}, nServers, nServers, false},
 298         } {
 299                 c.Logf("%+v", trial)
 300                 st := &StubPutHandler{
 301                         c:                    c,
 302                         expectPath:           "acbd18db4cc2f85cedef654fccc4a4d8",
 303                         expectAPIToken:       "abc123",
 304                         expectBody:           "foo",
 305                         expectStorageClass:   "*",
 306                         returnStorageClasses: "class1=2, class2=2",
 307                         handled:              make(chan string, 100),
 308                 }
 309                 ks := RunSomeFakeKeepServers(st, nServers)
 310                 arv, _ := arvadosclient.MakeArvadosClient()
 311                 kc, _ := MakeKeepClient(arv)
 312                 kc.Want_replicas = trial.replicas
 313                 kc.StorageClasses = trial.clientClasses
 314                 kc.DefaultStorageClasses = trial.defaultClasses
 315                 arv.ApiToken = "abc123"
 316                 localRoots := make(map[string]string)
 317                 writableLocalRoots := make(map[string]string)
 318                 for i, k := range ks {
 319                         localRoots[fmt.Sprintf("zzzzz-bi6l4-fakefakefake%03d", i)] = k.url
 320                         writableLocalRoots[fmt.Sprintf("zzzzz-bi6l4-fakefakefake%03d", i)] = k.url
 321                         defer k.listener.Close()
 322                 }
 323                 kc.SetServiceRoots(localRoots, writableLocalRoots, nil)
 324
 325                 _, err := kc.BlockWrite(context.Background(), arvados.BlockWriteOptions{
 326                         Data:           []byte("foo"),
 327                         StorageClasses: trial.putClasses,
 328                 })
 329                 if trial.success {
 330                         c.Check(err, IsNil)
 331                 } else {
 332                         c.Check(err, NotNil)
 333                 }
 334                 c.Check(len(st.handled) >= trial.minRequests, Equals, true, Commentf("len(st.handled)==%d, trial.minRequests==%d", len(st.handled), trial.minRequests))
 335                 c.Check(len(st.handled) <= trial.maxRequests, Equals, true, Commentf("len(st.handled)==%d, trial.maxRequests==%d", len(st.handled), trial.maxRequests))
 336                 if !trial.success && trial.replicas == 1 && c.Check(len(st.requests) >= 2, Equals, true) {
 337                         // Max concurrency should be 1. First request
 338                         // should have succeeded for class1. Second
 339                         // request should only ask for class404.
 340                         c.Check(st.requests[1].Header.Get("X-Keep-Storage-Classes"), Equals, "class404")
 341                 }
 342         }
 343 }
 344
 345 type FailHandler struct {
 346         handled chan string
 347 }
 348
 349 func (fh FailHandler) ServeHTTP(resp http.ResponseWriter, req *http.Request) {
 350         resp.WriteHeader(500)
 351         fh.handled <- fmt.Sprintf("http://%s", req.Host)
 352 }
 353
 354 type FailThenSucceedHandler struct {
 355         handled        chan string
 356         count          int
 357         successhandler http.Handler
 358         reqIDs         []string
 359 }
 360
 361 func (fh *FailThenSucceedHandler) ServeHTTP(resp http.ResponseWriter, req *http.Request) {
 362         fh.reqIDs = append(fh.reqIDs, req.Header.Get("X-Request-Id"))
 363         if fh.count == 0 {
 364                 resp.WriteHeader(500)
 365                 fh.count++
 366                 fh.handled <- fmt.Sprintf("http://%s", req.Host)
 367         } else {
 368                 fh.successhandler.ServeHTTP(resp, req)
 369         }
 370 }
 371
 372 type Error404Handler struct {
 373         handled chan string
 374 }
 375
 376 func (fh Error404Handler) ServeHTTP(resp http.ResponseWriter, req *http.Request) {
 377         resp.WriteHeader(404)
 378         fh.handled <- fmt.Sprintf("http://%s", req.Host)
 379 }
 380
 381 func (s *StandaloneSuite) TestFailedUploadToStubKeepServer(c *C) {
 382         st := FailHandler{
 383                 make(chan string)}
 384
 385         hash := "acbd18db4cc2f85cedef654fccc4a4d8"
 386
 387         UploadToStubHelper(c, st,
 388                 func(kc *KeepClient, url string, reader io.ReadCloser,
 389                         writer io.WriteCloser, uploadStatusChan chan uploadStatus) {
 390
 391                         go kc.uploadToKeepServer(url, hash, nil, reader, uploadStatusChan, 3, kc.getRequestID())
 392
 393                         writer.Write([]byte("foo"))
 394                         writer.Close()
 395
 396                         <-st.handled
 397
 398                         status := <-uploadStatusChan
 399                         c.Check(status.url, Equals, fmt.Sprintf("%s/%s", url, hash))
 400                         c.Check(status.statusCode, Equals, 500)
 401                 })
 402 }
 403
 404 type KeepServer struct {
 405         listener net.Listener
 406         url      string
 407 }
 408
 409 func RunSomeFakeKeepServers(st http.Handler, n int) (ks []KeepServer) {
 410         ks = make([]KeepServer, n)
 411
 412         for i := 0; i < n; i++ {
 413                 ks[i] = RunFakeKeepServer(st)
 414         }
 415
 416         return ks
 417 }
 418
 419 func (s *StandaloneSuite) TestPutB(c *C) {
 420         hash := Md5String("foo")
 421
 422         st := &StubPutHandler{
 423                 c:                    c,
 424                 expectPath:           hash,
 425                 expectAPIToken:       "abc123",
 426                 expectBody:           "foo",
 427                 expectStorageClass:   "default",
 428                 returnStorageClasses: "",
 429                 handled:              make(chan string, 5),
 430         }
 431
 432         arv, _ := arvadosclient.MakeArvadosClient()
 433         kc, _ := MakeKeepClient(arv)
 434
 435         kc.Want_replicas = 2
 436         arv.ApiToken = "abc123"
 437         localRoots := make(map[string]string)
 438         writableLocalRoots := make(map[string]string)
 439
 440         ks := RunSomeFakeKeepServers(st, 5)
 441
 442         for i, k := range ks {
 443                 localRoots[fmt.Sprintf("zzzzz-bi6l4-fakefakefake%03d", i)] = k.url
 444                 writableLocalRoots[fmt.Sprintf("zzzzz-bi6l4-fakefakefake%03d", i)] = k.url
 445                 defer k.listener.Close()
 446         }
 447
 448         kc.SetServiceRoots(localRoots, writableLocalRoots, nil)
 449
 450         kc.PutB([]byte("foo"))
 451
 452         shuff := NewRootSorter(
 453                 kc.LocalRoots(), Md5String("foo")).GetSortedRoots()
 454
 455         s1 := <-st.handled
 456         s2 := <-st.handled
 457         c.Check((s1 == shuff[0] && s2 == shuff[1]) ||
 458                 (s1 == shuff[1] && s2 == shuff[0]),
 459                 Equals,
 460                 true)
 461 }
 462
 463 func (s *StandaloneSuite) TestPutHR(c *C) {
 464         hash := fmt.Sprintf("%x", md5.Sum([]byte("foo")))
 465
 466         st := &StubPutHandler{
 467                 c:                    c,
 468                 expectPath:           hash,
 469                 expectAPIToken:       "abc123",
 470                 expectBody:           "foo",
 471                 expectStorageClass:   "default",
 472                 returnStorageClasses: "",
 473                 handled:              make(chan string, 5),
 474         }
 475
 476         arv, _ := arvadosclient.MakeArvadosClient()
 477         kc, _ := MakeKeepClient(arv)
 478
 479         kc.Want_replicas = 2
 480         arv.ApiToken = "abc123"
 481         localRoots := make(map[string]string)
 482         writableLocalRoots := make(map[string]string)
 483
 484         ks := RunSomeFakeKeepServers(st, 5)
 485
 486         for i, k := range ks {
 487                 localRoots[fmt.Sprintf("zzzzz-bi6l4-fakefakefake%03d", i)] = k.url
 488                 writableLocalRoots[fmt.Sprintf("zzzzz-bi6l4-fakefakefake%03d", i)] = k.url
 489                 defer k.listener.Close()
 490         }
 491
 492         kc.SetServiceRoots(localRoots, writableLocalRoots, nil)
 493
 494         reader, writer := io.Pipe()
 495
 496         go func() {
 497                 writer.Write([]byte("foo"))
 498                 writer.Close()
 499         }()
 500
 501         kc.PutHR(hash, reader, 3)
 502
 503         shuff := NewRootSorter(kc.LocalRoots(), hash).GetSortedRoots()
 504
 505         s1 := <-st.handled
 506         s2 := <-st.handled
 507
 508         c.Check((s1 == shuff[0] && s2 == shuff[1]) ||
 509                 (s1 == shuff[1] && s2 == shuff[0]),
 510                 Equals,
 511                 true)
 512 }
 513
 514 func (s *StandaloneSuite) TestPutWithFail(c *C) {
 515         hash := fmt.Sprintf("%x", md5.Sum([]byte("foo")))
 516
 517         st := &StubPutHandler{
 518                 c:                    c,
 519                 expectPath:           hash,
 520                 expectAPIToken:       "abc123",
 521                 expectBody:           "foo",
 522                 expectStorageClass:   "default",
 523                 returnStorageClasses: "",
 524                 handled:              make(chan string, 4),
 525         }
 526
 527         fh := FailHandler{
 528                 make(chan string, 1)}
 529
 530         arv, err := arvadosclient.MakeArvadosClient()
 531         c.Check(err, IsNil)
 532         kc, _ := MakeKeepClient(arv)
 533
 534         kc.Want_replicas = 2
 535         arv.ApiToken = "abc123"
 536         localRoots := make(map[string]string)
 537         writableLocalRoots := make(map[string]string)
 538
 539         ks1 := RunSomeFakeKeepServers(st, 4)
 540         ks2 := RunSomeFakeKeepServers(fh, 1)
 541
 542         for i, k := range ks1 {
 543                 localRoots[fmt.Sprintf("zzzzz-bi6l4-fakefakefake%03d", i)] = k.url
 544                 writableLocalRoots[fmt.Sprintf("zzzzz-bi6l4-fakefakefake%03d", i)] = k.url
 545                 defer k.listener.Close()
 546         }
 547         for i, k := range ks2 {
 548                 localRoots[fmt.Sprintf("zzzzz-bi6l4-fakefakefake%03d", i+len(ks1))] = k.url
 549                 writableLocalRoots[fmt.Sprintf("zzzzz-bi6l4-fakefakefake%03d", i+len(ks1))] = k.url
 550                 defer k.listener.Close()
 551         }
 552
 553         kc.SetServiceRoots(localRoots, writableLocalRoots, nil)
 554
 555         shuff := NewRootSorter(
 556                 kc.LocalRoots(), Md5String("foo")).GetSortedRoots()
 557         c.Logf("%+v", shuff)
 558
 559         phash, replicas, err := kc.PutB([]byte("foo"))
 560
 561         <-fh.handled
 562
 563         c.Check(err, Equals, nil)
 564         c.Check(phash, Equals, "")
 565         c.Check(replicas, Equals, 2)
 566
 567         s1 := <-st.handled
 568         s2 := <-st.handled
 569
 570         c.Check((s1 == shuff[1] && s2 == shuff[2]) ||
 571                 (s1 == shuff[2] && s2 == shuff[1]),
 572                 Equals,
 573                 true)
 574 }
 575
 576 func (s *StandaloneSuite) TestPutWithTooManyFail(c *C) {
 577         hash := fmt.Sprintf("%x", md5.Sum([]byte("foo")))
 578
 579         st := &StubPutHandler{
 580                 c:                    c,
 581                 expectPath:           hash,
 582                 expectAPIToken:       "abc123",
 583                 expectBody:           "foo",
 584                 expectStorageClass:   "default",
 585                 returnStorageClasses: "",
 586                 handled:              make(chan string, 1),
 587         }
 588
 589         fh := FailHandler{
 590                 make(chan string, 4)}
 591
 592         arv, err := arvadosclient.MakeArvadosClient()
 593         c.Check(err, IsNil)
 594         kc, _ := MakeKeepClient(arv)
 595
 596         kc.Want_replicas = 2
 597         kc.Retries = 0
 598         arv.ApiToken = "abc123"
 599         localRoots := make(map[string]string)
 600         writableLocalRoots := make(map[string]string)
 601
 602         ks1 := RunSomeFakeKeepServers(st, 1)
 603         ks2 := RunSomeFakeKeepServers(fh, 4)
 604
 605         for i, k := range ks1 {
 606                 localRoots[fmt.Sprintf("zzzzz-bi6l4-fakefakefake%03d", i)] = k.url
 607                 writableLocalRoots[fmt.Sprintf("zzzzz-bi6l4-fakefakefake%03d", i)] = k.url
 608                 defer k.listener.Close()
 609         }
 610         for i, k := range ks2 {
 611                 localRoots[fmt.Sprintf("zzzzz-bi6l4-fakefakefake%03d", i+len(ks1))] = k.url
 612                 writableLocalRoots[fmt.Sprintf("zzzzz-bi6l4-fakefakefake%03d", i+len(ks1))] = k.url
 613                 defer k.listener.Close()
 614         }
 615
 616         kc.SetServiceRoots(localRoots, writableLocalRoots, nil)
 617
 618         _, replicas, err := kc.PutB([]byte("foo"))
 619
 620         c.Check(err, FitsTypeOf, InsufficientReplicasError{})
 621         c.Check(replicas, Equals, 1)
 622         c.Check(<-st.handled, Equals, ks1[0].url)
 623 }
 624
 625 type StubGetHandler struct {
 626         c              *C
 627         expectPath     string
 628         expectAPIToken string
 629         httpStatus     int
 630         body           []byte
 631 }
 632
 633 func (sgh StubGetHandler) ServeHTTP(resp http.ResponseWriter, req *http.Request) {
 634         sgh.c.Check(req.URL.Path, Equals, "/"+sgh.expectPath)
 635         sgh.c.Check(req.Header.Get("Authorization"), Equals, fmt.Sprintf("OAuth2 %s", sgh.expectAPIToken))
 636         resp.WriteHeader(sgh.httpStatus)
 637         resp.Header().Set("Content-Length", fmt.Sprintf("%d", len(sgh.body)))
 638         resp.Write(sgh.body)
 639 }
 640
 641 func (s *StandaloneSuite) TestGet(c *C) {
 642         hash := fmt.Sprintf("%x", md5.Sum([]byte("foo")))
 643
 644         st := StubGetHandler{
 645                 c,
 646                 hash,
 647                 "abc123",
 648                 http.StatusOK,
 649                 []byte("foo")}
 650
 651         ks := RunFakeKeepServer(st)
 652         defer ks.listener.Close()
 653
 654         arv, err := arvadosclient.MakeArvadosClient()
 655         c.Check(err, IsNil)
 656         kc, _ := MakeKeepClient(arv)
 657         arv.ApiToken = "abc123"
 658         kc.SetServiceRoots(map[string]string{"x": ks.url}, nil, nil)
 659
 660         r, n, url2, err := kc.Get(hash)
 661         defer r.Close()
 662         c.Check(err, Equals, nil)
 663         c.Check(n, Equals, int64(3))
 664         c.Check(url2, Equals, fmt.Sprintf("%s/%s", ks.url, hash))
 665
 666         content, err2 := ioutil.ReadAll(r)
 667         c.Check(err2, Equals, nil)
 668         c.Check(content, DeepEquals, []byte("foo"))
 669 }
 670
 671 func (s *StandaloneSuite) TestGet404(c *C) {
 672         hash := fmt.Sprintf("%x", md5.Sum([]byte("foo")))
 673
 674         st := Error404Handler{make(chan string, 1)}
 675
 676         ks := RunFakeKeepServer(st)
 677         defer ks.listener.Close()
 678
 679         arv, err := arvadosclient.MakeArvadosClient()
 680         c.Check(err, IsNil)
 681         kc, _ := MakeKeepClient(arv)
 682         arv.ApiToken = "abc123"
 683         kc.SetServiceRoots(map[string]string{"x": ks.url}, nil, nil)
 684
 685         r, n, url2, err := kc.Get(hash)
 686         c.Check(err, Equals, BlockNotFound)
 687         c.Check(n, Equals, int64(0))
 688         c.Check(url2, Equals, "")
 689         c.Check(r, Equals, nil)
 690 }
 691
 692 func (s *StandaloneSuite) TestGetEmptyBlock(c *C) {
 693         st := Error404Handler{make(chan string, 1)}
 694
 695         ks := RunFakeKeepServer(st)
 696         defer ks.listener.Close()
 697
 698         arv, err := arvadosclient.MakeArvadosClient()
 699         c.Check(err, IsNil)
 700         kc, _ := MakeKeepClient(arv)
 701         arv.ApiToken = "abc123"
 702         kc.SetServiceRoots(map[string]string{"x": ks.url}, nil, nil)
 703
 704         r, n, url2, err := kc.Get("d41d8cd98f00b204e9800998ecf8427e+0")
 705         c.Check(err, IsNil)
 706         c.Check(n, Equals, int64(0))
 707         c.Check(url2, Equals, "")
 708         c.Assert(r, NotNil)
 709         buf, err := ioutil.ReadAll(r)
 710         c.Check(err, IsNil)
 711         c.Check(buf, DeepEquals, []byte{})
 712 }
 713
 714 func (s *StandaloneSuite) TestGetFail(c *C) {
 715         hash := fmt.Sprintf("%x", md5.Sum([]byte("foo")))
 716
 717         st := FailHandler{make(chan string, 1)}
 718
 719         ks := RunFakeKeepServer(st)
 720         defer ks.listener.Close()
 721
 722         arv, err := arvadosclient.MakeArvadosClient()
 723         c.Check(err, IsNil)
 724         kc, _ := MakeKeepClient(arv)
 725         arv.ApiToken = "abc123"
 726         kc.SetServiceRoots(map[string]string{"x": ks.url}, nil, nil)
 727         kc.Retries = 0
 728
 729         r, n, url2, err := kc.Get(hash)
 730         errNotFound, _ := err.(*ErrNotFound)
 731         c.Check(errNotFound, NotNil)
 732         c.Check(strings.Contains(errNotFound.Error(), "HTTP 500"), Equals, true)
 733         c.Check(errNotFound.Temporary(), Equals, true)
 734         c.Check(n, Equals, int64(0))
 735         c.Check(url2, Equals, "")
 736         c.Check(r, Equals, nil)
 737 }
 738
 739 func (s *StandaloneSuite) TestGetFailRetry(c *C) {
 740         hash := fmt.Sprintf("%x", md5.Sum([]byte("foo")))
 741
 742         st := &FailThenSucceedHandler{
 743                 handled: make(chan string, 1),
 744                 successhandler: StubGetHandler{
 745                         c,
 746                         hash,
 747                         "abc123",
 748                         http.StatusOK,
 749                         []byte("foo")}}
 750
 751         ks := RunFakeKeepServer(st)
 752         defer ks.listener.Close()
 753
 754         arv, err := arvadosclient.MakeArvadosClient()
 755         c.Check(err, IsNil)
 756         kc, _ := MakeKeepClient(arv)
 757         arv.ApiToken = "abc123"
 758         kc.SetServiceRoots(map[string]string{"x": ks.url}, nil, nil)
 759
 760         r, n, url2, err := kc.Get(hash)
 761         defer r.Close()
 762         c.Check(err, Equals, nil)
 763         c.Check(n, Equals, int64(3))
 764         c.Check(url2, Equals, fmt.Sprintf("%s/%s", ks.url, hash))
 765
 766         content, err2 := ioutil.ReadAll(r)
 767         c.Check(err2, Equals, nil)
 768         c.Check(content, DeepEquals, []byte("foo"))
 769
 770         c.Logf("%q", st.reqIDs)
 771         c.Assert(len(st.reqIDs) > 1, Equals, true)
 772         for _, reqid := range st.reqIDs {
 773                 c.Check(reqid, Not(Equals), "")
 774                 c.Check(reqid, Equals, st.reqIDs[0])
 775         }
 776 }
 777
 778 func (s *StandaloneSuite) TestGetNetError(c *C) {
 779         hash := fmt.Sprintf("%x", md5.Sum([]byte("foo")))
 780
 781         arv, err := arvadosclient.MakeArvadosClient()
 782         c.Check(err, IsNil)
 783         kc, _ := MakeKeepClient(arv)
 784         arv.ApiToken = "abc123"
 785         kc.SetServiceRoots(map[string]string{"x": "http://localhost:62222"}, nil, nil)
 786
 787         r, n, url2, err := kc.Get(hash)
 788         errNotFound, _ := err.(*ErrNotFound)
 789         c.Check(errNotFound, NotNil)
 790         c.Check(strings.Contains(errNotFound.Error(), "connection refused"), Equals, true)
 791         c.Check(errNotFound.Temporary(), Equals, true)
 792         c.Check(n, Equals, int64(0))
 793         c.Check(url2, Equals, "")
 794         c.Check(r, Equals, nil)
 795 }
 796
 797 func (s *StandaloneSuite) TestGetWithServiceHint(c *C) {
 798         uuid := "zzzzz-bi6l4-123451234512345"
 799         hash := fmt.Sprintf("%x", md5.Sum([]byte("foo")))
 800
 801         // This one shouldn't be used:
 802         ks0 := RunFakeKeepServer(StubGetHandler{
 803                 c,
 804                 "error if used",
 805                 "abc123",
 806                 http.StatusOK,
 807                 []byte("foo")})
 808         defer ks0.listener.Close()
 809         // This one should be used:
 810         ks := RunFakeKeepServer(StubGetHandler{
 811                 c,
 812                 hash + "+K@" + uuid,
 813                 "abc123",
 814                 http.StatusOK,
 815                 []byte("foo")})
 816         defer ks.listener.Close()
 817
 818         arv, err := arvadosclient.MakeArvadosClient()
 819         c.Check(err, IsNil)
 820         kc, _ := MakeKeepClient(arv)
 821         arv.ApiToken = "abc123"
 822         kc.SetServiceRoots(
 823                 map[string]string{"x": ks0.url},
 824                 nil,
 825                 map[string]string{uuid: ks.url})
 826
 827         r, n, uri, err := kc.Get(hash + "+K@" + uuid)
 828         defer r.Close()
 829         c.Check(err, Equals, nil)
 830         c.Check(n, Equals, int64(3))
 831         c.Check(uri, Equals, fmt.Sprintf("%s/%s", ks.url, hash+"+K@"+uuid))
 832
 833         content, err := ioutil.ReadAll(r)
 834         c.Check(err, Equals, nil)
 835         c.Check(content, DeepEquals, []byte("foo"))
 836 }
 837
 838 // Use a service hint to fetch from a local disk service, overriding
 839 // rendezvous probe order.
 840 func (s *StandaloneSuite) TestGetWithLocalServiceHint(c *C) {
 841         uuid := "zzzzz-bi6l4-zzzzzzzzzzzzzzz"
 842         hash := fmt.Sprintf("%x", md5.Sum([]byte("foo")))
 843
 844         // This one shouldn't be used, although it appears first in
 845         // rendezvous probe order:
 846         ks0 := RunFakeKeepServer(StubGetHandler{
 847                 c,
 848                 "error if used",
 849                 "abc123",
 850                 http.StatusOK,
 851                 []byte("foo")})
 852         defer ks0.listener.Close()
 853         // This one should be used:
 854         ks := RunFakeKeepServer(StubGetHandler{
 855                 c,
 856                 hash + "+K@" + uuid,
 857                 "abc123",
 858                 http.StatusOK,
 859                 []byte("foo")})
 860         defer ks.listener.Close()
 861
 862         arv, err := arvadosclient.MakeArvadosClient()
 863         c.Check(err, IsNil)
 864         kc, _ := MakeKeepClient(arv)
 865         arv.ApiToken = "abc123"
 866         kc.SetServiceRoots(
 867                 map[string]string{
 868                         "zzzzz-bi6l4-yyyyyyyyyyyyyyy": ks0.url,
 869                         "zzzzz-bi6l4-xxxxxxxxxxxxxxx": ks0.url,
 870                         "zzzzz-bi6l4-wwwwwwwwwwwwwww": ks0.url,
 871                         uuid:                          ks.url},
 872                 nil,
 873                 map[string]string{
 874                         "zzzzz-bi6l4-yyyyyyyyyyyyyyy": ks0.url,
 875                         "zzzzz-bi6l4-xxxxxxxxxxxxxxx": ks0.url,
 876                         "zzzzz-bi6l4-wwwwwwwwwwwwwww": ks0.url,
 877                         uuid:                          ks.url},
 878         )
 879
 880         r, n, uri, err := kc.Get(hash + "+K@" + uuid)
 881         defer r.Close()
 882         c.Check(err, Equals, nil)
 883         c.Check(n, Equals, int64(3))
 884         c.Check(uri, Equals, fmt.Sprintf("%s/%s", ks.url, hash+"+K@"+uuid))
 885
 886         content, err := ioutil.ReadAll(r)
 887         c.Check(err, Equals, nil)
 888         c.Check(content, DeepEquals, []byte("foo"))
 889 }
 890
 891 func (s *StandaloneSuite) TestGetWithServiceHintFailoverToLocals(c *C) {
 892         uuid := "zzzzz-bi6l4-123451234512345"
 893         hash := fmt.Sprintf("%x", md5.Sum([]byte("foo")))
 894
 895         ksLocal := RunFakeKeepServer(StubGetHandler{
 896                 c,
 897                 hash + "+K@" + uuid,
 898                 "abc123",
 899                 http.StatusOK,
 900                 []byte("foo")})
 901         defer ksLocal.listener.Close()
 902         ksGateway := RunFakeKeepServer(StubGetHandler{
 903                 c,
 904                 hash + "+K@" + uuid,
 905                 "abc123",
 906                 http.StatusInternalServerError,
 907                 []byte("Error")})
 908         defer ksGateway.listener.Close()
 909
 910         arv, err := arvadosclient.MakeArvadosClient()
 911         c.Check(err, IsNil)
 912         kc, _ := MakeKeepClient(arv)
 913         arv.ApiToken = "abc123"
 914         kc.SetServiceRoots(
 915                 map[string]string{"zzzzz-bi6l4-keepdisk0000000": ksLocal.url},
 916                 nil,
 917                 map[string]string{uuid: ksGateway.url})
 918
 919         r, n, uri, err := kc.Get(hash + "+K@" + uuid)
 920         c.Assert(err, Equals, nil)
 921         defer r.Close()
 922         c.Check(n, Equals, int64(3))
 923         c.Check(uri, Equals, fmt.Sprintf("%s/%s", ksLocal.url, hash+"+K@"+uuid))
 924
 925         content, err := ioutil.ReadAll(r)
 926         c.Check(err, Equals, nil)
 927         c.Check(content, DeepEquals, []byte("foo"))
 928 }
 929
 930 type BarHandler struct {
 931         handled chan string
 932 }
 933
 934 func (h BarHandler) ServeHTTP(resp http.ResponseWriter, req *http.Request) {
 935         resp.Write([]byte("bar"))
 936         h.handled <- fmt.Sprintf("http://%s", req.Host)
 937 }
 938
 939 func (s *StandaloneSuite) TestChecksum(c *C) {
 940         foohash := fmt.Sprintf("%x", md5.Sum([]byte("foo")))
 941         barhash := fmt.Sprintf("%x", md5.Sum([]byte("bar")))
 942
 943         st := BarHandler{make(chan string, 1)}
 944
 945         ks := RunFakeKeepServer(st)
 946         defer ks.listener.Close()
 947
 948         arv, err := arvadosclient.MakeArvadosClient()
 949         c.Check(err, IsNil)
 950         kc, _ := MakeKeepClient(arv)
 951         arv.ApiToken = "abc123"
 952         kc.SetServiceRoots(map[string]string{"x": ks.url}, nil, nil)
 953
 954         r, n, _, err := kc.Get(barhash)
 955         c.Check(err, IsNil)
 956         _, err = ioutil.ReadAll(r)
 957         c.Check(n, Equals, int64(3))
 958         c.Check(err, Equals, nil)
 959
 960         <-st.handled
 961
 962         r, n, _, err = kc.Get(foohash)
 963         c.Check(err, IsNil)
 964         _, err = ioutil.ReadAll(r)
 965         c.Check(n, Equals, int64(3))
 966         c.Check(err, Equals, BadChecksum)
 967
 968         <-st.handled
 969 }
 970
 971 func (s *StandaloneSuite) TestGetWithFailures(c *C) {
 972         content := []byte("waz")
 973         hash := fmt.Sprintf("%x", md5.Sum(content))
 974
 975         fh := Error404Handler{
 976                 make(chan string, 4)}
 977
 978         st := StubGetHandler{
 979                 c,
 980                 hash,
 981                 "abc123",
 982                 http.StatusOK,
 983                 content}
 984
 985         arv, err := arvadosclient.MakeArvadosClient()
 986         c.Check(err, IsNil)
 987         kc, _ := MakeKeepClient(arv)
 988         arv.ApiToken = "abc123"
 989         localRoots := make(map[string]string)
 990         writableLocalRoots := make(map[string]string)
 991
 992         ks1 := RunSomeFakeKeepServers(st, 1)
 993         ks2 := RunSomeFakeKeepServers(fh, 4)
 994
 995         for i, k := range ks1 {
 996                 localRoots[fmt.Sprintf("zzzzz-bi6l4-fakefakefake%03d", i)] = k.url
 997                 writableLocalRoots[fmt.Sprintf("zzzzz-bi6l4-fakefakefake%03d", i)] = k.url
 998                 defer k.listener.Close()
 999         }
1000         for i, k := range ks2 {
1001                 localRoots[fmt.Sprintf("zzzzz-bi6l4-fakefakefake%03d", i+len(ks1))] = k.url
1002                 writableLocalRoots[fmt.Sprintf("zzzzz-bi6l4-fakefakefake%03d", i+len(ks1))] = k.url
1003                 defer k.listener.Close()
1004         }
1005
1006         kc.SetServiceRoots(localRoots, writableLocalRoots, nil)
1007         kc.Retries = 0
1008
1009         // This test works only if one of the failing services is
1010         // attempted before the succeeding service. Otherwise,
1011         // <-fh.handled below will just hang! (Probe order depends on
1012         // the choice of block content "waz" and the UUIDs of the fake
1013         // servers, so we just tried different strings until we found
1014         // an example that passes this Assert.)
1015         c.Assert(NewRootSorter(localRoots, hash).GetSortedRoots()[0], Not(Equals), ks1[0].url)
1016
1017         r, n, url2, err := kc.Get(hash)
1018
1019         <-fh.handled
1020         c.Check(err, Equals, nil)
1021         c.Check(n, Equals, int64(3))
1022         c.Check(url2, Equals, fmt.Sprintf("%s/%s", ks1[0].url, hash))
1023
1024         readContent, err2 := ioutil.ReadAll(r)
1025         c.Check(err2, Equals, nil)
1026         c.Check(readContent, DeepEquals, content)
1027 }
1028
1029 func (s *ServerRequiredSuite) TestPutGetHead(c *C) {
1030         content := []byte("TestPutGetHead")
1031
1032         arv, err := arvadosclient.MakeArvadosClient()
1033         c.Check(err, IsNil)
1034         kc, err := MakeKeepClient(arv)
1035         c.Assert(err, Equals, nil)
1036
1037         hash := fmt.Sprintf("%x", md5.Sum(content))
1038
1039         {
1040                 n, _, err := kc.Ask(hash)
1041                 c.Check(err, Equals, BlockNotFound)
1042                 c.Check(n, Equals, int64(0))
1043         }
1044         {
1045                 hash2, replicas, err := kc.PutB(content)
1046                 c.Check(hash2, Matches, fmt.Sprintf(`%s\+%d\b.*`, hash, len(content)))
1047                 c.Check(replicas, Equals, 2)
1048                 c.Check(err, Equals, nil)
1049         }
1050         {
1051                 r, n, url2, err := kc.Get(hash)
1052                 c.Check(err, Equals, nil)
1053                 c.Check(n, Equals, int64(len(content)))
1054                 c.Check(url2, Matches, fmt.Sprintf("http://localhost:\\d+/%s", hash))
1055
1056                 readContent, err2 := ioutil.ReadAll(r)
1057                 c.Check(err2, Equals, nil)
1058                 c.Check(readContent, DeepEquals, content)
1059         }
1060         {
1061                 n, url2, err := kc.Ask(hash)
1062                 c.Check(err, Equals, nil)
1063                 c.Check(n, Equals, int64(len(content)))
1064                 c.Check(url2, Matches, fmt.Sprintf("http://localhost:\\d+/%s", hash))
1065         }
1066         {
1067                 loc, err := kc.LocalLocator(hash)
1068                 c.Check(err, Equals, nil)
1069                 c.Assert(len(loc) >= 32, Equals, true)
1070                 c.Check(loc[:32], Equals, hash[:32])
1071         }
1072         {
1073                 content := []byte("the perth county conspiracy")
1074                 loc, err := kc.LocalLocator(fmt.Sprintf("%x+%d+Rzaaaa-abcde@12345", md5.Sum(content), len(content)))
1075                 c.Check(loc, Equals, "")
1076                 c.Check(err, ErrorMatches, `.*HEAD .*\+R.*`)
1077                 c.Check(err, ErrorMatches, `.*HTTP 400.*`)
1078         }
1079 }
1080
1081 type StubProxyHandler struct {
1082         handled chan string
1083 }
1084
1085 func (h StubProxyHandler) ServeHTTP(resp http.ResponseWriter, req *http.Request) {
1086         resp.Header().Set("X-Keep-Replicas-Stored", "2")
1087         h.handled <- fmt.Sprintf("http://%s", req.Host)
1088 }
1089
1090 func (s *StandaloneSuite) TestPutProxy(c *C) {
1091         st := StubProxyHandler{make(chan string, 1)}
1092
1093         arv, err := arvadosclient.MakeArvadosClient()
1094         c.Check(err, IsNil)
1095         kc, _ := MakeKeepClient(arv)
1096
1097         kc.Want_replicas = 2
1098         arv.ApiToken = "abc123"
1099         localRoots := make(map[string]string)
1100         writableLocalRoots := make(map[string]string)
1101
1102         ks1 := RunSomeFakeKeepServers(st, 1)
1103
1104         for i, k := range ks1 {
1105                 localRoots[fmt.Sprintf("zzzzz-bi6l4-fakefakefake%03d", i)] = k.url
1106                 writableLocalRoots[fmt.Sprintf("zzzzz-bi6l4-fakefakefake%03d", i)] = k.url
1107                 defer k.listener.Close()
1108         }
1109
1110         kc.SetServiceRoots(localRoots, writableLocalRoots, nil)
1111
1112         _, replicas, err := kc.PutB([]byte("foo"))
1113         <-st.handled
1114
1115         c.Check(err, Equals, nil)
1116         c.Check(replicas, Equals, 2)
1117 }
1118
1119 func (s *StandaloneSuite) TestPutProxyInsufficientReplicas(c *C) {
1120         st := StubProxyHandler{make(chan string, 1)}
1121
1122         arv, err := arvadosclient.MakeArvadosClient()
1123         c.Check(err, IsNil)
1124         kc, _ := MakeKeepClient(arv)
1125
1126         kc.Want_replicas = 3
1127         arv.ApiToken = "abc123"
1128         localRoots := make(map[string]string)
1129         writableLocalRoots := make(map[string]string)
1130
1131         ks1 := RunSomeFakeKeepServers(st, 1)
1132
1133         for i, k := range ks1 {
1134                 localRoots[fmt.Sprintf("zzzzz-bi6l4-fakefakefake%03d", i)] = k.url
1135                 writableLocalRoots[fmt.Sprintf("zzzzz-bi6l4-fakefakefake%03d", i)] = k.url
1136                 defer k.listener.Close()
1137         }
1138         kc.SetServiceRoots(localRoots, writableLocalRoots, nil)
1139
1140         _, replicas, err := kc.PutB([]byte("foo"))
1141         <-st.handled
1142
1143         c.Check(err, FitsTypeOf, InsufficientReplicasError{})
1144         c.Check(replicas, Equals, 2)
1145 }
1146
1147 func (s *StandaloneSuite) TestMakeLocator(c *C) {
1148         l, err := MakeLocator("91f372a266fe2bf2823cb8ec7fda31ce+3+Aabcde@12345678")
1149         c.Check(err, Equals, nil)
1150         c.Check(l.Hash, Equals, "91f372a266fe2bf2823cb8ec7fda31ce")
1151         c.Check(l.Size, Equals, 3)
1152         c.Check(l.Hints, DeepEquals, []string{"3", "Aabcde@12345678"})
1153 }
1154
1155 func (s *StandaloneSuite) TestMakeLocatorNoHints(c *C) {
1156         l, err := MakeLocator("91f372a266fe2bf2823cb8ec7fda31ce")
1157         c.Check(err, Equals, nil)
1158         c.Check(l.Hash, Equals, "91f372a266fe2bf2823cb8ec7fda31ce")
1159         c.Check(l.Size, Equals, -1)
1160         c.Check(l.Hints, DeepEquals, []string{})
1161 }
1162
1163 func (s *StandaloneSuite) TestMakeLocatorNoSizeHint(c *C) {
1164         l, err := MakeLocator("91f372a266fe2bf2823cb8ec7fda31ce+Aabcde@12345678")
1165         c.Check(err, Equals, nil)
1166         c.Check(l.Hash, Equals, "91f372a266fe2bf2823cb8ec7fda31ce")
1167         c.Check(l.Size, Equals, -1)
1168         c.Check(l.Hints, DeepEquals, []string{"Aabcde@12345678"})
1169 }
1170
1171 func (s *StandaloneSuite) TestMakeLocatorPreservesUnrecognizedHints(c *C) {
1172         str := "91f372a266fe2bf2823cb8ec7fda31ce+3+Unknown+Kzzzzz+Afoobar"
1173         l, err := MakeLocator(str)
1174         c.Check(err, Equals, nil)
1175         c.Check(l.Hash, Equals, "91f372a266fe2bf2823cb8ec7fda31ce")
1176         c.Check(l.Size, Equals, 3)
1177         c.Check(l.Hints, DeepEquals, []string{"3", "Unknown", "Kzzzzz", "Afoobar"})
1178         c.Check(l.String(), Equals, str)
1179 }
1180
1181 func (s *StandaloneSuite) TestMakeLocatorInvalidInput(c *C) {
1182         _, err := MakeLocator("91f372a266fe2bf2823cb8ec7fda31c")
1183         c.Check(err, Equals, InvalidLocatorError)
1184 }
1185
1186 func (s *StandaloneSuite) TestPutBWant2ReplicasWithOnlyOneWritableLocalRoot(c *C) {
1187         hash := Md5String("foo")
1188
1189         st := &StubPutHandler{
1190                 c:                    c,
1191                 expectPath:           hash,
1192                 expectAPIToken:       "abc123",
1193                 expectBody:           "foo",
1194                 expectStorageClass:   "default",
1195                 returnStorageClasses: "",
1196                 handled:              make(chan string, 5),
1197         }
1198
1199         arv, _ := arvadosclient.MakeArvadosClient()
1200         kc, _ := MakeKeepClient(arv)
1201
1202         kc.Want_replicas = 2
1203         arv.ApiToken = "abc123"
1204         localRoots := make(map[string]string)
1205         writableLocalRoots := make(map[string]string)
1206
1207         ks := RunSomeFakeKeepServers(st, 5)
1208
1209         for i, k := range ks {
1210                 localRoots[fmt.Sprintf("zzzzz-bi6l4-fakefakefake%03d", i)] = k.url
1211                 if i == 0 {
1212                         writableLocalRoots[fmt.Sprintf("zzzzz-bi6l4-fakefakefake%03d", i)] = k.url
1213                 }
1214                 defer k.listener.Close()
1215         }
1216
1217         kc.SetServiceRoots(localRoots, writableLocalRoots, nil)
1218
1219         _, replicas, err := kc.PutB([]byte("foo"))
1220
1221         c.Check(err, FitsTypeOf, InsufficientReplicasError{})
1222         c.Check(replicas, Equals, 1)
1223
1224         c.Check(<-st.handled, Equals, localRoots[fmt.Sprintf("zzzzz-bi6l4-fakefakefake%03d", 0)])
1225 }
1226
1227 func (s *StandaloneSuite) TestPutBWithNoWritableLocalRoots(c *C) {
1228         hash := Md5String("foo")
1229
1230         st := &StubPutHandler{
1231                 c:                    c,
1232                 expectPath:           hash,
1233                 expectAPIToken:       "abc123",
1234                 expectBody:           "foo",
1235                 expectStorageClass:   "",
1236                 returnStorageClasses: "",
1237                 handled:              make(chan string, 5),
1238         }
1239
1240         arv, _ := arvadosclient.MakeArvadosClient()
1241         kc, _ := MakeKeepClient(arv)
1242
1243         kc.Want_replicas = 2
1244         arv.ApiToken = "abc123"
1245         localRoots := make(map[string]string)
1246         writableLocalRoots := make(map[string]string)
1247
1248         ks := RunSomeFakeKeepServers(st, 5)
1249
1250         for i, k := range ks {
1251                 localRoots[fmt.Sprintf("zzzzz-bi6l4-fakefakefake%03d", i)] = k.url
1252                 defer k.listener.Close()
1253         }
1254
1255         kc.SetServiceRoots(localRoots, writableLocalRoots, nil)
1256
1257         _, replicas, err := kc.PutB([]byte("foo"))
1258
1259         c.Check(err, FitsTypeOf, InsufficientReplicasError{})
1260         c.Check(replicas, Equals, 0)
1261 }
1262
1263 type StubGetIndexHandler struct {
1264         c              *C
1265         expectPath     string
1266         expectAPIToken string
1267         httpStatus     int
1268         body           []byte
1269 }
1270
1271 func (h StubGetIndexHandler) ServeHTTP(resp http.ResponseWriter, req *http.Request) {
1272         h.c.Check(req.URL.Path, Equals, h.expectPath)
1273         h.c.Check(req.Header.Get("Authorization"), Equals, fmt.Sprintf("OAuth2 %s", h.expectAPIToken))
1274         resp.WriteHeader(h.httpStatus)
1275         resp.Header().Set("Content-Length", fmt.Sprintf("%d", len(h.body)))
1276         resp.Write(h.body)
1277 }
1278
1279 func (s *StandaloneSuite) TestGetIndexWithNoPrefix(c *C) {
1280         hash := fmt.Sprintf("%x", md5.Sum([]byte("foo")))
1281
1282         st := StubGetIndexHandler{
1283                 c,
1284                 "/index",
1285                 "abc123",
1286                 http.StatusOK,
1287                 []byte(hash + "+3 1443559274\n\n")}
1288
1289         ks := RunFakeKeepServer(st)
1290         defer ks.listener.Close()
1291
1292         arv, err := arvadosclient.MakeArvadosClient()
1293         c.Assert(err, IsNil)
1294         kc, err := MakeKeepClient(arv)
1295         c.Assert(err, IsNil)
1296         arv.ApiToken = "abc123"
1297         kc.SetServiceRoots(map[string]string{"x": ks.url}, nil, nil)
1298
1299         r, err := kc.GetIndex("x", "")
1300         c.Check(err, IsNil)
1301
1302         content, err2 := ioutil.ReadAll(r)
1303         c.Check(err2, Equals, nil)
1304         c.Check(content, DeepEquals, st.body[0:len(st.body)-1])
1305 }
1306
1307 func (s *StandaloneSuite) TestGetIndexWithPrefix(c *C) {
1308         hash := fmt.Sprintf("%x", md5.Sum([]byte("foo")))
1309
1310         st := StubGetIndexHandler{
1311                 c,
1312                 "/index/" + hash[0:3],
1313                 "abc123",
1314                 http.StatusOK,
1315                 []byte(hash + "+3 1443559274\n\n")}
1316
1317         ks := RunFakeKeepServer(st)
1318         defer ks.listener.Close()
1319
1320         arv, err := arvadosclient.MakeArvadosClient()
1321         c.Check(err, IsNil)
1322         kc, _ := MakeKeepClient(arv)
1323         arv.ApiToken = "abc123"
1324         kc.SetServiceRoots(map[string]string{"x": ks.url}, nil, nil)
1325
1326         r, err := kc.GetIndex("x", hash[0:3])
1327         c.Assert(err, Equals, nil)
1328
1329         content, err2 := ioutil.ReadAll(r)
1330         c.Check(err2, Equals, nil)
1331         c.Check(content, DeepEquals, st.body[0:len(st.body)-1])
1332 }
1333
1334 func (s *StandaloneSuite) TestGetIndexIncomplete(c *C) {
1335         hash := fmt.Sprintf("%x", md5.Sum([]byte("foo")))
1336
1337         st := StubGetIndexHandler{
1338                 c,
1339                 "/index/" + hash[0:3],
1340                 "abc123",
1341                 http.StatusOK,
1342                 []byte(hash)}
1343
1344         ks := RunFakeKeepServer(st)
1345         defer ks.listener.Close()
1346
1347         arv, err := arvadosclient.MakeArvadosClient()
1348         c.Check(err, IsNil)
1349         kc, _ := MakeKeepClient(arv)
1350         arv.ApiToken = "abc123"
1351         kc.SetServiceRoots(map[string]string{"x": ks.url}, nil, nil)
1352
1353         _, err = kc.GetIndex("x", hash[0:3])
1354         c.Check(err, Equals, ErrIncompleteIndex)
1355 }
1356
1357 func (s *StandaloneSuite) TestGetIndexWithNoSuchServer(c *C) {
1358         hash := fmt.Sprintf("%x", md5.Sum([]byte("foo")))
1359
1360         st := StubGetIndexHandler{
1361                 c,
1362                 "/index/" + hash[0:3],
1363                 "abc123",
1364                 http.StatusOK,
1365                 []byte(hash)}
1366
1367         ks := RunFakeKeepServer(st)
1368         defer ks.listener.Close()
1369
1370         arv, err := arvadosclient.MakeArvadosClient()
1371         c.Check(err, IsNil)
1372         kc, _ := MakeKeepClient(arv)
1373         arv.ApiToken = "abc123"
1374         kc.SetServiceRoots(map[string]string{"x": ks.url}, nil, nil)
1375
1376         _, err = kc.GetIndex("y", hash[0:3])
1377         c.Check(err, Equals, ErrNoSuchKeepServer)
1378 }
1379
1380 func (s *StandaloneSuite) TestGetIndexWithNoSuchPrefix(c *C) {
1381         st := StubGetIndexHandler{
1382                 c,
1383                 "/index/abcd",
1384                 "abc123",
1385                 http.StatusOK,
1386                 []byte("\n")}
1387
1388         ks := RunFakeKeepServer(st)
1389         defer ks.listener.Close()
1390
1391         arv, err := arvadosclient.MakeArvadosClient()
1392         c.Check(err, IsNil)
1393         kc, _ := MakeKeepClient(arv)
1394         arv.ApiToken = "abc123"
1395         kc.SetServiceRoots(map[string]string{"x": ks.url}, nil, nil)
1396
1397         r, err := kc.GetIndex("x", "abcd")
1398         c.Check(err, Equals, nil)
1399
1400         content, err2 := ioutil.ReadAll(r)
1401         c.Check(err2, Equals, nil)
1402         c.Check(content, DeepEquals, st.body[0:len(st.body)-1])
1403 }
1404
1405 func (s *StandaloneSuite) TestPutBRetry(c *C) {
1406         st := &FailThenSucceedHandler{
1407                 handled: make(chan string, 1),
1408                 successhandler: &StubPutHandler{
1409                         c:                    c,
1410                         expectPath:           Md5String("foo"),
1411                         expectAPIToken:       "abc123",
1412                         expectBody:           "foo",
1413                         expectStorageClass:   "default",
1414                         returnStorageClasses: "",
1415                         handled:              make(chan string, 5),
1416                 },
1417         }
1418
1419         arv, _ := arvadosclient.MakeArvadosClient()
1420         kc, _ := MakeKeepClient(arv)
1421
1422         kc.Want_replicas = 2
1423         arv.ApiToken = "abc123"
1424         localRoots := make(map[string]string)
1425         writableLocalRoots := make(map[string]string)
1426
1427         ks := RunSomeFakeKeepServers(st, 2)
1428
1429         for i, k := range ks {
1430                 localRoots[fmt.Sprintf("zzzzz-bi6l4-fakefakefake%03d", i)] = k.url
1431                 writableLocalRoots[fmt.Sprintf("zzzzz-bi6l4-fakefakefake%03d", i)] = k.url
1432                 defer k.listener.Close()
1433         }
1434
1435         kc.SetServiceRoots(localRoots, writableLocalRoots, nil)
1436
1437         hash, replicas, err := kc.PutB([]byte("foo"))
1438
1439         c.Check(err, Equals, nil)
1440         c.Check(hash, Equals, "")
1441         c.Check(replicas, Equals, 2)
1442 }
1443
1444 func (s *ServerRequiredSuite) TestMakeKeepClientWithNonDiskTypeService(c *C) {
1445         arv, err := arvadosclient.MakeArvadosClient()
1446         c.Assert(err, Equals, nil)
1447
1448         // Add an additional "testblobstore" keepservice
1449         blobKeepService := make(arvadosclient.Dict)
1450         err = arv.Create("keep_services",
1451                 arvadosclient.Dict{"keep_service": arvadosclient.Dict{
1452                         "service_host": "localhost",
1453                         "service_port": "21321",
1454                         "service_type": "testblobstore"}},
1455                 &blobKeepService)
1456         c.Assert(err, Equals, nil)
1457         defer func() { arv.Delete("keep_services", blobKeepService["uuid"].(string), nil, nil) }()
1458         RefreshServiceDiscovery()
1459
1460         // Make a keepclient and ensure that the testblobstore is included
1461         kc, err := MakeKeepClient(arv)
1462         c.Assert(err, Equals, nil)
1463
1464         // verify kc.LocalRoots
1465         c.Check(len(kc.LocalRoots()), Equals, 3)
1466         for _, root := range kc.LocalRoots() {
1467                 c.Check(root, Matches, "http://localhost:\\d+")
1468         }
1469         c.Assert(kc.LocalRoots()[blobKeepService["uuid"].(string)], Not(Equals), "")
1470
1471         // verify kc.GatewayRoots
1472         c.Check(len(kc.GatewayRoots()), Equals, 3)
1473         for _, root := range kc.GatewayRoots() {
1474                 c.Check(root, Matches, "http://localhost:\\d+")
1475         }
1476         c.Assert(kc.GatewayRoots()[blobKeepService["uuid"].(string)], Not(Equals), "")
1477
1478         // verify kc.WritableLocalRoots
1479         c.Check(len(kc.WritableLocalRoots()), Equals, 3)
1480         for _, root := range kc.WritableLocalRoots() {
1481                 c.Check(root, Matches, "http://localhost:\\d+")
1482         }
1483         c.Assert(kc.WritableLocalRoots()[blobKeepService["uuid"].(string)], Not(Equals), "")
1484
1485         c.Assert(kc.replicasPerService, Equals, 0)
1486         c.Assert(kc.foundNonDiskSvc, Equals, true)
1487         c.Assert(kc.httpClient().(*http.Client).Timeout, Equals, 300*time.Second)
1488 }