package main
import (
+ "bytes"
"crypto/md5"
- "crypto/tls"
"fmt"
"git.curoverse.com/arvados.git/sdk/go/arvadosclient"
"git.curoverse.com/arvados.git/sdk/go/arvadostest"
"git.curoverse.com/arvados.git/sdk/go/keepclient"
- . "gopkg.in/check.v1"
- "io"
"io/ioutil"
"log"
"net/http"
- "net/url"
+ "net/http/httptest"
"os"
"strings"
"testing"
"time"
+
+ . "gopkg.in/check.v1"
)
// Gocheck boilerplate
// Tests that require the Keep server running
type ServerRequiredSuite struct{}
+// Gocheck boilerplate
+var _ = Suite(&NoKeepServerSuite{})
+
+// Test with no keepserver to simulate errors
+type NoKeepServerSuite struct{}
+
+var TestProxyUUID = "zzzzz-bi6l4-lrixqc4fxofbmzz"
+
// Wait (up to 1 second) for keepproxy to listen on a port. This
// avoids a race condition where we hit a "connection refused" error
// because we start testing the proxy too soon.
const (
ms = 5
)
- for i := 0; listener == nil && i < 1000; i += ms {
+ for i := 0; listener == nil && i < 10000; i += ms {
time.Sleep(ms * time.Millisecond)
}
if listener == nil {
func (s *ServerRequiredSuite) SetUpSuite(c *C) {
arvadostest.StartAPI()
- arvadostest.StartKeep()
+ arvadostest.StartKeep(2, false)
}
func (s *ServerRequiredSuite) SetUpTest(c *C) {
}
func (s *ServerRequiredSuite) TearDownSuite(c *C) {
- arvadostest.StopKeep()
+ arvadostest.StopKeep(2)
arvadostest.StopAPI()
}
-func setupProxyService() {
-
- client := &http.Client{Transport: &http.Transport{
- TLSClientConfig: &tls.Config{InsecureSkipVerify: true}}}
-
- var req *http.Request
- var err error
- if req, err = http.NewRequest("POST", fmt.Sprintf("https://%s/arvados/v1/keep_services", os.Getenv("ARVADOS_API_HOST")), nil); err != nil {
- panic(err.Error())
- }
- req.Header.Add("Authorization", fmt.Sprintf("OAuth2 %s", os.Getenv("ARVADOS_API_TOKEN")))
-
- reader, writer := io.Pipe()
-
- req.Body = reader
-
- go func() {
- data := url.Values{}
- data.Set("keep_service", `{
- "service_host": "localhost",
- "service_port": 29950,
- "service_ssl_flag": false,
- "service_type": "proxy"
-}`)
+func (s *NoKeepServerSuite) SetUpSuite(c *C) {
+ arvadostest.StartAPI()
+ // We need API to have some keep services listed, but the
+ // services themselves should be unresponsive.
+ arvadostest.StartKeep(2, false)
+ arvadostest.StopKeep(2)
+}
- writer.Write([]byte(data.Encode()))
- writer.Close()
- }()
+func (s *NoKeepServerSuite) SetUpTest(c *C) {
+ arvadostest.ResetEnv()
+}
- var resp *http.Response
- if resp, err = client.Do(req); err != nil {
- panic(err.Error())
- }
- if resp.StatusCode != 200 {
- panic(resp.Status)
- }
+func (s *NoKeepServerSuite) TearDownSuite(c *C) {
+ arvadostest.StopAPI()
}
-func runProxy(c *C, args []string, port int, bogusClientToken bool) keepclient.KeepClient {
- if bogusClientToken {
- os.Setenv("ARVADOS_API_TOKEN", "bogus-token")
- }
+func runProxy(c *C, args []string, bogusClientToken bool) *keepclient.KeepClient {
+ args = append([]string{"keepproxy"}, args...)
+ os.Args = append(args, "-listen=:0")
+ listener = nil
+ go main()
+ waitForListener()
+
arv, err := arvadosclient.MakeArvadosClient()
c.Assert(err, Equals, nil)
- kc := keepclient.KeepClient{
- Arvados: &arv,
- Want_replicas: 2,
- Using_proxy: true,
- Client: &http.Client{},
- }
- locals := map[string]string{
- "proxy": fmt.Sprintf("http://localhost:%v", port),
- }
- writableLocals := map[string]string{
- "proxy": fmt.Sprintf("http://localhost:%v", port),
- }
- kc.SetServiceRoots(locals, writableLocals, nil)
- c.Check(kc.Using_proxy, Equals, true)
- c.Check(len(kc.LocalRoots()), Equals, 1)
- for _, root := range kc.LocalRoots() {
- c.Check(root, Equals, fmt.Sprintf("http://localhost:%v", port))
- }
- log.Print("keepclient created")
if bogusClientToken {
- arvadostest.ResetEnv()
+ arv.ApiToken = "bogus-token"
}
-
- {
- os.Args = append(args, fmt.Sprintf("-listen=:%v", port))
- listener = nil
- go main()
+ kc := keepclient.New(&arv)
+ sr := map[string]string{
+ TestProxyUUID: "http://" + listener.Addr().String(),
}
+ kc.SetServiceRoots(sr, sr, sr)
+ kc.Arvados.External = true
return kc
}
-func (s *ServerRequiredSuite) TestPutAskGet(c *C) {
- log.Print("TestPutAndGet start")
+func (s *ServerRequiredSuite) TestDesiredReplicas(c *C) {
+ kc := runProxy(c, nil, false)
+ defer closeListener()
- os.Args = []string{"keepproxy", "-listen=:29950"}
- listener = nil
- go main()
- time.Sleep(100 * time.Millisecond)
+ content := []byte("TestDesiredReplicas")
+ hash := fmt.Sprintf("%x", md5.Sum(content))
- setupProxyService()
+ for _, kc.Want_replicas = range []int{0, 1, 2} {
+ locator, rep, err := kc.PutB(content)
+ c.Check(err, Equals, nil)
+ c.Check(rep, Equals, kc.Want_replicas)
+ if rep > 0 {
+ c.Check(locator, Matches, fmt.Sprintf(`^%s\+%d(\+.+)?$`, hash, len(content)))
+ }
+ }
+}
- os.Setenv("ARVADOS_EXTERNAL_CLIENT", "true")
- arv, err := arvadosclient.MakeArvadosClient()
- c.Assert(err, Equals, nil)
- kc, err := keepclient.MakeKeepClient(&arv)
- c.Assert(err, Equals, nil)
- c.Check(kc.Arvados.External, Equals, true)
- c.Check(kc.Using_proxy, Equals, true)
- c.Check(len(kc.LocalRoots()), Equals, 1)
- for _, root := range kc.LocalRoots() {
- c.Check(root, Equals, "http://localhost:29950")
+func (s *ServerRequiredSuite) TestPutWrongContentLength(c *C) {
+ kc := runProxy(c, nil, false)
+ defer closeListener()
+
+ content := []byte("TestPutWrongContentLength")
+ hash := fmt.Sprintf("%x", md5.Sum(content))
+
+ // If we use http.Client to send these requests to the network
+ // server we just started, the Go http library automatically
+ // fixes the invalid Content-Length header. In order to test
+ // our server behavior, we have to call the handler directly
+ // using an httptest.ResponseRecorder.
+ rtr := MakeRESTRouter(true, true, kc)
+
+ type testcase struct {
+ sendLength string
+ expectStatus int
+ }
+
+ for _, t := range []testcase{
+ {"1", http.StatusBadRequest},
+ {"", http.StatusLengthRequired},
+ {"-1", http.StatusLengthRequired},
+ {"abcdef", http.StatusLengthRequired},
+ } {
+ req, err := http.NewRequest("PUT",
+ fmt.Sprintf("http://%s/%s+%d", listener.Addr().String(), hash, len(content)),
+ bytes.NewReader(content))
+ c.Assert(err, IsNil)
+ req.Header.Set("Content-Length", t.sendLength)
+ req.Header.Set("Authorization", "OAuth2 4axaw8zxe0qm22wa6urpp5nskcne8z88cvbupv653y1njyi05h")
+ req.Header.Set("Content-Type", "application/octet-stream")
+
+ resp := httptest.NewRecorder()
+ rtr.ServeHTTP(resp, req)
+ c.Check(resp.Code, Equals, t.expectStatus)
}
- os.Setenv("ARVADOS_EXTERNAL_CLIENT", "")
+}
- waitForListener()
+func (s *ServerRequiredSuite) TestPutAskGet(c *C) {
+ kc := runProxy(c, nil, false)
defer closeListener()
hash := fmt.Sprintf("%x", md5.Sum([]byte("foo")))
c.Check(blocklen, Equals, int64(0))
log.Print("Finished Get zero block")
}
-
- log.Print("TestPutAndGet done")
}
func (s *ServerRequiredSuite) TestPutAskGetForbidden(c *C) {
- log.Print("TestPutAskGetForbidden start")
-
- kc := runProxy(c, []string{"keepproxy"}, 29951, true)
- waitForListener()
+ kc := runProxy(c, nil, true)
defer closeListener()
hash := fmt.Sprintf("%x", md5.Sum([]byte("bar")))
{
_, _, err := kc.Ask(hash)
- c.Check(err, Equals, keepclient.BlockNotFound)
+ errNotFound, _ := err.(keepclient.ErrNotFound)
+ c.Check(errNotFound, NotNil)
+ c.Assert(strings.Contains(err.Error(), "HTTP 403"), Equals, true)
log.Print("Ask 1")
}
{
blocklen, _, err := kc.Ask(hash)
- c.Assert(err, Equals, keepclient.BlockNotFound)
+ errNotFound, _ := err.(keepclient.ErrNotFound)
+ c.Check(errNotFound, NotNil)
+ c.Assert(strings.Contains(err.Error(), "HTTP 403"), Equals, true)
c.Check(blocklen, Equals, int64(0))
log.Print("Ask 2")
}
{
_, blocklen, _, err := kc.Get(hash)
- c.Assert(err, Equals, keepclient.BlockNotFound)
+ errNotFound, _ := err.(keepclient.ErrNotFound)
+ c.Check(errNotFound, NotNil)
+ c.Assert(strings.Contains(err.Error(), "HTTP 403"), Equals, true)
c.Check(blocklen, Equals, int64(0))
log.Print("Get")
}
-
- log.Print("TestPutAskGetForbidden done")
}
func (s *ServerRequiredSuite) TestGetDisabled(c *C) {
- log.Print("TestGetDisabled start")
-
- kc := runProxy(c, []string{"keepproxy", "-no-get"}, 29952, false)
- waitForListener()
+ kc := runProxy(c, []string{"-no-get"}, false)
defer closeListener()
hash := fmt.Sprintf("%x", md5.Sum([]byte("baz")))
{
_, _, err := kc.Ask(hash)
- c.Check(err, Equals, keepclient.BlockNotFound)
+ errNotFound, _ := err.(keepclient.ErrNotFound)
+ c.Check(errNotFound, NotNil)
+ c.Assert(strings.Contains(err.Error(), "HTTP 400"), Equals, true)
log.Print("Ask 1")
}
{
blocklen, _, err := kc.Ask(hash)
- c.Assert(err, Equals, keepclient.BlockNotFound)
+ errNotFound, _ := err.(keepclient.ErrNotFound)
+ c.Check(errNotFound, NotNil)
+ c.Assert(strings.Contains(err.Error(), "HTTP 400"), Equals, true)
c.Check(blocklen, Equals, int64(0))
log.Print("Ask 2")
}
{
_, blocklen, _, err := kc.Get(hash)
- c.Assert(err, Equals, keepclient.BlockNotFound)
+ errNotFound, _ := err.(keepclient.ErrNotFound)
+ c.Check(errNotFound, NotNil)
+ c.Assert(strings.Contains(err.Error(), "HTTP 400"), Equals, true)
c.Check(blocklen, Equals, int64(0))
log.Print("Get")
}
-
- log.Print("TestGetDisabled done")
}
func (s *ServerRequiredSuite) TestPutDisabled(c *C) {
- log.Print("TestPutDisabled start")
-
- kc := runProxy(c, []string{"keepproxy", "-no-put"}, 29953, false)
- waitForListener()
+ kc := runProxy(c, []string{"-no-put"}, false)
defer closeListener()
- {
- hash2, rep, err := kc.PutB([]byte("quux"))
- c.Check(hash2, Equals, "")
- c.Check(rep, Equals, 0)
- c.Check(err, Equals, keepclient.InsufficientReplicasError)
- log.Print("PutB")
- }
-
- log.Print("TestPutDisabled done")
+ hash2, rep, err := kc.PutB([]byte("quux"))
+ c.Check(hash2, Equals, "")
+ c.Check(rep, Equals, 0)
+ c.Check(err, Equals, keepclient.InsufficientReplicasError)
}
func (s *ServerRequiredSuite) TestCorsHeaders(c *C) {
- runProxy(c, []string{"keepproxy"}, 29954, false)
- waitForListener()
+ runProxy(c, nil, false)
defer closeListener()
{
client := http.Client{}
req, err := http.NewRequest("OPTIONS",
- fmt.Sprintf("http://localhost:29954/%x+3",
- md5.Sum([]byte("foo"))),
+ fmt.Sprintf("http://%s/%x+3", listener.Addr().String(), md5.Sum([]byte("foo"))),
nil)
req.Header.Add("Access-Control-Request-Method", "PUT")
req.Header.Add("Access-Control-Request-Headers", "Authorization, X-Keep-Desired-Replicas")
{
resp, err := http.Get(
- fmt.Sprintf("http://localhost:29954/%x+3",
- md5.Sum([]byte("foo"))))
+ fmt.Sprintf("http://%s/%x+3", listener.Addr().String(), md5.Sum([]byte("foo"))))
c.Check(err, Equals, nil)
c.Check(resp.Header.Get("Access-Control-Allow-Headers"), Equals, "Authorization, Content-Length, Content-Type, X-Keep-Desired-Replicas")
c.Check(resp.Header.Get("Access-Control-Allow-Origin"), Equals, "*")
}
func (s *ServerRequiredSuite) TestPostWithoutHash(c *C) {
- runProxy(c, []string{"keepproxy"}, 29955, false)
- waitForListener()
+ runProxy(c, nil, false)
defer closeListener()
{
client := http.Client{}
req, err := http.NewRequest("POST",
- "http://localhost:29955/",
+ "http://"+listener.Addr().String()+"/",
strings.NewReader("qux"))
req.Header.Add("Authorization", "OAuth2 4axaw8zxe0qm22wa6urpp5nskcne8z88cvbupv653y1njyi05h")
req.Header.Add("Content-Type", "application/octet-stream")
}
+// Test GetIndex
+// Put one block, with 2 replicas
+// With no prefix (expect the block locator, twice)
+// With an existing prefix (expect the block locator, twice)
+// With a valid but non-existing prefix (expect "\n")
+// With an invalid prefix (expect error)
func (s *ServerRequiredSuite) TestGetIndex(c *C) {
- kc := runProxy(c, []string{"keepproxy"}, 28852, false)
- waitForListener()
+ kc := runProxy(c, nil, false)
defer closeListener()
+ // Put "index-data" blocks
data := []byte("index-data")
hash := fmt.Sprintf("%x", md5.Sum(data))
all, err := ioutil.ReadAll(reader)
c.Check(all, DeepEquals, data)
- // GetIndex with no prefix
- indexReader, err := kc.GetIndex("proxy", "")
- c.Assert(err, Equals, nil)
- indexResp, err := ioutil.ReadAll(indexReader)
- locators := strings.Split(string(indexResp), "\n")
- count := 0
- for _, locator := range locators {
- if strings.HasPrefix(locator, hash) {
- count += 1
+ // Put some more blocks
+ _, rep, err = kc.PutB([]byte("some-more-index-data"))
+ c.Check(err, Equals, nil)
+
+ kc.Arvados.ApiToken = arvadostest.DataManagerToken
+
+ // Invoke GetIndex
+ for _, spec := range []struct {
+ prefix string
+ expectTestHash bool
+ expectOther bool
+ }{
+ {"", true, true}, // with no prefix
+ {hash[:3], true, false}, // with matching prefix
+ {"abcdef", false, false}, // with no such prefix
+ } {
+ indexReader, err := kc.GetIndex(TestProxyUUID, spec.prefix)
+ c.Assert(err, Equals, nil)
+ indexResp, err := ioutil.ReadAll(indexReader)
+ c.Assert(err, Equals, nil)
+ locators := strings.Split(string(indexResp), "\n")
+ gotTestHash := 0
+ gotOther := 0
+ for _, locator := range locators {
+ if locator == "" {
+ continue
+ }
+ c.Check(locator[:len(spec.prefix)], Equals, spec.prefix)
+ if locator[:32] == hash {
+ gotTestHash++
+ } else {
+ gotOther++
+ }
}
+ c.Check(gotTestHash == 2, Equals, spec.expectTestHash)
+ c.Check(gotOther > 0, Equals, spec.expectOther)
+ }
+
+ // GetIndex with invalid prefix
+ _, err = kc.GetIndex(TestProxyUUID, "xyz")
+ c.Assert((err != nil), Equals, true)
+}
+
+func (s *ServerRequiredSuite) TestPutAskGetInvalidToken(c *C) {
+ kc := runProxy(c, nil, false)
+ defer closeListener()
+
+ // Put a test block
+ hash, rep, err := kc.PutB([]byte("foo"))
+ c.Check(err, Equals, nil)
+ c.Check(rep, Equals, 2)
+
+ for _, token := range []string{
+ "nosuchtoken",
+ "2ym314ysp27sk7h943q6vtc378srb06se3pq6ghurylyf3pdmx", // expired
+ } {
+ // Change token to given bad token
+ kc.Arvados.ApiToken = token
+
+ // Ask should result in error
+ _, _, err = kc.Ask(hash)
+ c.Check(err, NotNil)
+ errNotFound, _ := err.(keepclient.ErrNotFound)
+ c.Check(errNotFound.Temporary(), Equals, false)
+ c.Assert(strings.Contains(err.Error(), "HTTP 403"), Equals, true)
+
+ // Get should result in error
+ _, _, _, err = kc.Get(hash)
+ c.Check(err, NotNil)
+ errNotFound, _ = err.(keepclient.ErrNotFound)
+ c.Check(errNotFound.Temporary(), Equals, false)
+ c.Assert(strings.Contains(err.Error(), "HTTP 403 \"Missing or invalid Authorization header\""), Equals, true)
}
- c.Assert(2, Equals, count)
+}
- // GetIndex with prefix
- indexReader, err = kc.GetIndex("proxy", hash[0:3])
+func (s *ServerRequiredSuite) TestAskGetKeepProxyConnectionError(c *C) {
+ arv, err := arvadosclient.MakeArvadosClient()
c.Assert(err, Equals, nil)
- indexResp, err = ioutil.ReadAll(indexReader)
- locators = strings.Split(string(indexResp), "\n")
- count = 0
- for _, locator := range locators {
- if strings.HasPrefix(locator, hash) {
- count += 1
- }
+
+ // keepclient with no such keep server
+ kc := keepclient.New(&arv)
+ locals := map[string]string{
+ TestProxyUUID: "http://localhost:12345",
}
- c.Assert(2, Equals, count)
+ kc.SetServiceRoots(locals, nil, nil)
- // GetIndex with no such prefix
- indexReader, err = kc.GetIndex("proxy", "xyz")
- c.Assert((err != nil), Equals, true)
+ // Ask should result in temporary connection refused error
+ hash := fmt.Sprintf("%x", md5.Sum([]byte("foo")))
+ _, _, err = kc.Ask(hash)
+ c.Check(err, NotNil)
+ errNotFound, _ := err.(*keepclient.ErrNotFound)
+ c.Check(errNotFound.Temporary(), Equals, true)
+ c.Assert(strings.Contains(err.Error(), "connection refused"), Equals, true)
+
+ // Get should result in temporary connection refused error
+ _, _, _, err = kc.Get(hash)
+ c.Check(err, NotNil)
+ errNotFound, _ = err.(*keepclient.ErrNotFound)
+ c.Check(errNotFound.Temporary(), Equals, true)
+ c.Assert(strings.Contains(err.Error(), "connection refused"), Equals, true)
+}
+
+func (s *NoKeepServerSuite) TestAskGetNoKeepServerError(c *C) {
+ kc := runProxy(c, nil, false)
+ defer closeListener()
+
+ hash := fmt.Sprintf("%x", md5.Sum([]byte("foo")))
+ for _, f := range []func() error{
+ func() error {
+ _, _, err := kc.Ask(hash)
+ return err
+ },
+ func() error {
+ _, _, _, err := kc.Get(hash)
+ return err
+ },
+ } {
+ err := f()
+ c.Assert(err, NotNil)
+ errNotFound, _ := err.(*keepclient.ErrNotFound)
+ c.Check(errNotFound.Temporary(), Equals, true)
+ c.Check(err, ErrorMatches, `.*HTTP 502.*`)
+ }
}
projects / arvados.git / blobdiff