sdk/go/src/arvados.org/keepclient/keepclient_test.go

   1 package keepclient
   2
   3 import (
   4         "arvados.org/streamer"
   5         "crypto/md5"
   6         "flag"
   7         "fmt"
   8         . "gopkg.in/check.v1"
   9         "io"
  10         "io/ioutil"
  11         "log"
  12         "net"
  13         "net/http"
  14         "os"
  15         "os/exec"
  16         "strings"
  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         gopath := os.Getenv("GOPATH")
  39         return fmt.Sprintf("%s/../python", strings.Split(gopath, ":")[0])
  40 }
  41
  42 func (s *ServerRequiredSuite) SetUpSuite(c *C) {
  43         if *no_server {
  44                 c.Skip("Skipping tests that require server")
  45         } else {
  46                 os.Chdir(pythonDir())
  47                 if err := exec.Command("python", "run_test_server.py", "start").Run(); err != nil {
  48                         panic("'python run_test_server.py start' returned error")
  49                 }
  50                 if err := exec.Command("python", "run_test_server.py", "start_keep").Run(); err != nil {
  51                         panic("'python run_test_server.py start_keep' returned error")
  52                 }
  53         }
  54 }
  55
  56 func (s *ServerRequiredSuite) TearDownSuite(c *C) {
  57         os.Chdir(pythonDir())
  58         exec.Command("python", "run_test_server.py", "stop_keep").Run()
  59         exec.Command("python", "run_test_server.py", "stop").Run()
  60 }
  61
  62 func (s *ServerRequiredSuite) TestMakeKeepClient(c *C) {
  63         os.Setenv("ARVADOS_API_HOST", "localhost:3001")
  64         os.Setenv("ARVADOS_API_TOKEN", "4axaw8zxe0qm22wa6urpp5nskcne8z88cvbupv653y1njyi05h")
  65         os.Setenv("ARVADOS_API_HOST_INSECURE", "")
  66
  67         kc, err := MakeKeepClient()
  68         c.Check(kc.ApiServer, Equals, "localhost:3001")
  69         c.Check(kc.ApiToken, Equals, "4axaw8zxe0qm22wa6urpp5nskcne8z88cvbupv653y1njyi05h")
  70         c.Check(kc.ApiInsecure, Equals, false)
  71
  72         os.Setenv("ARVADOS_API_HOST_INSECURE", "true")
  73
  74         kc, err = MakeKeepClient()
  75         c.Check(kc.ApiServer, Equals, "localhost:3001")
  76         c.Check(kc.ApiToken, Equals, "4axaw8zxe0qm22wa6urpp5nskcne8z88cvbupv653y1njyi05h")
  77         c.Check(kc.ApiInsecure, Equals, true)
  78         c.Check(kc.Client.Transport.(*http.Transport).TLSClientConfig.InsecureSkipVerify, Equals, true)
  79
  80         c.Assert(err, Equals, nil)
  81         c.Check(len(kc.ServiceRoots()), Equals, 2)
  82         c.Check(kc.ServiceRoots()[0], Equals, "http://localhost:25107")
  83         c.Check(kc.ServiceRoots()[1], Equals, "http://localhost:25108")
  84 }
  85
  86 func (s *StandaloneSuite) TestShuffleServiceRoots(c *C) {
  87         kc := KeepClient{}
  88         kc.SetServiceRoots([]string{"http://localhost:25107", "http://localhost:25108", "http://localhost:25109", "http://localhost:25110", "http://localhost:25111", "http://localhost:25112", "http://localhost:25113", "http://localhost:25114", "http://localhost:25115", "http://localhost:25116", "http://localhost:25117", "http://localhost:25118", "http://localhost:25119", "http://localhost:25120", "http://localhost:25121", "http://localhost:25122", "http://localhost:25123"})
  89
  90         // "foo" acbd18db4cc2f85cedef654fccc4a4d8
  91         foo_shuffle := []string{"http://localhost:25116", "http://localhost:25120", "http://localhost:25119", "http://localhost:25122", "http://localhost:25108", "http://localhost:25114", "http://localhost:25112", "http://localhost:25107", "http://localhost:25118", "http://localhost:25111", "http://localhost:25113", "http://localhost:25121", "http://localhost:25110", "http://localhost:25117", "http://localhost:25109", "http://localhost:25115", "http://localhost:25123"}
  92         c.Check(kc.shuffledServiceRoots("acbd18db4cc2f85cedef654fccc4a4d8"), DeepEquals, foo_shuffle)
  93
  94         // "bar" 37b51d194a7513e45b56f6524f2d51f2
  95         bar_shuffle := []string{"http://localhost:25108", "http://localhost:25112", "http://localhost:25119", "http://localhost:25107", "http://localhost:25110", "http://localhost:25116", "http://localhost:25122", "http://localhost:25120", "http://localhost:25121", "http://localhost:25117", "http://localhost:25111", "http://localhost:25123", "http://localhost:25118", "http://localhost:25113", "http://localhost:25114", "http://localhost:25115", "http://localhost:25109"}
  96         c.Check(kc.shuffledServiceRoots("37b51d194a7513e45b56f6524f2d51f2"), DeepEquals, bar_shuffle)
  97 }
  98
  99 type StubPutHandler struct {
 100         c              *C
 101         expectPath     string
 102         expectApiToken string
 103         expectBody     string
 104         handled        chan string
 105 }
 106
 107 func (this StubPutHandler) ServeHTTP(resp http.ResponseWriter, req *http.Request) {
 108         this.c.Check(req.URL.Path, Equals, "/"+this.expectPath)
 109         this.c.Check(req.Header.Get("Authorization"), Equals, fmt.Sprintf("OAuth2 %s", this.expectApiToken))
 110         body, err := ioutil.ReadAll(req.Body)
 111         this.c.Check(err, Equals, nil)
 112         this.c.Check(body, DeepEquals, []byte(this.expectBody))
 113         resp.WriteHeader(200)
 114         this.handled <- fmt.Sprintf("http://%s", req.Host)
 115 }
 116
 117 func RunBogusKeepServer(st http.Handler, port int) (listener net.Listener, url string) {
 118         var err error
 119         listener, err = net.ListenTCP("tcp", &net.TCPAddr{Port: port})
 120         if err != nil {
 121                 panic(fmt.Sprintf("Could not listen on tcp port %v", port))
 122         }
 123
 124         url = fmt.Sprintf("http://localhost:%d", port)
 125
 126         go http.Serve(listener, st)
 127         return listener, url
 128 }
 129
 130 func UploadToStubHelper(c *C, st http.Handler, f func(KeepClient, string,
 131         io.ReadCloser, io.WriteCloser, chan uploadStatus)) {
 132
 133         listener, url := RunBogusKeepServer(st, 2990)
 134         defer listener.Close()
 135
 136         kc, _ := MakeKeepClient()
 137         kc.ApiToken = "abc123"
 138
 139         reader, writer := io.Pipe()
 140         upload_status := make(chan uploadStatus)
 141
 142         f(kc, url, reader, writer, upload_status)
 143 }
 144
 145 func (s *StandaloneSuite) TestUploadToStubKeepServer(c *C) {
 146         log.Printf("TestUploadToStubKeepServer")
 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                         go kc.uploadToKeepServer(url, st.expectPath, reader, upload_status, int64(len("foo")))
 160
 161                         writer.Write([]byte("foo"))
 162                         writer.Close()
 163
 164                         <-st.handled
 165                         status := <-upload_status
 166                         c.Check(status, DeepEquals, uploadStatus{nil, fmt.Sprintf("%s/%s", url, st.expectPath), 200, 1})
 167                 })
 168
 169         log.Printf("TestUploadToStubKeepServer done")
 170 }
 171
 172 func (s *StandaloneSuite) TestUploadToStubKeepServerBufferReader(c *C) {
 173         log.Printf("TestUploadToStubKeepServerBufferReader")
 174
 175         st := StubPutHandler{
 176                 c,
 177                 "acbd18db4cc2f85cedef654fccc4a4d8",
 178                 "abc123",
 179                 "foo",
 180                 make(chan string)}
 181
 182         UploadToStubHelper(c, st,
 183                 func(kc KeepClient, url string, reader io.ReadCloser,
 184                         writer io.WriteCloser, upload_status chan uploadStatus) {
 185
 186                         tr := streamer.AsyncStreamFromReader(512, reader)
 187                         defer tr.Close()
 188
 189                         br1 := tr.MakeStreamReader()
 190
 191                         go kc.uploadToKeepServer(url, st.expectPath, br1, upload_status, 3)
 192
 193                         writer.Write([]byte("foo"))
 194                         writer.Close()
 195
 196                         <-st.handled
 197
 198                         status := <-upload_status
 199                         c.Check(status, DeepEquals, uploadStatus{nil, fmt.Sprintf("%s/%s", url, st.expectPath), 200, 1})
 200                 })
 201
 202         log.Printf("TestUploadToStubKeepServerBufferReader done")
 203 }
 204
 205 type FailHandler struct {
 206         handled chan string
 207 }
 208
 209 func (this FailHandler) ServeHTTP(resp http.ResponseWriter, req *http.Request) {
 210         resp.WriteHeader(500)
 211         this.handled <- fmt.Sprintf("http://%s", req.Host)
 212 }
 213
 214 func (s *StandaloneSuite) TestFailedUploadToStubKeepServer(c *C) {
 215         log.Printf("TestFailedUploadToStubKeepServer")
 216
 217         st := FailHandler{
 218                 make(chan string)}
 219
 220         hash := "acbd18db4cc2f85cedef654fccc4a4d8"
 221
 222         UploadToStubHelper(c, st,
 223                 func(kc KeepClient, url string, reader io.ReadCloser,
 224                         writer io.WriteCloser, upload_status chan uploadStatus) {
 225
 226                         go kc.uploadToKeepServer(url, hash, reader, upload_status, 3)
 227
 228                         writer.Write([]byte("foo"))
 229                         writer.Close()
 230
 231                         <-st.handled
 232
 233                         status := <-upload_status
 234                         c.Check(status.url, Equals, fmt.Sprintf("%s/%s", url, hash))
 235                         c.Check(status.statusCode, Equals, 500)
 236                 })
 237         log.Printf("TestFailedUploadToStubKeepServer done")
 238 }
 239
 240 type KeepServer struct {
 241         listener net.Listener
 242         url      string
 243 }
 244
 245 func RunSomeFakeKeepServers(st http.Handler, n int, port int) (ks []KeepServer) {
 246         ks = make([]KeepServer, n)
 247
 248         for i := 0; i < n; i += 1 {
 249                 boguslistener, bogusurl := RunBogusKeepServer(st, port+i)
 250                 ks[i] = KeepServer{boguslistener, bogusurl}
 251         }
 252
 253         return ks
 254 }
 255
 256 func (s *StandaloneSuite) TestPutB(c *C) {
 257         log.Printf("TestPutB")
 258
 259         hash := fmt.Sprintf("%x", md5.Sum([]byte("foo")))
 260
 261         st := StubPutHandler{
 262                 c,
 263                 hash,
 264                 "abc123",
 265                 "foo",
 266                 make(chan string, 2)}
 267
 268         kc, _ := MakeKeepClient()
 269
 270         kc.Want_replicas = 2
 271         kc.ApiToken = "abc123"
 272         service_roots := make([]string, 5)
 273
 274         ks := RunSomeFakeKeepServers(st, 5, 2990)
 275
 276         for i := 0; i < len(ks); i += 1 {
 277                 service_roots[i] = ks[i].url
 278                 defer ks[i].listener.Close()
 279         }
 280
 281         kc.SetServiceRoots(service_roots)
 282
 283         kc.PutB([]byte("foo"))
 284
 285         shuff := kc.shuffledServiceRoots(fmt.Sprintf("%x", md5.Sum([]byte("foo"))))
 286
 287         s1 := <-st.handled
 288         s2 := <-st.handled
 289         c.Check((s1 == shuff[0] && s2 == shuff[1]) ||
 290                 (s1 == shuff[1] && s2 == shuff[0]),
 291                 Equals,
 292                 true)
 293
 294         log.Printf("TestPutB done")
 295 }
 296
 297 func (s *StandaloneSuite) TestPutHR(c *C) {
 298         log.Printf("TestPutHR")
 299
 300         hash := fmt.Sprintf("%x", md5.Sum([]byte("foo")))
 301
 302         st := StubPutHandler{
 303                 c,
 304                 hash,
 305                 "abc123",
 306                 "foo",
 307                 make(chan string, 2)}
 308
 309         kc, _ := MakeKeepClient()
 310
 311         kc.Want_replicas = 2
 312         kc.ApiToken = "abc123"
 313         service_roots := make([]string, 5)
 314
 315         ks := RunSomeFakeKeepServers(st, 5, 2990)
 316
 317         for i := 0; i < len(ks); i += 1 {
 318                 service_roots[i] = ks[i].url
 319                 defer ks[i].listener.Close()
 320         }
 321
 322         kc.SetServiceRoots(service_roots)
 323
 324         reader, writer := io.Pipe()
 325
 326         go func() {
 327                 writer.Write([]byte("foo"))
 328                 writer.Close()
 329         }()
 330
 331         kc.PutHR(hash, reader, 3)
 332
 333         shuff := kc.shuffledServiceRoots(hash)
 334         log.Print(shuff)
 335
 336         s1 := <-st.handled
 337         s2 := <-st.handled
 338
 339         c.Check((s1 == shuff[0] && s2 == shuff[1]) ||
 340                 (s1 == shuff[1] && s2 == shuff[0]),
 341                 Equals,
 342                 true)
 343
 344         log.Printf("TestPutHR done")
 345 }
 346
 347 func (s *StandaloneSuite) TestPutWithFail(c *C) {
 348         log.Printf("TestPutWithFail")
 349
 350         hash := fmt.Sprintf("%x", md5.Sum([]byte("foo")))
 351
 352         st := StubPutHandler{
 353                 c,
 354                 hash,
 355                 "abc123",
 356                 "foo",
 357                 make(chan string, 2)}
 358
 359         fh := FailHandler{
 360                 make(chan string, 1)}
 361
 362         kc, _ := MakeKeepClient()
 363
 364         kc.Want_replicas = 2
 365         kc.ApiToken = "abc123"
 366         service_roots := make([]string, 5)
 367
 368         ks1 := RunSomeFakeKeepServers(st, 4, 2990)
 369         ks2 := RunSomeFakeKeepServers(fh, 1, 2995)
 370
 371         for i, k := range ks1 {
 372                 service_roots[i] = k.url
 373                 defer k.listener.Close()
 374         }
 375         for i, k := range ks2 {
 376                 service_roots[len(ks1)+i] = k.url
 377                 defer k.listener.Close()
 378         }
 379
 380         kc.SetServiceRoots(service_roots)
 381
 382         shuff := kc.shuffledServiceRoots(fmt.Sprintf("%x", md5.Sum([]byte("foo"))))
 383
 384         phash, replicas, err := kc.PutB([]byte("foo"))
 385
 386         <-fh.handled
 387
 388         c.Check(err, Equals, nil)
 389         c.Check(phash, Equals, hash)
 390         c.Check(replicas, Equals, 2)
 391         c.Check(<-st.handled, Equals, shuff[1])
 392         c.Check(<-st.handled, Equals, shuff[2])
 393 }
 394
 395 func (s *StandaloneSuite) TestPutWithTooManyFail(c *C) {
 396         log.Printf("TestPutWithTooManyFail")
 397
 398         hash := fmt.Sprintf("%x", md5.Sum([]byte("foo")))
 399
 400         st := StubPutHandler{
 401                 c,
 402                 hash,
 403                 "abc123",
 404                 "foo",
 405                 make(chan string, 1)}
 406
 407         fh := FailHandler{
 408                 make(chan string, 4)}
 409
 410         kc, _ := MakeKeepClient()
 411
 412         kc.Want_replicas = 2
 413         kc.ApiToken = "abc123"
 414         service_roots := make([]string, 5)
 415
 416         ks1 := RunSomeFakeKeepServers(st, 1, 2990)
 417         ks2 := RunSomeFakeKeepServers(fh, 4, 2991)
 418
 419         for i, k := range ks1 {
 420                 service_roots[i] = k.url
 421                 defer k.listener.Close()
 422         }
 423         for i, k := range ks2 {
 424                 service_roots[len(ks1)+i] = k.url
 425                 defer k.listener.Close()
 426         }
 427
 428         kc.SetServiceRoots(service_roots)
 429
 430         shuff := kc.shuffledServiceRoots(fmt.Sprintf("%x", md5.Sum([]byte("foo"))))
 431
 432         _, replicas, err := kc.PutB([]byte("foo"))
 433
 434         c.Check(err, Equals, InsufficientReplicasError)
 435         c.Check(replicas, Equals, 1)
 436         c.Check(<-st.handled, Equals, shuff[1])
 437
 438         log.Printf("TestPutWithTooManyFail done")
 439 }
 440
 441 type StubGetHandler struct {
 442         c              *C
 443         expectPath     string
 444         expectApiToken string
 445         returnBody     []byte
 446 }
 447
 448 func (this StubGetHandler) ServeHTTP(resp http.ResponseWriter, req *http.Request) {
 449         this.c.Check(req.URL.Path, Equals, "/"+this.expectPath)
 450         this.c.Check(req.Header.Get("Authorization"), Equals, fmt.Sprintf("OAuth2 %s", this.expectApiToken))
 451         resp.Header().Set("Content-Length", fmt.Sprintf("%d", len(this.returnBody)))
 452         resp.Write(this.returnBody)
 453 }
 454
 455 func (s *StandaloneSuite) TestGet(c *C) {
 456         log.Printf("TestGet")
 457
 458         hash := fmt.Sprintf("%x", md5.Sum([]byte("foo")))
 459
 460         st := StubGetHandler{
 461                 c,
 462                 hash,
 463                 "abc123",
 464                 []byte("foo")}
 465
 466         listener, url := RunBogusKeepServer(st, 2990)
 467         defer listener.Close()
 468
 469         kc, _ := MakeKeepClient()
 470         kc.ApiToken = "abc123"
 471         kc.SetServiceRoots([]string{url})
 472
 473         r, n, url2, err := kc.Get(hash)
 474         defer r.Close()
 475         c.Check(err, Equals, nil)
 476         c.Check(n, Equals, int64(3))
 477         c.Check(url2, Equals, fmt.Sprintf("%s/%s", url, hash))
 478
 479         content, err2 := ioutil.ReadAll(r)
 480         c.Check(err2, Equals, nil)
 481         c.Check(content, DeepEquals, []byte("foo"))
 482
 483         log.Printf("TestGet done")
 484 }
 485
 486 func (s *StandaloneSuite) TestGetFail(c *C) {
 487         hash := fmt.Sprintf("%x", md5.Sum([]byte("foo")))
 488
 489         st := FailHandler{make(chan string, 1)}
 490
 491         listener, url := RunBogusKeepServer(st, 2990)
 492         defer listener.Close()
 493
 494         kc, _ := MakeKeepClient()
 495         kc.ApiToken = "abc123"
 496         kc.SetServiceRoots([]string{url})
 497
 498         r, n, url2, err := kc.Get(hash)
 499         c.Check(err, Equals, BlockNotFound)
 500         c.Check(n, Equals, int64(0))
 501         c.Check(url2, Equals, "")
 502         c.Check(r, Equals, nil)
 503 }
 504
 505 type BarHandler struct {
 506         handled chan string
 507 }
 508
 509 func (this BarHandler) ServeHTTP(resp http.ResponseWriter, req *http.Request) {
 510         resp.Write([]byte("bar"))
 511         this.handled <- fmt.Sprintf("http://%s", req.Host)
 512 }
 513
 514 func (s *StandaloneSuite) TestChecksum(c *C) {
 515         foohash := fmt.Sprintf("%x", md5.Sum([]byte("foo")))
 516         barhash := fmt.Sprintf("%x", md5.Sum([]byte("bar")))
 517
 518         st := BarHandler{make(chan string, 1)}
 519
 520         listener, url := RunBogusKeepServer(st, 2990)
 521         defer listener.Close()
 522
 523         kc, _ := MakeKeepClient()
 524         kc.ApiToken = "abc123"
 525         kc.SetServiceRoots([]string{url})
 526
 527         r, n, _, err := kc.Get(barhash)
 528         _, err = ioutil.ReadAll(r)
 529         c.Check(n, Equals, int64(3))
 530         c.Check(err, Equals, nil)
 531
 532         <-st.handled
 533
 534         r, n, _, err = kc.Get(foohash)
 535         _, err = ioutil.ReadAll(r)
 536         c.Check(n, Equals, int64(3))
 537         c.Check(err, Equals, BadChecksum)
 538
 539         <-st.handled
 540 }
 541
 542 func (s *StandaloneSuite) TestGetWithFailures(c *C) {
 543
 544         hash := fmt.Sprintf("%x", md5.Sum([]byte("foo")))
 545
 546         fh := FailHandler{
 547                 make(chan string, 1)}
 548
 549         st := StubGetHandler{
 550                 c,
 551                 hash,
 552                 "abc123",
 553                 []byte("foo")}
 554
 555         kc, _ := MakeKeepClient()
 556         kc.ApiToken = "abc123"
 557         service_roots := make([]string, 5)
 558
 559         ks1 := RunSomeFakeKeepServers(st, 1, 2990)
 560         ks2 := RunSomeFakeKeepServers(fh, 4, 2991)
 561
 562         for i, k := range ks1 {
 563                 service_roots[i] = k.url
 564                 defer k.listener.Close()
 565         }
 566         for i, k := range ks2 {
 567                 service_roots[len(ks1)+i] = k.url
 568                 defer k.listener.Close()
 569         }
 570
 571         kc.SetServiceRoots(service_roots)
 572
 573         r, n, url2, err := kc.Get(hash)
 574         <-fh.handled
 575         c.Check(err, Equals, nil)
 576         c.Check(n, Equals, int64(3))
 577         c.Check(url2, Equals, fmt.Sprintf("%s/%s", ks1[0].url, hash))
 578
 579         content, err2 := ioutil.ReadAll(r)
 580         c.Check(err2, Equals, nil)
 581         c.Check(content, DeepEquals, []byte("foo"))
 582 }
 583
 584 func (s *ServerRequiredSuite) TestPutGetHead(c *C) {
 585         os.Setenv("ARVADOS_API_HOST", "localhost:3001")
 586         os.Setenv("ARVADOS_API_TOKEN", "4axaw8zxe0qm22wa6urpp5nskcne8z88cvbupv653y1njyi05h")
 587         os.Setenv("ARVADOS_API_HOST_INSECURE", "true")
 588
 589         kc, err := MakeKeepClient()
 590         c.Assert(err, Equals, nil)
 591
 592         hash := fmt.Sprintf("%x", md5.Sum([]byte("foo")))
 593
 594         {
 595                 n, _, err := kc.Ask(hash)
 596                 c.Check(err, Equals, BlockNotFound)
 597                 c.Check(n, Equals, int64(0))
 598         }
 599         {
 600                 hash2, replicas, err := kc.PutB([]byte("foo"))
 601                 c.Check(hash2, Equals, hash)
 602                 c.Check(replicas, Equals, 2)
 603                 c.Check(err, Equals, nil)
 604         }
 605         {
 606                 r, n, url2, err := kc.Get(hash)
 607                 c.Check(err, Equals, nil)
 608                 c.Check(n, Equals, int64(3))
 609                 c.Check(url2, Equals, fmt.Sprintf("http://localhost:25108/%s", hash))
 610
 611                 content, err2 := ioutil.ReadAll(r)
 612                 c.Check(err2, Equals, nil)
 613                 c.Check(content, DeepEquals, []byte("foo"))
 614         }
 615         {
 616                 n, url2, err := kc.Ask(hash)
 617                 c.Check(err, Equals, nil)
 618                 c.Check(n, Equals, int64(3))
 619                 c.Check(url2, Equals, fmt.Sprintf("http://localhost:25108/%s", hash))
 620         }
 621 }
 622
 623 type StubProxyHandler struct {
 624         handled chan string
 625 }
 626
 627 func (this StubProxyHandler) ServeHTTP(resp http.ResponseWriter, req *http.Request) {
 628         resp.Header().Set("X-Keep-Replicas-Stored", "2")
 629         this.handled <- fmt.Sprintf("http://%s", req.Host)
 630 }
 631
 632 func (s *StandaloneSuite) TestPutProxy(c *C) {
 633         log.Printf("TestPutProxy")
 634
 635         st := StubProxyHandler{make(chan string, 1)}
 636
 637         kc, _ := MakeKeepClient()
 638
 639         kc.Want_replicas = 2
 640         kc.Using_proxy = true
 641         kc.ApiToken = "abc123"
 642         service_roots := make([]string, 1)
 643
 644         ks1 := RunSomeFakeKeepServers(st, 1, 2990)
 645
 646         for i, k := range ks1 {
 647                 service_roots[i] = k.url
 648                 defer k.listener.Close()
 649         }
 650
 651         kc.SetServiceRoots(service_roots)
 652
 653         _, replicas, err := kc.PutB([]byte("foo"))
 654         <-st.handled
 655
 656         c.Check(err, Equals, nil)
 657         c.Check(replicas, Equals, 2)
 658
 659         log.Printf("TestPutProxy done")
 660 }
 661
 662 func (s *StandaloneSuite) TestPutProxyInsufficientReplicas(c *C) {
 663         log.Printf("TestPutProxy")
 664
 665         st := StubProxyHandler{make(chan string, 1)}
 666
 667         kc, _ := MakeKeepClient()
 668
 669         kc.Want_replicas = 3
 670         kc.Using_proxy = true
 671         kc.ApiToken = "abc123"
 672         service_roots := make([]string, 1)
 673
 674         ks1 := RunSomeFakeKeepServers(st, 1, 2990)
 675
 676         for i, k := range ks1 {
 677                 service_roots[i] = k.url
 678                 defer k.listener.Close()
 679         }
 680         kc.SetServiceRoots(service_roots)
 681
 682         _, replicas, err := kc.PutB([]byte("foo"))
 683         <-st.handled
 684
 685         c.Check(err, Equals, InsufficientReplicasError)
 686         c.Check(replicas, Equals, 2)
 687
 688         log.Printf("TestPutProxy done")
 689 }