sdk/go/keepclient/keepclient_test.go

   1 package keepclient
   2
   3 import (
   4         "crypto/md5"
   5         "flag"
   6         "fmt"
   7         "git.curoverse.com/arvados.git/sdk/go/arvadosclient"
   8         "git.curoverse.com/arvados.git/sdk/go/arvadostest"
   9         "git.curoverse.com/arvados.git/sdk/go/streamer"
  10         . "gopkg.in/check.v1"
  11         "io"
  12         "io/ioutil"
  13         "log"
  14         "net"
  15         "net/http"
  16         "os"
  17         "testing"
  18 )
  19
  20 // Gocheck boilerplate
  21 func Test(t *testing.T) {
  22         TestingT(t)
  23 }
  24
  25 // Gocheck boilerplate
  26 var _ = Suite(&ServerRequiredSuite{})
  27 var _ = Suite(&StandaloneSuite{})
  28
  29 var no_server = flag.Bool("no-server", false, "Skip 'ServerRequireSuite'")
  30
  31 // Tests that require the Keep server running
  32 type ServerRequiredSuite struct{}
  33
  34 // Standalone tests
  35 type StandaloneSuite struct{}
  36
  37 func pythonDir() string {
  38         cwd, _ := os.Getwd()
  39         return fmt.Sprintf("%s/../../python/tests", cwd)
  40 }
  41
  42 func (s *ServerRequiredSuite) SetUpSuite(c *C) {
  43         if *no_server {
  44                 c.Skip("Skipping tests that require server")
  45                 return
  46         }
  47         arvadostest.StartAPI()
  48         arvadostest.StartKeep()
  49 }
  50
  51 func (s *ServerRequiredSuite) TearDownSuite(c *C) {
  52         if *no_server {
  53                 return
  54         }
  55         arvadostest.StopKeep()
  56         arvadostest.StopAPI()
  57 }
  58
  59 func (s *ServerRequiredSuite) TestMakeKeepClient(c *C) {
  60         arv, err := arvadosclient.MakeArvadosClient()
  61         c.Assert(err, Equals, nil)
  62
  63         kc, err := MakeKeepClient(&arv)
  64
  65         c.Assert(err, Equals, nil)
  66         c.Check(len(kc.LocalRoots()), Equals, 2)
  67         for _, root := range kc.LocalRoots() {
  68                 c.Check(root, Matches, "http://localhost:\\d+")
  69         }
  70 }
  71
  72 type StubPutHandler struct {
  73         c              *C
  74         expectPath     string
  75         expectApiToken string
  76         expectBody     string
  77         handled        chan string
  78 }
  79
  80 func (sph StubPutHandler) ServeHTTP(resp http.ResponseWriter, req *http.Request) {
  81         sph.c.Check(req.URL.Path, Equals, "/"+sph.expectPath)
  82         sph.c.Check(req.Header.Get("Authorization"), Equals, fmt.Sprintf("OAuth2 %s", sph.expectApiToken))
  83         body, err := ioutil.ReadAll(req.Body)
  84         sph.c.Check(err, Equals, nil)
  85         sph.c.Check(body, DeepEquals, []byte(sph.expectBody))
  86         resp.WriteHeader(200)
  87         sph.handled <- fmt.Sprintf("http://%s", req.Host)
  88 }
  89
  90 func RunFakeKeepServer(st http.Handler) (ks KeepServer) {
  91         var err error
  92         ks.listener, err = net.ListenTCP("tcp", &net.TCPAddr{Port: 0})
  93         if err != nil {
  94                 panic(fmt.Sprintf("Could not listen on any port"))
  95         }
  96         ks.url = fmt.Sprintf("http://%s", ks.listener.Addr().String())
  97         go http.Serve(ks.listener, st)
  98         return
  99 }
 100
 101 func UploadToStubHelper(c *C, st http.Handler, f func(*KeepClient, string,
 102         io.ReadCloser, io.WriteCloser, chan uploadStatus)) {
 103
 104         ks := RunFakeKeepServer(st)
 105         defer ks.listener.Close()
 106
 107         arv, _ := arvadosclient.MakeArvadosClient()
 108         arv.ApiToken = "abc123"
 109
 110         kc, _ := MakeKeepClient(&arv)
 111
 112         reader, writer := io.Pipe()
 113         upload_status := make(chan uploadStatus)
 114
 115         f(kc, ks.url, reader, writer, upload_status)
 116 }
 117
 118 func (s *StandaloneSuite) TestUploadToStubKeepServer(c *C) {
 119         log.Printf("TestUploadToStubKeepServer")
 120
 121         st := StubPutHandler{
 122                 c,
 123                 "acbd18db4cc2f85cedef654fccc4a4d8",
 124                 "abc123",
 125                 "foo",
 126                 make(chan string)}
 127
 128         UploadToStubHelper(c, st,
 129                 func(kc *KeepClient, url string, reader io.ReadCloser,
 130                         writer io.WriteCloser, upload_status chan uploadStatus) {
 131
 132                         go kc.uploadToKeepServer(url, st.expectPath, reader, upload_status, int64(len("foo")), "TestUploadToStubKeepServer")
 133
 134                         writer.Write([]byte("foo"))
 135                         writer.Close()
 136
 137                         <-st.handled
 138                         status := <-upload_status
 139                         c.Check(status, DeepEquals, uploadStatus{nil, fmt.Sprintf("%s/%s", url, st.expectPath), 200, 1, ""})
 140                 })
 141
 142         log.Printf("TestUploadToStubKeepServer done")
 143 }
 144
 145 func (s *StandaloneSuite) TestUploadToStubKeepServerBufferReader(c *C) {
 146         log.Printf("TestUploadToStubKeepServerBufferReader")
 147
 148         st := StubPutHandler{
 149                 c,
 150                 "acbd18db4cc2f85cedef654fccc4a4d8",
 151                 "abc123",
 152                 "foo",
 153                 make(chan string)}
 154
 155         UploadToStubHelper(c, st,
 156                 func(kc *KeepClient, url string, reader io.ReadCloser,
 157                         writer io.WriteCloser, upload_status chan uploadStatus) {
 158
 159                         tr := streamer.AsyncStreamFromReader(512, reader)
 160                         defer tr.Close()
 161
 162                         br1 := tr.MakeStreamReader()
 163
 164                         go kc.uploadToKeepServer(url, st.expectPath, br1, upload_status, 3, "TestUploadToStubKeepServerBufferReader")
 165
 166                         writer.Write([]byte("foo"))
 167                         writer.Close()
 168
 169                         <-st.handled
 170
 171                         status := <-upload_status
 172                         c.Check(status, DeepEquals, uploadStatus{nil, fmt.Sprintf("%s/%s", url, st.expectPath), 200, 1, ""})
 173                 })
 174
 175         log.Printf("TestUploadToStubKeepServerBufferReader done")
 176 }
 177
 178 type FailHandler struct {
 179         handled chan string
 180 }
 181
 182 func (fh FailHandler) ServeHTTP(resp http.ResponseWriter, req *http.Request) {
 183         resp.WriteHeader(500)
 184         fh.handled <- fmt.Sprintf("http://%s", req.Host)
 185 }
 186
 187 func (s *StandaloneSuite) TestFailedUploadToStubKeepServer(c *C) {
 188         log.Printf("TestFailedUploadToStubKeepServer")
 189
 190         st := FailHandler{
 191                 make(chan string)}
 192
 193         hash := "acbd18db4cc2f85cedef654fccc4a4d8"
 194
 195         UploadToStubHelper(c, st,
 196                 func(kc *KeepClient, url string, reader io.ReadCloser,
 197                         writer io.WriteCloser, upload_status chan uploadStatus) {
 198
 199                         go kc.uploadToKeepServer(url, hash, reader, upload_status, 3, "TestFailedUploadToStubKeepServer")
 200
 201                         writer.Write([]byte("foo"))
 202                         writer.Close()
 203
 204                         <-st.handled
 205
 206                         status := <-upload_status
 207                         c.Check(status.url, Equals, fmt.Sprintf("%s/%s", url, hash))
 208                         c.Check(status.statusCode, Equals, 500)
 209                 })
 210         log.Printf("TestFailedUploadToStubKeepServer done")
 211 }
 212
 213 type KeepServer struct {
 214         listener net.Listener
 215         url      string
 216 }
 217
 218 func RunSomeFakeKeepServers(st http.Handler, n int) (ks []KeepServer) {
 219         ks = make([]KeepServer, n)
 220
 221         for i := 0; i < n; i += 1 {
 222                 ks[i] = RunFakeKeepServer(st)
 223         }
 224
 225         return ks
 226 }
 227
 228 func (s *StandaloneSuite) TestPutB(c *C) {
 229         log.Printf("TestPutB")
 230
 231         hash := Md5String("foo")
 232
 233         st := StubPutHandler{
 234                 c,
 235                 hash,
 236                 "abc123",
 237                 "foo",
 238                 make(chan string, 5)}
 239
 240         arv, _ := arvadosclient.MakeArvadosClient()
 241         kc, _ := MakeKeepClient(&arv)
 242
 243         kc.Want_replicas = 2
 244         arv.ApiToken = "abc123"
 245         localRoots := make(map[string]string)
 246         writableLocalRoots := make(map[string]string)
 247
 248         ks := RunSomeFakeKeepServers(st, 5)
 249
 250         for i, k := range ks {
 251                 localRoots[fmt.Sprintf("zzzzz-bi6l4-fakefakefake%03d", i)] = k.url
 252                 writableLocalRoots[fmt.Sprintf("zzzzz-bi6l4-fakefakefake%03d", i)] = k.url
 253                 defer k.listener.Close()
 254         }
 255
 256         kc.SetServiceRoots(localRoots, writableLocalRoots, nil)
 257
 258         kc.PutB([]byte("foo"))
 259
 260         shuff := NewRootSorter(
 261                 kc.LocalRoots(), Md5String("foo")).GetSortedRoots()
 262
 263         s1 := <-st.handled
 264         s2 := <-st.handled
 265         c.Check((s1 == shuff[0] && s2 == shuff[1]) ||
 266                 (s1 == shuff[1] && s2 == shuff[0]),
 267                 Equals,
 268                 true)
 269
 270         log.Printf("TestPutB done")
 271 }
 272
 273 func (s *StandaloneSuite) TestPutHR(c *C) {
 274         log.Printf("TestPutHR")
 275
 276         hash := fmt.Sprintf("%x", md5.Sum([]byte("foo")))
 277
 278         st := StubPutHandler{
 279                 c,
 280                 hash,
 281                 "abc123",
 282                 "foo",
 283                 make(chan string, 5)}
 284
 285         arv, _ := arvadosclient.MakeArvadosClient()
 286         kc, _ := MakeKeepClient(&arv)
 287
 288         kc.Want_replicas = 2
 289         arv.ApiToken = "abc123"
 290         localRoots := make(map[string]string)
 291         writableLocalRoots := make(map[string]string)
 292
 293         ks := RunSomeFakeKeepServers(st, 5)
 294
 295         for i, k := range ks {
 296                 localRoots[fmt.Sprintf("zzzzz-bi6l4-fakefakefake%03d", i)] = k.url
 297                 writableLocalRoots[fmt.Sprintf("zzzzz-bi6l4-fakefakefake%03d", i)] = k.url
 298                 defer k.listener.Close()
 299         }
 300
 301         kc.SetServiceRoots(localRoots, writableLocalRoots, nil)
 302
 303         reader, writer := io.Pipe()
 304
 305         go func() {
 306                 writer.Write([]byte("foo"))
 307                 writer.Close()
 308         }()
 309
 310         kc.PutHR(hash, reader, 3)
 311
 312         shuff := NewRootSorter(kc.LocalRoots(), hash).GetSortedRoots()
 313         log.Print(shuff)
 314
 315         s1 := <-st.handled
 316         s2 := <-st.handled
 317
 318         c.Check((s1 == shuff[0] && s2 == shuff[1]) ||
 319                 (s1 == shuff[1] && s2 == shuff[0]),
 320                 Equals,
 321                 true)
 322
 323         log.Printf("TestPutHR done")
 324 }
 325
 326 func (s *StandaloneSuite) TestPutWithFail(c *C) {
 327         log.Printf("TestPutWithFail")
 328
 329         hash := fmt.Sprintf("%x", md5.Sum([]byte("foo")))
 330
 331         st := StubPutHandler{
 332                 c,
 333                 hash,
 334                 "abc123",
 335                 "foo",
 336                 make(chan string, 4)}
 337
 338         fh := FailHandler{
 339                 make(chan string, 1)}
 340
 341         arv, err := arvadosclient.MakeArvadosClient()
 342         kc, _ := MakeKeepClient(&arv)
 343
 344         kc.Want_replicas = 2
 345         arv.ApiToken = "abc123"
 346         localRoots := make(map[string]string)
 347         writableLocalRoots := make(map[string]string)
 348
 349         ks1 := RunSomeFakeKeepServers(st, 4)
 350         ks2 := RunSomeFakeKeepServers(fh, 1)
 351
 352         for i, k := range ks1 {
 353                 localRoots[fmt.Sprintf("zzzzz-bi6l4-fakefakefake%03d", i)] = k.url
 354                 writableLocalRoots[fmt.Sprintf("zzzzz-bi6l4-fakefakefake%03d", i)] = k.url
 355                 defer k.listener.Close()
 356         }
 357         for i, k := range ks2 {
 358                 localRoots[fmt.Sprintf("zzzzz-bi6l4-fakefakefake%03d", i+len(ks1))] = k.url
 359                 writableLocalRoots[fmt.Sprintf("zzzzz-bi6l4-fakefakefake%03d", i+len(ks1))] = k.url
 360                 defer k.listener.Close()
 361         }
 362
 363         kc.SetServiceRoots(localRoots, writableLocalRoots, nil)
 364
 365         shuff := NewRootSorter(
 366                 kc.LocalRoots(), Md5String("foo")).GetSortedRoots()
 367
 368         phash, replicas, err := kc.PutB([]byte("foo"))
 369
 370         <-fh.handled
 371
 372         c.Check(err, Equals, nil)
 373         c.Check(phash, Equals, "")
 374         c.Check(replicas, Equals, 2)
 375
 376         s1 := <-st.handled
 377         s2 := <-st.handled
 378
 379         c.Check((s1 == shuff[1] && s2 == shuff[2]) ||
 380                 (s1 == shuff[2] && s2 == shuff[1]),
 381                 Equals,
 382                 true)
 383 }
 384
 385 func (s *StandaloneSuite) TestPutWithTooManyFail(c *C) {
 386         log.Printf("TestPutWithTooManyFail")
 387
 388         hash := fmt.Sprintf("%x", md5.Sum([]byte("foo")))
 389
 390         st := StubPutHandler{
 391                 c,
 392                 hash,
 393                 "abc123",
 394                 "foo",
 395                 make(chan string, 1)}
 396
 397         fh := FailHandler{
 398                 make(chan string, 4)}
 399
 400         arv, err := arvadosclient.MakeArvadosClient()
 401         kc, _ := MakeKeepClient(&arv)
 402
 403         kc.Want_replicas = 2
 404         arv.ApiToken = "abc123"
 405         localRoots := make(map[string]string)
 406         writableLocalRoots := make(map[string]string)
 407
 408         ks1 := RunSomeFakeKeepServers(st, 1)
 409         ks2 := RunSomeFakeKeepServers(fh, 4)
 410
 411         for i, k := range ks1 {
 412                 localRoots[fmt.Sprintf("zzzzz-bi6l4-fakefakefake%03d", i)] = k.url
 413                 writableLocalRoots[fmt.Sprintf("zzzzz-bi6l4-fakefakefake%03d", i)] = k.url
 414                 defer k.listener.Close()
 415         }
 416         for i, k := range ks2 {
 417                 localRoots[fmt.Sprintf("zzzzz-bi6l4-fakefakefake%03d", i+len(ks1))] = k.url
 418                 writableLocalRoots[fmt.Sprintf("zzzzz-bi6l4-fakefakefake%03d", i+len(ks1))] = k.url
 419                 defer k.listener.Close()
 420         }
 421
 422         kc.SetServiceRoots(localRoots, writableLocalRoots, nil)
 423
 424         _, replicas, err := kc.PutB([]byte("foo"))
 425
 426         c.Check(err, Equals, InsufficientReplicasError)
 427         c.Check(replicas, Equals, 1)
 428         c.Check(<-st.handled, Equals, ks1[0].url)
 429
 430         log.Printf("TestPutWithTooManyFail done")
 431 }
 432
 433 type StubGetHandler struct {
 434         c              *C
 435         expectPath     string
 436         expectApiToken string
 437         httpStatus     int
 438         body           []byte
 439 }
 440
 441 func (sgh StubGetHandler) ServeHTTP(resp http.ResponseWriter, req *http.Request) {
 442         sgh.c.Check(req.URL.Path, Equals, "/"+sgh.expectPath)
 443         sgh.c.Check(req.Header.Get("Authorization"), Equals, fmt.Sprintf("OAuth2 %s", sgh.expectApiToken))
 444         resp.WriteHeader(sgh.httpStatus)
 445         resp.Header().Set("Content-Length", fmt.Sprintf("%d", len(sgh.body)))
 446         resp.Write(sgh.body)
 447 }
 448
 449 func (s *StandaloneSuite) TestGet(c *C) {
 450         log.Printf("TestGet")
 451
 452         hash := fmt.Sprintf("%x", md5.Sum([]byte("foo")))
 453
 454         st := StubGetHandler{
 455                 c,
 456                 hash,
 457                 "abc123",
 458                 http.StatusOK,
 459                 []byte("foo")}
 460
 461         ks := RunFakeKeepServer(st)
 462         defer ks.listener.Close()
 463
 464         arv, err := arvadosclient.MakeArvadosClient()
 465         kc, _ := MakeKeepClient(&arv)
 466         arv.ApiToken = "abc123"
 467         kc.SetServiceRoots(map[string]string{"x": ks.url}, map[string]string{ks.url: ""}, nil)
 468
 469         r, n, url2, err := kc.Get(hash)
 470         defer r.Close()
 471         c.Check(err, Equals, nil)
 472         c.Check(n, Equals, int64(3))
 473         c.Check(url2, Equals, fmt.Sprintf("%s/%s", ks.url, hash))
 474
 475         content, err2 := ioutil.ReadAll(r)
 476         c.Check(err2, Equals, nil)
 477         c.Check(content, DeepEquals, []byte("foo"))
 478
 479         log.Printf("TestGet done")
 480 }
 481
 482 func (s *StandaloneSuite) TestGetFail(c *C) {
 483         hash := fmt.Sprintf("%x", md5.Sum([]byte("foo")))
 484
 485         st := FailHandler{make(chan string, 1)}
 486
 487         ks := RunFakeKeepServer(st)
 488         defer ks.listener.Close()
 489
 490         arv, err := arvadosclient.MakeArvadosClient()
 491         kc, _ := MakeKeepClient(&arv)
 492         arv.ApiToken = "abc123"
 493         kc.SetServiceRoots(map[string]string{"x": ks.url}, map[string]string{ks.url: ""}, nil)
 494
 495         r, n, url2, err := kc.Get(hash)
 496         c.Check(err, Equals, BlockNotFound)
 497         c.Check(n, Equals, int64(0))
 498         c.Check(url2, Equals, "")
 499         c.Check(r, Equals, nil)
 500 }
 501
 502 func (s *StandaloneSuite) TestGetWithServiceHint(c *C) {
 503         uuid := "zzzzz-bi6l4-123451234512345"
 504         hash := fmt.Sprintf("%x", md5.Sum([]byte("foo")))
 505
 506         // This one shouldn't be used:
 507         ks0 := RunFakeKeepServer(StubGetHandler{
 508                 c,
 509                 "error if used",
 510                 "abc123",
 511                 http.StatusOK,
 512                 []byte("foo")})
 513         defer ks0.listener.Close()
 514         // This one should be used:
 515         ks := RunFakeKeepServer(StubGetHandler{
 516                 c,
 517                 hash + "+K@" + uuid,
 518                 "abc123",
 519                 http.StatusOK,
 520                 []byte("foo")})
 521         defer ks.listener.Close()
 522
 523         arv, err := arvadosclient.MakeArvadosClient()
 524         kc, _ := MakeKeepClient(&arv)
 525         arv.ApiToken = "abc123"
 526         kc.SetServiceRoots(
 527                 map[string]string{"x": ks0.url},
 528                 map[string]string{"x": ks0.url},
 529                 map[string]string{uuid: ks.url})
 530
 531         r, n, uri, err := kc.Get(hash + "+K@" + uuid)
 532         defer r.Close()
 533         c.Check(err, Equals, nil)
 534         c.Check(n, Equals, int64(3))
 535         c.Check(uri, Equals, fmt.Sprintf("%s/%s", ks.url, hash+"+K@"+uuid))
 536
 537         content, err := ioutil.ReadAll(r)
 538         c.Check(err, Equals, nil)
 539         c.Check(content, DeepEquals, []byte("foo"))
 540 }
 541
 542 // Use a service hint to fetch from a local disk service, overriding
 543 // rendezvous probe order.
 544 func (s *StandaloneSuite) TestGetWithLocalServiceHint(c *C) {
 545         uuid := "zzzzz-bi6l4-zzzzzzzzzzzzzzz"
 546         hash := fmt.Sprintf("%x", md5.Sum([]byte("foo")))
 547
 548         // This one shouldn't be used, although it appears first in
 549         // rendezvous probe order:
 550         ks0 := RunFakeKeepServer(StubGetHandler{
 551                 c,
 552                 "error if used",
 553                 "abc123",
 554                 http.StatusOK,
 555                 []byte("foo")})
 556         defer ks0.listener.Close()
 557         // This one should be used:
 558         ks := RunFakeKeepServer(StubGetHandler{
 559                 c,
 560                 hash + "+K@" + uuid,
 561                 "abc123",
 562                 http.StatusOK,
 563                 []byte("foo")})
 564         defer ks.listener.Close()
 565
 566         arv, err := arvadosclient.MakeArvadosClient()
 567         kc, _ := MakeKeepClient(&arv)
 568         arv.ApiToken = "abc123"
 569         kc.SetServiceRoots(
 570                 map[string]string{
 571                         "zzzzz-bi6l4-yyyyyyyyyyyyyyy": ks0.url,
 572                         "zzzzz-bi6l4-xxxxxxxxxxxxxxx": ks0.url,
 573                         "zzzzz-bi6l4-wwwwwwwwwwwwwww": ks0.url,
 574                         uuid: ks.url},
 575                 map[string]string{
 576                         "zzzzz-bi6l4-yyyyyyyyyyyyyyy": ks0.url,
 577                         "zzzzz-bi6l4-xxxxxxxxxxxxxxx": ks0.url,
 578                         "zzzzz-bi6l4-wwwwwwwwwwwwwww": ks0.url,
 579                         uuid: ks.url},
 580                 map[string]string{
 581                         "zzzzz-bi6l4-yyyyyyyyyyyyyyy": ks0.url,
 582                         "zzzzz-bi6l4-xxxxxxxxxxxxxxx": ks0.url,
 583                         "zzzzz-bi6l4-wwwwwwwwwwwwwww": ks0.url,
 584                         uuid: ks.url},
 585         )
 586
 587         r, n, uri, err := kc.Get(hash + "+K@" + uuid)
 588         defer r.Close()
 589         c.Check(err, Equals, nil)
 590         c.Check(n, Equals, int64(3))
 591         c.Check(uri, Equals, fmt.Sprintf("%s/%s", ks.url, hash+"+K@"+uuid))
 592
 593         content, err := ioutil.ReadAll(r)
 594         c.Check(err, Equals, nil)
 595         c.Check(content, DeepEquals, []byte("foo"))
 596 }
 597
 598 func (s *StandaloneSuite) TestGetWithServiceHintFailoverToLocals(c *C) {
 599         uuid := "zzzzz-bi6l4-123451234512345"
 600         hash := fmt.Sprintf("%x", md5.Sum([]byte("foo")))
 601
 602         ksLocal := RunFakeKeepServer(StubGetHandler{
 603                 c,
 604                 hash + "+K@" + uuid,
 605                 "abc123",
 606                 http.StatusOK,
 607                 []byte("foo")})
 608         defer ksLocal.listener.Close()
 609         ksGateway := RunFakeKeepServer(StubGetHandler{
 610                 c,
 611                 hash + "+K@" + uuid,
 612                 "abc123",
 613                 http.StatusInternalServerError,
 614                 []byte("Error")})
 615         defer ksGateway.listener.Close()
 616
 617         arv, err := arvadosclient.MakeArvadosClient()
 618         kc, _ := MakeKeepClient(&arv)
 619         arv.ApiToken = "abc123"
 620         kc.SetServiceRoots(
 621                 map[string]string{"zzzzz-bi6l4-keepdisk0000000": ksLocal.url},
 622                 map[string]string{"zzzzz-bi6l4-keepdisk0000000": ksLocal.url},
 623                 map[string]string{uuid: ksGateway.url})
 624
 625         r, n, uri, err := kc.Get(hash + "+K@" + uuid)
 626         c.Assert(err, Equals, nil)
 627         defer r.Close()
 628         c.Check(n, Equals, int64(3))
 629         c.Check(uri, Equals, fmt.Sprintf("%s/%s", ksLocal.url, hash+"+K@"+uuid))
 630
 631         content, err := ioutil.ReadAll(r)
 632         c.Check(err, Equals, nil)
 633         c.Check(content, DeepEquals, []byte("foo"))
 634 }
 635
 636 type BarHandler struct {
 637         handled chan string
 638 }
 639
 640 func (this BarHandler) ServeHTTP(resp http.ResponseWriter, req *http.Request) {
 641         resp.Write([]byte("bar"))
 642         this.handled <- fmt.Sprintf("http://%s", req.Host)
 643 }
 644
 645 func (s *StandaloneSuite) TestChecksum(c *C) {
 646         foohash := fmt.Sprintf("%x", md5.Sum([]byte("foo")))
 647         barhash := fmt.Sprintf("%x", md5.Sum([]byte("bar")))
 648
 649         st := BarHandler{make(chan string, 1)}
 650
 651         ks := RunFakeKeepServer(st)
 652         defer ks.listener.Close()
 653
 654         arv, err := arvadosclient.MakeArvadosClient()
 655         kc, _ := MakeKeepClient(&arv)
 656         arv.ApiToken = "abc123"
 657         kc.SetServiceRoots(map[string]string{"x": ks.url}, map[string]string{ks.url: ""}, nil)
 658
 659         r, n, _, err := kc.Get(barhash)
 660         _, err = ioutil.ReadAll(r)
 661         c.Check(n, Equals, int64(3))
 662         c.Check(err, Equals, nil)
 663
 664         <-st.handled
 665
 666         r, n, _, err = kc.Get(foohash)
 667         _, err = ioutil.ReadAll(r)
 668         c.Check(n, Equals, int64(3))
 669         c.Check(err, Equals, BadChecksum)
 670
 671         <-st.handled
 672 }
 673
 674 func (s *StandaloneSuite) TestGetWithFailures(c *C) {
 675         content := []byte("waz")
 676         hash := fmt.Sprintf("%x", md5.Sum(content))
 677
 678         fh := FailHandler{
 679                 make(chan string, 4)}
 680
 681         st := StubGetHandler{
 682                 c,
 683                 hash,
 684                 "abc123",
 685                 http.StatusOK,
 686                 content}
 687
 688         arv, err := arvadosclient.MakeArvadosClient()
 689         kc, _ := MakeKeepClient(&arv)
 690         arv.ApiToken = "abc123"
 691         localRoots := make(map[string]string)
 692         writableLocalRoots := make(map[string]string)
 693
 694         ks1 := RunSomeFakeKeepServers(st, 1)
 695         ks2 := RunSomeFakeKeepServers(fh, 4)
 696
 697         for i, k := range ks1 {
 698                 localRoots[fmt.Sprintf("zzzzz-bi6l4-fakefakefake%03d", i)] = k.url
 699                 writableLocalRoots[fmt.Sprintf("zzzzz-bi6l4-fakefakefake%03d", i)] = k.url
 700                 defer k.listener.Close()
 701         }
 702         for i, k := range ks2 {
 703                 localRoots[fmt.Sprintf("zzzzz-bi6l4-fakefakefake%03d", i+len(ks1))] = k.url
 704                 writableLocalRoots[fmt.Sprintf("zzzzz-bi6l4-fakefakefake%03d", i+len(ks1))] = k.url
 705                 defer k.listener.Close()
 706         }
 707
 708         kc.SetServiceRoots(localRoots, writableLocalRoots, nil)
 709
 710         // This test works only if one of the failing services is
 711         // attempted before the succeeding service. Otherwise,
 712         // <-fh.handled below will just hang! (Probe order depends on
 713         // the choice of block content "waz" and the UUIDs of the fake
 714         // servers, so we just tried different strings until we found
 715         // an example that passes this Assert.)
 716         c.Assert(NewRootSorter(localRoots, hash).GetSortedRoots()[0], Not(Equals), ks1[0].url)
 717
 718         r, n, url2, err := kc.Get(hash)
 719
 720         <-fh.handled
 721         c.Check(err, Equals, nil)
 722         c.Check(n, Equals, int64(3))
 723         c.Check(url2, Equals, fmt.Sprintf("%s/%s", ks1[0].url, hash))
 724
 725         read_content, err2 := ioutil.ReadAll(r)
 726         c.Check(err2, Equals, nil)
 727         c.Check(read_content, DeepEquals, content)
 728 }
 729
 730 func (s *ServerRequiredSuite) TestPutGetHead(c *C) {
 731         content := []byte("TestPutGetHead")
 732
 733         arv, err := arvadosclient.MakeArvadosClient()
 734         kc, err := MakeKeepClient(&arv)
 735         c.Assert(err, Equals, nil)
 736
 737         hash := fmt.Sprintf("%x", md5.Sum(content))
 738
 739         {
 740                 n, _, err := kc.Ask(hash)
 741                 c.Check(err, Equals, BlockNotFound)
 742                 c.Check(n, Equals, int64(0))
 743         }
 744         {
 745                 hash2, replicas, err := kc.PutB(content)
 746                 c.Check(hash2, Matches, fmt.Sprintf(`%s\+%d\b.*`, hash, len(content)))
 747                 c.Check(replicas, Equals, 2)
 748                 c.Check(err, Equals, nil)
 749         }
 750         {
 751                 r, n, url2, err := kc.Get(hash)
 752                 c.Check(err, Equals, nil)
 753                 c.Check(n, Equals, int64(len(content)))
 754                 c.Check(url2, Matches, fmt.Sprintf("http://localhost:\\d+/%s", hash))
 755
 756                 read_content, err2 := ioutil.ReadAll(r)
 757                 c.Check(err2, Equals, nil)
 758                 c.Check(read_content, DeepEquals, content)
 759         }
 760         {
 761                 n, url2, err := kc.Ask(hash)
 762                 c.Check(err, Equals, nil)
 763                 c.Check(n, Equals, int64(len(content)))
 764                 c.Check(url2, Matches, fmt.Sprintf("http://localhost:\\d+/%s", hash))
 765         }
 766 }
 767
 768 type StubProxyHandler struct {
 769         handled chan string
 770 }
 771
 772 func (this StubProxyHandler) ServeHTTP(resp http.ResponseWriter, req *http.Request) {
 773         resp.Header().Set("X-Keep-Replicas-Stored", "2")
 774         this.handled <- fmt.Sprintf("http://%s", req.Host)
 775 }
 776
 777 func (s *StandaloneSuite) TestPutProxy(c *C) {
 778         log.Printf("TestPutProxy")
 779
 780         st := StubProxyHandler{make(chan string, 1)}
 781
 782         arv, err := arvadosclient.MakeArvadosClient()
 783         kc, _ := MakeKeepClient(&arv)
 784
 785         kc.Want_replicas = 2
 786         kc.Using_proxy = true
 787         arv.ApiToken = "abc123"
 788         localRoots := make(map[string]string)
 789         writableLocalRoots := make(map[string]string)
 790
 791         ks1 := RunSomeFakeKeepServers(st, 1)
 792
 793         for i, k := range ks1 {
 794                 localRoots[fmt.Sprintf("zzzzz-bi6l4-fakefakefake%03d", i)] = k.url
 795                 writableLocalRoots[fmt.Sprintf("zzzzz-bi6l4-fakefakefake%03d", i)] = k.url
 796                 defer k.listener.Close()
 797         }
 798
 799         kc.SetServiceRoots(localRoots, writableLocalRoots, nil)
 800
 801         _, replicas, err := kc.PutB([]byte("foo"))
 802         <-st.handled
 803
 804         c.Check(err, Equals, nil)
 805         c.Check(replicas, Equals, 2)
 806
 807         log.Printf("TestPutProxy done")
 808 }
 809
 810 func (s *StandaloneSuite) TestPutProxyInsufficientReplicas(c *C) {
 811         log.Printf("TestPutProxy")
 812
 813         st := StubProxyHandler{make(chan string, 1)}
 814
 815         arv, err := arvadosclient.MakeArvadosClient()
 816         kc, _ := MakeKeepClient(&arv)
 817
 818         kc.Want_replicas = 3
 819         kc.Using_proxy = true
 820         arv.ApiToken = "abc123"
 821         localRoots := make(map[string]string)
 822         writableLocalRoots := make(map[string]string)
 823
 824         ks1 := RunSomeFakeKeepServers(st, 1)
 825
 826         for i, k := range ks1 {
 827                 localRoots[fmt.Sprintf("zzzzz-bi6l4-fakefakefake%03d", i)] = k.url
 828                 writableLocalRoots[fmt.Sprintf("zzzzz-bi6l4-fakefakefake%03d", i)] = k.url
 829                 defer k.listener.Close()
 830         }
 831         kc.SetServiceRoots(localRoots, writableLocalRoots, nil)
 832
 833         _, replicas, err := kc.PutB([]byte("foo"))
 834         <-st.handled
 835
 836         c.Check(err, Equals, InsufficientReplicasError)
 837         c.Check(replicas, Equals, 2)
 838
 839         log.Printf("TestPutProxy done")
 840 }
 841
 842 func (s *StandaloneSuite) TestMakeLocator(c *C) {
 843         l, err := MakeLocator("91f372a266fe2bf2823cb8ec7fda31ce+3+Aabcde@12345678")
 844         c.Check(err, Equals, nil)
 845         c.Check(l.Hash, Equals, "91f372a266fe2bf2823cb8ec7fda31ce")
 846         c.Check(l.Size, Equals, 3)
 847         c.Check(l.Hints, DeepEquals, []string{"3", "Aabcde@12345678"})
 848 }
 849
 850 func (s *StandaloneSuite) TestMakeLocatorNoHints(c *C) {
 851         l, err := MakeLocator("91f372a266fe2bf2823cb8ec7fda31ce")
 852         c.Check(err, Equals, nil)
 853         c.Check(l.Hash, Equals, "91f372a266fe2bf2823cb8ec7fda31ce")
 854         c.Check(l.Size, Equals, -1)
 855         c.Check(l.Hints, DeepEquals, []string{})
 856 }
 857
 858 func (s *StandaloneSuite) TestMakeLocatorNoSizeHint(c *C) {
 859         l, err := MakeLocator("91f372a266fe2bf2823cb8ec7fda31ce+Aabcde@12345678")
 860         c.Check(err, Equals, nil)
 861         c.Check(l.Hash, Equals, "91f372a266fe2bf2823cb8ec7fda31ce")
 862         c.Check(l.Size, Equals, -1)
 863         c.Check(l.Hints, DeepEquals, []string{"Aabcde@12345678"})
 864 }
 865
 866 func (s *StandaloneSuite) TestMakeLocatorPreservesUnrecognizedHints(c *C) {
 867         str := "91f372a266fe2bf2823cb8ec7fda31ce+3+Unknown+Kzzzzz+Afoobar"
 868         l, err := MakeLocator(str)
 869         c.Check(err, Equals, nil)
 870         c.Check(l.Hash, Equals, "91f372a266fe2bf2823cb8ec7fda31ce")
 871         c.Check(l.Size, Equals, 3)
 872         c.Check(l.Hints, DeepEquals, []string{"3", "Unknown", "Kzzzzz", "Afoobar"})
 873         c.Check(l.String(), Equals, str)
 874 }
 875
 876 func (s *StandaloneSuite) TestMakeLocatorInvalidInput(c *C) {
 877         _, err := MakeLocator("91f372a266fe2bf2823cb8ec7fda31c")
 878         c.Check(err, Equals, InvalidLocatorError)
 879 }
 880
 881 func (s *StandaloneSuite) TestPutBWant2ReplicasWithOnlyOneWritableLocalRoot(c *C) {
 882         hash := Md5String("foo")
 883
 884         st := StubPutHandler{
 885                 c,
 886                 hash,
 887                 "abc123",
 888                 "foo",
 889                 make(chan string, 5)}
 890
 891         arv, _ := arvadosclient.MakeArvadosClient()
 892         kc, _ := MakeKeepClient(&arv)
 893
 894         kc.Want_replicas = 2
 895         arv.ApiToken = "abc123"
 896         localRoots := make(map[string]string)
 897         writableLocalRoots := make(map[string]string)
 898
 899         ks := RunSomeFakeKeepServers(st, 5)
 900
 901         for i, k := range ks {
 902                 localRoots[fmt.Sprintf("zzzzz-bi6l4-fakefakefake%03d", i)] = k.url
 903                 if i == 0 {
 904                         writableLocalRoots[fmt.Sprintf("zzzzz-bi6l4-fakefakefake%03d", i)] = k.url
 905                 }
 906                 defer k.listener.Close()
 907         }
 908
 909         kc.SetServiceRoots(localRoots, writableLocalRoots, nil)
 910
 911         _, replicas, err := kc.PutB([]byte("foo"))
 912
 913         c.Check(err, Equals, InsufficientReplicasError)
 914         c.Check(replicas, Equals, 1)
 915
 916         c.Check(<-st.handled, Equals, localRoots[fmt.Sprintf("zzzzz-bi6l4-fakefakefake%03d", 0)])
 917 }
 918
 919 func (s *StandaloneSuite) TestPutBWithNoWritableLocalRoots(c *C) {
 920         hash := Md5String("foo")
 921
 922         st := StubPutHandler{
 923                 c,
 924                 hash,
 925                 "abc123",
 926                 "foo",
 927                 make(chan string, 5)}
 928
 929         arv, _ := arvadosclient.MakeArvadosClient()
 930         kc, _ := MakeKeepClient(&arv)
 931
 932         kc.Want_replicas = 2
 933         arv.ApiToken = "abc123"
 934         localRoots := make(map[string]string)
 935         writableLocalRoots := make(map[string]string)
 936
 937         ks := RunSomeFakeKeepServers(st, 5)
 938
 939         for i, k := range ks {
 940                 localRoots[fmt.Sprintf("zzzzz-bi6l4-fakefakefake%03d", i)] = k.url
 941                 defer k.listener.Close()
 942         }
 943
 944         kc.SetServiceRoots(localRoots, writableLocalRoots, nil)
 945
 946         _, replicas, err := kc.PutB([]byte("foo"))
 947
 948         c.Check(err, Equals, InsufficientReplicasError)
 949         c.Check(replicas, Equals, 0)
 950 }
 951
 952 type StubGetIndexHandler struct {
 953         c              *C
 954         expectPath     string
 955         expectAPIToken string
 956         httpStatus     int
 957         body           []byte
 958 }
 959
 960 func (h StubGetIndexHandler) ServeHTTP(resp http.ResponseWriter, req *http.Request) {
 961         h.c.Check(req.URL.Path, Equals, h.expectPath)
 962         h.c.Check(req.Header.Get("Authorization"), Equals, fmt.Sprintf("OAuth2 %s", h.expectAPIToken))
 963         resp.WriteHeader(h.httpStatus)
 964         resp.Header().Set("Content-Length", fmt.Sprintf("%d", len(h.body)))
 965         resp.Write(h.body)
 966 }
 967
 968 func (s *StandaloneSuite) TestGetIndexWithNoPrefix(c *C) {
 969         hash := fmt.Sprintf("%x", md5.Sum([]byte("foo")))
 970
 971         st := StubGetIndexHandler{
 972                 c,
 973                 "/index",
 974                 "abc123",
 975                 http.StatusOK,
 976                 []byte(hash + "+3 1443559274\n\n")}
 977
 978         ks := RunFakeKeepServer(st)
 979         defer ks.listener.Close()
 980
 981         arv, err := arvadosclient.MakeArvadosClient()
 982         kc, _ := MakeKeepClient(&arv)
 983         arv.ApiToken = "abc123"
 984         kc.SetServiceRoots(map[string]string{"x": ks.url}, map[string]string{ks.url: ""}, nil)
 985
 986         r, err := kc.GetIndex("x", "")
 987         c.Check(err, Equals, nil)
 988
 989         content, err2 := ioutil.ReadAll(r)
 990         c.Check(err2, Equals, nil)
 991         c.Check(content, DeepEquals, st.body[0:len(st.body)-1])
 992 }
 993
 994 func (s *StandaloneSuite) TestGetIndexWithPrefix(c *C) {
 995         hash := fmt.Sprintf("%x", md5.Sum([]byte("foo")))
 996
 997         st := StubGetIndexHandler{
 998                 c,
 999                 "/index/" + hash[0:3],
1000                 "abc123",
1001                 http.StatusOK,
1002                 []byte(hash + "+3 1443559274\n\n")}
1003
1004         ks := RunFakeKeepServer(st)
1005         defer ks.listener.Close()
1006
1007         arv, err := arvadosclient.MakeArvadosClient()
1008         kc, _ := MakeKeepClient(&arv)
1009         arv.ApiToken = "abc123"
1010         kc.SetServiceRoots(map[string]string{"x": ks.url}, map[string]string{ks.url: ""}, nil)
1011
1012         r, err := kc.GetIndex("x", hash[0:3])
1013         c.Check(err, Equals, nil)
1014
1015         content, err2 := ioutil.ReadAll(r)
1016         c.Check(err2, Equals, nil)
1017         c.Check(content, DeepEquals, st.body[0:len(st.body)-1])
1018 }
1019
1020 func (s *StandaloneSuite) TestGetIndexIncomplete(c *C) {
1021         hash := fmt.Sprintf("%x", md5.Sum([]byte("foo")))
1022
1023         st := StubGetIndexHandler{
1024                 c,
1025                 "/index/" + hash[0:3],
1026                 "abc123",
1027                 http.StatusOK,
1028                 []byte(hash)}
1029
1030         ks := RunFakeKeepServer(st)
1031         defer ks.listener.Close()
1032
1033         arv, err := arvadosclient.MakeArvadosClient()
1034         kc, _ := MakeKeepClient(&arv)
1035         arv.ApiToken = "abc123"
1036         kc.SetServiceRoots(map[string]string{"x": ks.url}, map[string]string{ks.url: ""}, nil)
1037
1038         _, err = kc.GetIndex("x", hash[0:3])
1039         c.Check(err, Equals, ErrIncompleteIndex)
1040 }
1041
1042 func (s *StandaloneSuite) TestGetIndexWithNoSuchServer(c *C) {
1043         hash := fmt.Sprintf("%x", md5.Sum([]byte("foo")))
1044
1045         st := StubGetIndexHandler{
1046                 c,
1047                 "/index/" + hash[0:3],
1048                 "abc123",
1049                 http.StatusOK,
1050                 []byte(hash)}
1051
1052         ks := RunFakeKeepServer(st)
1053         defer ks.listener.Close()
1054
1055         arv, err := arvadosclient.MakeArvadosClient()
1056         kc, _ := MakeKeepClient(&arv)
1057         arv.ApiToken = "abc123"
1058         kc.SetServiceRoots(map[string]string{"x": ks.url}, map[string]string{ks.url: ""}, nil)
1059
1060         _, err = kc.GetIndex("y", hash[0:3])
1061         c.Check(err, Equals, ErrNoSuchKeepServer)
1062 }
1063
1064 func (s *StandaloneSuite) TestGetIndexWithNoSuchPrefix(c *C) {
1065         st := StubGetIndexHandler{
1066                 c,
1067                 "/index/abcd",
1068                 "abc123",
1069                 http.StatusOK,
1070                 []byte("\n")}
1071
1072         ks := RunFakeKeepServer(st)
1073         defer ks.listener.Close()
1074
1075         arv, err := arvadosclient.MakeArvadosClient()
1076         kc, _ := MakeKeepClient(&arv)
1077         arv.ApiToken = "abc123"
1078         kc.SetServiceRoots(map[string]string{"x": ks.url}, map[string]string{ks.url: ""}, nil)
1079
1080         r, err := kc.GetIndex("x", "abcd")
1081         c.Check(err, Equals, nil)
1082
1083         content, err2 := ioutil.ReadAll(r)
1084         c.Check(err2, Equals, nil)
1085         c.Check(content, DeepEquals, st.body[0:len(st.body)-1])
1086 }