services/keepproxy/keepproxy_test.go

   1 package main
   2
   3 import (
   4         "git.curoverse.com/arvados.git/sdk/go/keepclient"
   5         "git.curoverse.com/arvados.git/sdk/go/arvadosclient"
   6         "crypto/md5"
   7         "crypto/tls"
   8         "fmt"
   9         . "gopkg.in/check.v1"
  10         "io"
  11         "io/ioutil"
  12         "log"
  13         "net/http"
  14         "net/url"
  15         "os"
  16         "os/exec"
  17         "testing"
  18         "time"
  19 )
  20
  21 // Gocheck boilerplate
  22 func Test(t *testing.T) {
  23         TestingT(t)
  24 }
  25
  26 // Gocheck boilerplate
  27 var _ = Suite(&ServerRequiredSuite{})
  28
  29 // Tests that require the Keep server running
  30 type ServerRequiredSuite struct{}
  31
  32 func pythonDir() string {
  33         cwd, _ := os.Getwd()
  34         return fmt.Sprintf("%s/../../sdk/python/tests", cwd)
  35 }
  36
  37 // Wait (up to 1 second) for keepproxy to listen on a port. This
  38 // avoids a race condition where we hit a "connection refused" error
  39 // because we start testing the proxy too soon.
  40 func waitForListener() {
  41         const (ms = 5)
  42         for i := 0; listener == nil && i < 1000; i += ms {
  43                 time.Sleep(ms * time.Millisecond)
  44         }
  45         if listener == nil {
  46                 log.Fatalf("Timed out waiting for listener to start")
  47         }
  48 }
  49
  50 func closeListener() {
  51         if listener != nil {
  52                 listener.Close()
  53         }
  54 }
  55
  56 func (s *ServerRequiredSuite) SetUpSuite(c *C) {
  57         cwd, _ := os.Getwd()
  58         defer os.Chdir(cwd)
  59
  60         os.Chdir(pythonDir())
  61         {
  62                 cmd := exec.Command("python", "run_test_server.py", "start")
  63                 stderr, err := cmd.StderrPipe()
  64                 if err != nil {
  65                         log.Fatalf("Setting up stderr pipe: %s", err)
  66                 }
  67                 go io.Copy(os.Stderr, stderr)
  68                 if err := cmd.Run(); err != nil {
  69                         panic(fmt.Sprintf("'python run_test_server.py start' returned error %s", err))
  70                 }
  71         }
  72         {
  73                 cmd := exec.Command("python", "run_test_server.py", "start_keep")
  74                 stderr, err := cmd.StderrPipe()
  75                 if err != nil {
  76                         log.Fatalf("Setting up stderr pipe: %s", err)
  77                 }
  78                 go io.Copy(os.Stderr, stderr)
  79                 if err := cmd.Run(); err != nil {
  80                         panic(fmt.Sprintf("'python run_test_server.py start_keep' returned error %s", err))
  81                 }
  82         }
  83
  84         os.Setenv("ARVADOS_API_HOST", "localhost:3000")
  85         os.Setenv("ARVADOS_API_TOKEN", "4axaw8zxe0qm22wa6urpp5nskcne8z88cvbupv653y1njyi05h")
  86         os.Setenv("ARVADOS_API_HOST_INSECURE", "true")
  87 }
  88
  89 func (s *ServerRequiredSuite) TearDownSuite(c *C) {
  90         cwd, _ := os.Getwd()
  91         defer os.Chdir(cwd)
  92
  93         os.Chdir(pythonDir())
  94         exec.Command("python", "run_test_server.py", "stop_keep").Run()
  95         exec.Command("python", "run_test_server.py", "stop").Run()
  96 }
  97
  98 func setupProxyService() {
  99
 100         client := &http.Client{Transport: &http.Transport{
 101                 TLSClientConfig: &tls.Config{InsecureSkipVerify: true}}}
 102
 103         var req *http.Request
 104         var err error
 105         if req, err = http.NewRequest("POST", fmt.Sprintf("https://%s/arvados/v1/keep_services", os.Getenv("ARVADOS_API_HOST")), nil); err != nil {
 106                 panic(err.Error())
 107         }
 108         req.Header.Add("Authorization", fmt.Sprintf("OAuth2 %s", os.Getenv("ARVADOS_API_TOKEN")))
 109
 110         reader, writer := io.Pipe()
 111
 112         req.Body = reader
 113
 114         go func() {
 115                 data := url.Values{}
 116                 data.Set("keep_service", `{
 117   "service_host": "localhost",
 118   "service_port": 29950,
 119   "service_ssl_flag": false,
 120   "service_type": "proxy"
 121 }`)
 122
 123                 writer.Write([]byte(data.Encode()))
 124                 writer.Close()
 125         }()
 126
 127         var resp *http.Response
 128         if resp, err = client.Do(req); err != nil {
 129                 panic(err.Error())
 130         }
 131         if resp.StatusCode != 200 {
 132                 panic(resp.Status)
 133         }
 134 }
 135
 136 func runProxy(c *C, args []string, token string, port int) keepclient.KeepClient {
 137         os.Args = append(args, fmt.Sprintf("-listen=:%v", port))
 138         os.Setenv("ARVADOS_API_TOKEN", "4axaw8zxe0qm22wa6urpp5nskcne8z88cvbupv653y1njyi05h")
 139
 140         go main()
 141         time.Sleep(100 * time.Millisecond)
 142
 143         os.Setenv("ARVADOS_KEEP_PROXY", fmt.Sprintf("http://localhost:%v", port))
 144         os.Setenv("ARVADOS_API_TOKEN", token)
 145         arv, err := arvadosclient.MakeArvadosClient()
 146         c.Assert(err, Equals, nil)
 147         kc, err := keepclient.MakeKeepClient(&arv)
 148         c.Assert(err, Equals, nil)
 149         c.Check(kc.Using_proxy, Equals, true)
 150         c.Check(len(kc.ServiceRoots()), Equals, 1)
 151         for _, root := range(kc.ServiceRoots()) {
 152                 c.Check(root, Equals, fmt.Sprintf("http://localhost:%v", port))
 153         }
 154         os.Setenv("ARVADOS_KEEP_PROXY", "")
 155         log.Print("keepclient created")
 156         return kc
 157 }
 158
 159 func (s *ServerRequiredSuite) TestPutAskGet(c *C) {
 160         log.Print("TestPutAndGet start")
 161
 162         os.Args = []string{"keepproxy", "-listen=:29950"}
 163         os.Setenv("ARVADOS_API_TOKEN", "4axaw8zxe0qm22wa6urpp5nskcne8z88cvbupv653y1njyi05h")
 164         go main()
 165         time.Sleep(100 * time.Millisecond)
 166
 167         setupProxyService()
 168
 169         os.Setenv("ARVADOS_EXTERNAL_CLIENT", "true")
 170         arv, err := arvadosclient.MakeArvadosClient()
 171         c.Assert(err, Equals, nil)
 172         kc, err := keepclient.MakeKeepClient(&arv)
 173         c.Assert(err, Equals, nil)
 174         c.Check(kc.Arvados.External, Equals, true)
 175         c.Check(kc.Using_proxy, Equals, true)
 176         c.Check(len(kc.ServiceRoots()), Equals, 1)
 177         for _, root := range kc.ServiceRoots() {
 178                 c.Check(root, Equals, "http://localhost:29950")
 179         }
 180         os.Setenv("ARVADOS_EXTERNAL_CLIENT", "")
 181         log.Print("keepclient created")
 182
 183         waitForListener()
 184         defer closeListener()
 185
 186         hash := fmt.Sprintf("%x", md5.Sum([]byte("foo")))
 187         var hash2 string
 188
 189         {
 190                 _, _, err := kc.Ask(hash)
 191                 c.Check(err, Equals, keepclient.BlockNotFound)
 192                 log.Print("Ask 1")
 193         }
 194
 195         {
 196                 var rep int
 197                 var err error
 198                 hash2, rep, err = kc.PutB([]byte("foo"))
 199                 c.Check(hash2, Matches, fmt.Sprintf(`^%s\+3(\+.+)?$`, hash))
 200                 c.Check(rep, Equals, 2)
 201                 c.Check(err, Equals, nil)
 202                 log.Print("PutB")
 203         }
 204
 205         {
 206                 blocklen, _, err := kc.Ask(hash2)
 207                 c.Assert(err, Equals, nil)
 208                 c.Check(blocklen, Equals, int64(3))
 209                 log.Print("Ask 2")
 210         }
 211
 212         {
 213                 reader, blocklen, _, err := kc.Get(hash2)
 214                 c.Assert(err, Equals, nil)
 215                 all, err := ioutil.ReadAll(reader)
 216                 c.Check(all, DeepEquals, []byte("foo"))
 217                 c.Check(blocklen, Equals, int64(3))
 218                 log.Print("Get")
 219         }
 220
 221         log.Print("TestPutAndGet done")
 222 }
 223
 224 func (s *ServerRequiredSuite) TestPutAskGetForbidden(c *C) {
 225         log.Print("TestPutAndGet start")
 226
 227         kc := runProxy(c, []string{"keepproxy"}, "123abc", 29951)
 228         waitForListener()
 229         defer closeListener()
 230
 231         log.Print("keepclient created")
 232
 233         hash := fmt.Sprintf("%x", md5.Sum([]byte("bar")))
 234
 235         {
 236                 _, _, err := kc.Ask(hash)
 237                 c.Check(err, Equals, keepclient.BlockNotFound)
 238                 log.Print("Ask 1")
 239         }
 240
 241         {
 242                 hash2, rep, err := kc.PutB([]byte("bar"))
 243                 c.Check(hash2, Equals, "")
 244                 c.Check(rep, Equals, 0)
 245                 c.Check(err, Equals, keepclient.InsufficientReplicasError)
 246                 log.Print("PutB")
 247         }
 248
 249         {
 250                 blocklen, _, err := kc.Ask(hash)
 251                 c.Assert(err, Equals, keepclient.BlockNotFound)
 252                 c.Check(blocklen, Equals, int64(0))
 253                 log.Print("Ask 2")
 254         }
 255
 256         {
 257                 _, blocklen, _, err := kc.Get(hash)
 258                 c.Assert(err, Equals, keepclient.BlockNotFound)
 259                 c.Check(blocklen, Equals, int64(0))
 260                 log.Print("Get")
 261         }
 262
 263         log.Print("TestPutAndGetForbidden done")
 264 }
 265
 266 func (s *ServerRequiredSuite) TestGetDisabled(c *C) {
 267         log.Print("TestGetDisabled start")
 268
 269         kc := runProxy(c, []string{"keepproxy", "-no-get"}, "4axaw8zxe0qm22wa6urpp5nskcne8z88cvbupv653y1njyi05h", 29952)
 270         waitForListener()
 271         defer closeListener()
 272
 273         hash := fmt.Sprintf("%x", md5.Sum([]byte("baz")))
 274
 275         {
 276                 _, _, err := kc.Ask(hash)
 277                 c.Check(err, Equals, keepclient.BlockNotFound)
 278                 log.Print("Ask 1")
 279         }
 280
 281         {
 282                 hash2, rep, err := kc.PutB([]byte("baz"))
 283                 c.Check(hash2, Matches, fmt.Sprintf(`^%s\+3(\+.+)?$`, hash))
 284                 c.Check(rep, Equals, 2)
 285                 c.Check(err, Equals, nil)
 286                 log.Print("PutB")
 287         }
 288
 289         {
 290                 blocklen, _, err := kc.Ask(hash)
 291                 c.Assert(err, Equals, keepclient.BlockNotFound)
 292                 c.Check(blocklen, Equals, int64(0))
 293                 log.Print("Ask 2")
 294         }
 295
 296         {
 297                 _, blocklen, _, err := kc.Get(hash)
 298                 c.Assert(err, Equals, keepclient.BlockNotFound)
 299                 c.Check(blocklen, Equals, int64(0))
 300                 log.Print("Get")
 301         }
 302
 303         log.Print("TestGetDisabled done")
 304 }
 305
 306 func (s *ServerRequiredSuite) TestPutDisabled(c *C) {
 307         log.Print("TestPutDisabled start")
 308
 309         kc := runProxy(c, []string{"keepproxy", "-no-put"}, "4axaw8zxe0qm22wa6urpp5nskcne8z88cvbupv653y1njyi05h", 29953)
 310         waitForListener()
 311         defer closeListener()
 312
 313         {
 314                 hash2, rep, err := kc.PutB([]byte("quux"))
 315                 c.Check(hash2, Equals, "")
 316                 c.Check(rep, Equals, 0)
 317                 c.Check(err, Equals, keepclient.InsufficientReplicasError)
 318                 log.Print("PutB")
 319         }
 320
 321         log.Print("TestPutDisabled done")
 322 }