4 "git.curoverse.com/arvados.git/sdk/go/arvadosclient"
5 "git.curoverse.com/arvados.git/sdk/go/streamer"
20 // Gocheck boilerplate
21 func Test(t *testing.T) {
25 // Gocheck boilerplate
26 var _ = Suite(&ServerRequiredSuite{})
27 var _ = Suite(&StandaloneSuite{})
29 var no_server = flag.Bool("no-server", false, "Skip 'ServerRequireSuite'")
31 // Tests that require the Keep server running
32 type ServerRequiredSuite struct{}
35 type StandaloneSuite struct{}
37 func pythonDir() string {
39 return fmt.Sprintf("%s/../../python/tests", cwd)
42 func (s *ServerRequiredSuite) SetUpSuite(c *C) {
44 c.Skip("Skipping tests that require server")
49 cmd := exec.Command("python", "run_test_server.py", "start")
50 stderr, err := cmd.StderrPipe()
52 log.Fatalf("Setting up stderr pipe: %s", err)
54 go io.Copy(os.Stderr, stderr)
55 if err := cmd.Run(); err != nil {
56 panic(fmt.Sprintf("'python run_test_server.py start' returned error %s", err))
60 cmd := exec.Command("python", "run_test_server.py", "start_keep")
61 stderr, err := cmd.StderrPipe()
63 log.Fatalf("Setting up stderr pipe: %s", err)
65 go io.Copy(os.Stderr, stderr)
66 if err := cmd.Run(); err != nil {
67 panic(fmt.Sprintf("'python run_test_server.py start_keep' returned error %s", err))
72 func (s *ServerRequiredSuite) TearDownSuite(c *C) {
74 exec.Command("python", "run_test_server.py", "stop_keep").Run()
75 exec.Command("python", "run_test_server.py", "stop").Run()
78 func (s *ServerRequiredSuite) TestMakeKeepClient(c *C) {
79 os.Setenv("ARVADOS_API_HOST", "localhost:3000")
80 os.Setenv("ARVADOS_API_TOKEN", "4axaw8zxe0qm22wa6urpp5nskcne8z88cvbupv653y1njyi05h")
81 os.Setenv("ARVADOS_API_HOST_INSECURE", "true")
83 arv, err := arvadosclient.MakeArvadosClient()
84 c.Assert(err, Equals, nil)
86 kc, err := MakeKeepClient(&arv)
88 c.Assert(err, Equals, nil)
89 c.Check(len(kc.ServiceRoots()), Equals, 2)
90 for _, root := range kc.ServiceRoots() {
91 c.Check(root, Matches, "http://localhost:2510[\\d]")
95 type StubPutHandler struct {
103 func (this StubPutHandler) ServeHTTP(resp http.ResponseWriter, req *http.Request) {
104 this.c.Check(req.URL.Path, Equals, "/"+this.expectPath)
105 this.c.Check(req.Header.Get("Authorization"), Equals, fmt.Sprintf("OAuth2 %s", this.expectApiToken))
106 body, err := ioutil.ReadAll(req.Body)
107 this.c.Check(err, Equals, nil)
108 this.c.Check(body, DeepEquals, []byte(this.expectBody))
109 resp.WriteHeader(200)
110 this.handled <- fmt.Sprintf("http://%s", req.Host)
113 func RunBogusKeepServer(st http.Handler, port int) (listener net.Listener, url string) {
115 listener, err = net.ListenTCP("tcp", &net.TCPAddr{Port: port})
117 panic(fmt.Sprintf("Could not listen on tcp port %v", port))
120 url = fmt.Sprintf("http://localhost:%d", port)
122 go http.Serve(listener, st)
126 func UploadToStubHelper(c *C, st http.Handler, f func(KeepClient, string,
127 io.ReadCloser, io.WriteCloser, chan uploadStatus)) {
129 listener, url := RunBogusKeepServer(st, 2990)
130 defer listener.Close()
132 arv, _ := arvadosclient.MakeArvadosClient()
133 arv.ApiToken = "abc123"
135 kc, _ := MakeKeepClient(&arv)
137 reader, writer := io.Pipe()
138 upload_status := make(chan uploadStatus)
140 f(kc, url, reader, writer, upload_status)
143 func (s *StandaloneSuite) TestUploadToStubKeepServer(c *C) {
144 log.Printf("TestUploadToStubKeepServer")
146 st := StubPutHandler{
148 "acbd18db4cc2f85cedef654fccc4a4d8",
153 UploadToStubHelper(c, st,
154 func(kc KeepClient, url string, reader io.ReadCloser,
155 writer io.WriteCloser, upload_status chan uploadStatus) {
157 go kc.uploadToKeepServer(url, st.expectPath, reader, upload_status, int64(len("foo")))
159 writer.Write([]byte("foo"))
163 status := <-upload_status
164 c.Check(status, DeepEquals, uploadStatus{nil, fmt.Sprintf("%s/%s", url, st.expectPath), 200, 1, ""})
167 log.Printf("TestUploadToStubKeepServer done")
170 func (s *StandaloneSuite) TestUploadToStubKeepServerBufferReader(c *C) {
171 log.Printf("TestUploadToStubKeepServerBufferReader")
173 st := StubPutHandler{
175 "acbd18db4cc2f85cedef654fccc4a4d8",
180 UploadToStubHelper(c, st,
181 func(kc KeepClient, url string, reader io.ReadCloser,
182 writer io.WriteCloser, upload_status chan uploadStatus) {
184 tr := streamer.AsyncStreamFromReader(512, reader)
187 br1 := tr.MakeStreamReader()
189 go kc.uploadToKeepServer(url, st.expectPath, br1, upload_status, 3)
191 writer.Write([]byte("foo"))
196 status := <-upload_status
197 c.Check(status, DeepEquals, uploadStatus{nil, fmt.Sprintf("%s/%s", url, st.expectPath), 200, 1, ""})
200 log.Printf("TestUploadToStubKeepServerBufferReader done")
203 type FailHandler struct {
207 func (this FailHandler) ServeHTTP(resp http.ResponseWriter, req *http.Request) {
208 resp.WriteHeader(500)
209 this.handled <- fmt.Sprintf("http://%s", req.Host)
212 func (s *StandaloneSuite) TestFailedUploadToStubKeepServer(c *C) {
213 log.Printf("TestFailedUploadToStubKeepServer")
218 hash := "acbd18db4cc2f85cedef654fccc4a4d8"
220 UploadToStubHelper(c, st,
221 func(kc KeepClient, url string, reader io.ReadCloser,
222 writer io.WriteCloser, upload_status chan uploadStatus) {
224 go kc.uploadToKeepServer(url, hash, reader, upload_status, 3)
226 writer.Write([]byte("foo"))
231 status := <-upload_status
232 c.Check(status.url, Equals, fmt.Sprintf("%s/%s", url, hash))
233 c.Check(status.statusCode, Equals, 500)
235 log.Printf("TestFailedUploadToStubKeepServer done")
238 type KeepServer struct {
239 listener net.Listener
243 func RunSomeFakeKeepServers(st http.Handler, n int, port int) (ks []KeepServer) {
244 ks = make([]KeepServer, n)
246 for i := 0; i < n; i += 1 {
247 boguslistener, bogusurl := RunBogusKeepServer(st, port+i)
248 ks[i] = KeepServer{boguslistener, bogusurl}
254 func (s *StandaloneSuite) TestPutB(c *C) {
255 log.Printf("TestPutB")
257 hash := Md5String("foo")
259 st := StubPutHandler{
264 make(chan string, 5)}
266 arv, _ := arvadosclient.MakeArvadosClient()
267 kc, _ := MakeKeepClient(&arv)
270 arv.ApiToken = "abc123"
271 service_roots := make(map[string]string)
273 ks := RunSomeFakeKeepServers(st, 5, 2990)
275 for i := 0; i < len(ks); i += 1 {
276 service_roots[fmt.Sprintf("zzzzz-bi6l4-fakefakefake%03d", i)] = ks[i].url
277 defer ks[i].listener.Close()
280 kc.SetServiceRoots(service_roots)
282 kc.PutB([]byte("foo"))
284 shuff := NewRootSorter(
285 kc.ServiceRoots(), Md5String("foo")).GetSortedRoots()
289 c.Check((s1 == shuff[0] && s2 == shuff[1]) ||
290 (s1 == shuff[1] && s2 == shuff[0]),
294 log.Printf("TestPutB done")
297 func (s *StandaloneSuite) TestPutHR(c *C) {
298 log.Printf("TestPutHR")
300 hash := fmt.Sprintf("%x", md5.Sum([]byte("foo")))
302 st := StubPutHandler{
307 make(chan string, 5)}
309 arv, _ := arvadosclient.MakeArvadosClient()
310 kc, _ := MakeKeepClient(&arv)
313 arv.ApiToken = "abc123"
314 service_roots := make(map[string]string)
316 ks := RunSomeFakeKeepServers(st, 5, 2990)
318 for i := 0; i < len(ks); i += 1 {
319 service_roots[fmt.Sprintf("zzzzz-bi6l4-fakefakefake%03d", i)] = ks[i].url
320 defer ks[i].listener.Close()
323 kc.SetServiceRoots(service_roots)
325 reader, writer := io.Pipe()
328 writer.Write([]byte("foo"))
332 kc.PutHR(hash, reader, 3)
334 shuff := NewRootSorter(kc.ServiceRoots(), hash).GetSortedRoots()
340 c.Check((s1 == shuff[0] && s2 == shuff[1]) ||
341 (s1 == shuff[1] && s2 == shuff[0]),
345 log.Printf("TestPutHR done")
348 func (s *StandaloneSuite) TestPutWithFail(c *C) {
349 log.Printf("TestPutWithFail")
351 hash := fmt.Sprintf("%x", md5.Sum([]byte("foo")))
353 st := StubPutHandler{
358 make(chan string, 4)}
361 make(chan string, 1)}
363 arv, err := arvadosclient.MakeArvadosClient()
364 kc, _ := MakeKeepClient(&arv)
367 arv.ApiToken = "abc123"
368 service_roots := make(map[string]string)
370 ks1 := RunSomeFakeKeepServers(st, 4, 2990)
371 ks2 := RunSomeFakeKeepServers(fh, 1, 2995)
373 for i, k := range ks1 {
374 service_roots[fmt.Sprintf("zzzzz-bi6l4-fakefakefake%03d", i)] = k.url
375 defer k.listener.Close()
377 for i, k := range ks2 {
378 service_roots[fmt.Sprintf("zzzzz-bi6l4-fakefakefake%03d", i+len(ks1))] = k.url
379 defer k.listener.Close()
382 kc.SetServiceRoots(service_roots)
384 shuff := NewRootSorter(
385 kc.ServiceRoots(), Md5String("foo")).GetSortedRoots()
387 phash, replicas, err := kc.PutB([]byte("foo"))
391 c.Check(err, Equals, nil)
392 c.Check(phash, Equals, "")
393 c.Check(replicas, Equals, 2)
394 c.Check(<-st.handled, Equals, shuff[1])
395 c.Check(<-st.handled, Equals, shuff[2])
398 func (s *StandaloneSuite) TestPutWithTooManyFail(c *C) {
399 log.Printf("TestPutWithTooManyFail")
401 hash := fmt.Sprintf("%x", md5.Sum([]byte("foo")))
403 st := StubPutHandler{
408 make(chan string, 1)}
411 make(chan string, 4)}
413 arv, err := arvadosclient.MakeArvadosClient()
414 kc, _ := MakeKeepClient(&arv)
417 arv.ApiToken = "abc123"
418 service_roots := make(map[string]string)
420 ks1 := RunSomeFakeKeepServers(st, 1, 2990)
421 ks2 := RunSomeFakeKeepServers(fh, 4, 2991)
423 for i, k := range ks1 {
424 service_roots[fmt.Sprintf("zzzzz-bi6l4-fakefakefake%03d", i)] = k.url
425 defer k.listener.Close()
427 for i, k := range ks2 {
428 service_roots[fmt.Sprintf("zzzzz-bi6l4-fakefakefake%03d", i+len(ks1))] = k.url
429 defer k.listener.Close()
432 kc.SetServiceRoots(service_roots)
434 _, replicas, err := kc.PutB([]byte("foo"))
436 c.Check(err, Equals, InsufficientReplicasError)
437 c.Check(replicas, Equals, 1)
438 c.Check(<-st.handled, Matches, ".*2990")
440 log.Printf("TestPutWithTooManyFail done")
443 type StubGetHandler struct {
446 expectApiToken string
450 func (this StubGetHandler) ServeHTTP(resp http.ResponseWriter, req *http.Request) {
451 this.c.Check(req.URL.Path, Equals, "/"+this.expectPath)
452 this.c.Check(req.Header.Get("Authorization"), Equals, fmt.Sprintf("OAuth2 %s", this.expectApiToken))
453 resp.Header().Set("Content-Length", fmt.Sprintf("%d", len(this.returnBody)))
454 resp.Write(this.returnBody)
457 func (s *StandaloneSuite) TestGet(c *C) {
458 log.Printf("TestGet")
460 hash := fmt.Sprintf("%x", md5.Sum([]byte("foo")))
462 st := StubGetHandler{
468 listener, url := RunBogusKeepServer(st, 2990)
469 defer listener.Close()
471 arv, err := arvadosclient.MakeArvadosClient()
472 kc, _ := MakeKeepClient(&arv)
473 arv.ApiToken = "abc123"
474 kc.SetServiceRoots(map[string]string{"x":url})
476 r, n, url2, err := kc.Get(hash)
478 c.Check(err, Equals, nil)
479 c.Check(n, Equals, int64(3))
480 c.Check(url2, Equals, fmt.Sprintf("%s/%s", url, hash))
482 content, err2 := ioutil.ReadAll(r)
483 c.Check(err2, Equals, nil)
484 c.Check(content, DeepEquals, []byte("foo"))
486 log.Printf("TestGet done")
489 func (s *StandaloneSuite) TestGetFail(c *C) {
490 hash := fmt.Sprintf("%x", md5.Sum([]byte("foo")))
492 st := FailHandler{make(chan string, 1)}
494 listener, url := RunBogusKeepServer(st, 2990)
495 defer listener.Close()
497 arv, err := arvadosclient.MakeArvadosClient()
498 kc, _ := MakeKeepClient(&arv)
499 arv.ApiToken = "abc123"
500 kc.SetServiceRoots(map[string]string{"x":url})
502 r, n, url2, err := kc.Get(hash)
503 c.Check(err, Equals, BlockNotFound)
504 c.Check(n, Equals, int64(0))
505 c.Check(url2, Equals, "")
506 c.Check(r, Equals, nil)
509 type BarHandler struct {
513 func (this BarHandler) ServeHTTP(resp http.ResponseWriter, req *http.Request) {
514 resp.Write([]byte("bar"))
515 this.handled <- fmt.Sprintf("http://%s", req.Host)
518 func (s *StandaloneSuite) TestChecksum(c *C) {
519 foohash := fmt.Sprintf("%x", md5.Sum([]byte("foo")))
520 barhash := fmt.Sprintf("%x", md5.Sum([]byte("bar")))
522 st := BarHandler{make(chan string, 1)}
524 listener, url := RunBogusKeepServer(st, 2990)
525 defer listener.Close()
527 arv, err := arvadosclient.MakeArvadosClient()
528 kc, _ := MakeKeepClient(&arv)
529 arv.ApiToken = "abc123"
530 kc.SetServiceRoots(map[string]string{"x":url})
532 r, n, _, err := kc.Get(barhash)
533 _, err = ioutil.ReadAll(r)
534 c.Check(n, Equals, int64(3))
535 c.Check(err, Equals, nil)
539 r, n, _, err = kc.Get(foohash)
540 _, err = ioutil.ReadAll(r)
541 c.Check(n, Equals, int64(3))
542 c.Check(err, Equals, BadChecksum)
547 func (s *StandaloneSuite) TestGetWithFailures(c *C) {
548 content := []byte("waz")
549 hash := fmt.Sprintf("%x", md5.Sum(content))
552 make(chan string, 4)}
554 st := StubGetHandler{
560 arv, err := arvadosclient.MakeArvadosClient()
561 kc, _ := MakeKeepClient(&arv)
562 arv.ApiToken = "abc123"
563 service_roots := make(map[string]string)
565 ks1 := RunSomeFakeKeepServers(st, 1, 2990)
566 ks2 := RunSomeFakeKeepServers(fh, 4, 2991)
568 for i, k := range ks1 {
569 service_roots[fmt.Sprintf("zzzzz-bi6l4-fakefakefake%03d", i)] = k.url
570 defer k.listener.Close()
572 for i, k := range ks2 {
573 service_roots[fmt.Sprintf("zzzzz-bi6l4-fakefakefake%03d", i+len(ks1))] = k.url
574 defer k.listener.Close()
577 kc.SetServiceRoots(service_roots)
579 // This test works only if one of the failing services is
580 // attempted before the succeeding service. Otherwise,
581 // <-fh.handled below will just hang! (Probe order depends on
582 // the choice of block content "waz" and the UUIDs of the fake
583 // servers, so we just tried different strings until we found
584 // an example that passes this Assert.)
585 c.Assert(NewRootSorter(service_roots, hash).GetSortedRoots()[0], Matches, ".*299[1-4]")
587 r, n, url2, err := kc.Get(hash)
590 c.Check(err, Equals, nil)
591 c.Check(n, Equals, int64(3))
592 c.Check(url2, Equals, fmt.Sprintf("%s/%s", ks1[0].url, hash))
594 read_content, err2 := ioutil.ReadAll(r)
595 c.Check(err2, Equals, nil)
596 c.Check(read_content, DeepEquals, content)
599 func (s *ServerRequiredSuite) TestPutGetHead(c *C) {
600 os.Setenv("ARVADOS_API_HOST", "localhost:3000")
601 os.Setenv("ARVADOS_API_TOKEN", "4axaw8zxe0qm22wa6urpp5nskcne8z88cvbupv653y1njyi05h")
602 os.Setenv("ARVADOS_API_HOST_INSECURE", "true")
603 content := []byte("TestPutGetHead")
605 arv, err := arvadosclient.MakeArvadosClient()
606 kc, err := MakeKeepClient(&arv)
607 c.Assert(err, Equals, nil)
609 hash := fmt.Sprintf("%x", md5.Sum(content))
612 n, _, err := kc.Ask(hash)
613 c.Check(err, Equals, BlockNotFound)
614 c.Check(n, Equals, int64(0))
617 hash2, replicas, err := kc.PutB(content)
618 c.Check(hash2, Equals, fmt.Sprintf("%s+%d", hash, len(content)))
619 c.Check(replicas, Equals, 2)
620 c.Check(err, Equals, nil)
623 r, n, url2, err := kc.Get(hash)
624 c.Check(err, Equals, nil)
625 c.Check(n, Equals, int64(len(content)))
626 c.Check(url2, Equals, fmt.Sprintf("http://localhost:25108/%s", hash))
628 read_content, err2 := ioutil.ReadAll(r)
629 c.Check(err2, Equals, nil)
630 c.Check(read_content, DeepEquals, content)
633 n, url2, err := kc.Ask(hash)
634 c.Check(err, Equals, nil)
635 c.Check(n, Equals, int64(len(content)))
636 c.Check(url2, Equals, fmt.Sprintf("http://localhost:25108/%s", hash))
640 type StubProxyHandler struct {
644 func (this StubProxyHandler) ServeHTTP(resp http.ResponseWriter, req *http.Request) {
645 resp.Header().Set("X-Keep-Replicas-Stored", "2")
646 this.handled <- fmt.Sprintf("http://%s", req.Host)
649 func (s *StandaloneSuite) TestPutProxy(c *C) {
650 log.Printf("TestPutProxy")
652 st := StubProxyHandler{make(chan string, 1)}
654 arv, err := arvadosclient.MakeArvadosClient()
655 kc, _ := MakeKeepClient(&arv)
658 kc.Using_proxy = true
659 arv.ApiToken = "abc123"
660 service_roots := make(map[string]string)
662 ks1 := RunSomeFakeKeepServers(st, 1, 2990)
664 for i, k := range ks1 {
665 service_roots[fmt.Sprintf("zzzzz-bi6l4-fakefakefake%03d", i)] = k.url
666 defer k.listener.Close()
669 kc.SetServiceRoots(service_roots)
671 _, replicas, err := kc.PutB([]byte("foo"))
674 c.Check(err, Equals, nil)
675 c.Check(replicas, Equals, 2)
677 log.Printf("TestPutProxy done")
680 func (s *StandaloneSuite) TestPutProxyInsufficientReplicas(c *C) {
681 log.Printf("TestPutProxy")
683 st := StubProxyHandler{make(chan string, 1)}
685 arv, err := arvadosclient.MakeArvadosClient()
686 kc, _ := MakeKeepClient(&arv)
689 kc.Using_proxy = true
690 arv.ApiToken = "abc123"
691 service_roots := make(map[string]string)
693 ks1 := RunSomeFakeKeepServers(st, 1, 2990)
695 for i, k := range ks1 {
696 service_roots[fmt.Sprintf("zzzzz-bi6l4-fakefakefake%03d", i)] = k.url
697 defer k.listener.Close()
699 kc.SetServiceRoots(service_roots)
701 _, replicas, err := kc.PutB([]byte("foo"))
704 c.Check(err, Equals, InsufficientReplicasError)
705 c.Check(replicas, Equals, 2)
707 log.Printf("TestPutProxy done")
710 func (s *StandaloneSuite) TestMakeLocator(c *C) {
711 l := MakeLocator("91f372a266fe2bf2823cb8ec7fda31ce+3+Aabcde@12345678")
713 c.Check(l.Hash, Equals, "91f372a266fe2bf2823cb8ec7fda31ce")
714 c.Check(l.Size, Equals, 3)
715 c.Check(l.Signature, Equals, "abcde")
716 c.Check(l.Timestamp, Equals, "12345678")