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         kc, err := keepclient.MakeKeepClient(&arv)
 147         c.Check(kc.Using_proxy, Equals, true)
 148         c.Check(len(kc.ServiceRoots()), Equals, 1)
 149         c.Check(kc.ServiceRoots()[0], Equals, fmt.Sprintf("http://localhost:%v", port))
 150         c.Check(err, Equals, nil)
 151         os.Setenv("ARVADOS_KEEP_PROXY", "")
 152         log.Print("keepclient created")
 153         return kc
 154 }
 155
 156 func (s *ServerRequiredSuite) TestPutAskGet(c *C) {
 157         log.Print("TestPutAndGet start")
 158
 159         os.Args = []string{"keepproxy", "-listen=:29950"}
 160         os.Setenv("ARVADOS_API_TOKEN", "4axaw8zxe0qm22wa6urpp5nskcne8z88cvbupv653y1njyi05h")
 161         go main()
 162         time.Sleep(100 * time.Millisecond)
 163
 164         setupProxyService()
 165
 166         os.Setenv("ARVADOS_EXTERNAL_CLIENT", "true")
 167         arv, err := arvadosclient.MakeArvadosClient()
 168         kc, err := keepclient.MakeKeepClient(&arv)
 169         c.Check(kc.Arvados.External, Equals, true)
 170         c.Check(kc.Using_proxy, Equals, true)
 171         c.Check(len(kc.ServiceRoots()), Equals, 1)
 172         c.Check(kc.ServiceRoots()[0], Equals, "http://localhost:29950")
 173         c.Check(err, Equals, nil)
 174         os.Setenv("ARVADOS_EXTERNAL_CLIENT", "")
 175         log.Print("keepclient created")
 176
 177         waitForListener()
 178         defer closeListener()
 179
 180         hash := fmt.Sprintf("%x", md5.Sum([]byte("foo")))
 181         var hash2 string
 182
 183         {
 184                 _, _, err := kc.Ask(hash)
 185                 c.Check(err, Equals, keepclient.BlockNotFound)
 186                 log.Print("Ask 1")
 187         }
 188
 189         {
 190                 var rep int
 191                 var err error
 192                 hash2, rep, err = kc.PutB([]byte("foo"))
 193                 c.Check(hash2, Matches, fmt.Sprintf(`^%s\+3(\+.+)?$`, hash))
 194                 c.Check(rep, Equals, 2)
 195                 c.Check(err, Equals, nil)
 196                 log.Print("PutB")
 197         }
 198
 199         {
 200                 blocklen, _, err := kc.Ask(hash2)
 201                 c.Assert(err, Equals, nil)
 202                 c.Check(blocklen, Equals, int64(3))
 203                 log.Print("Ask 2")
 204         }
 205
 206         {
 207                 reader, blocklen, _, err := kc.Get(hash2)
 208                 c.Assert(err, Equals, nil)
 209                 all, err := ioutil.ReadAll(reader)
 210                 c.Check(all, DeepEquals, []byte("foo"))
 211                 c.Check(blocklen, Equals, int64(3))
 212                 log.Print("Get")
 213         }
 214
 215         log.Print("TestPutAndGet done")
 216 }
 217
 218 func (s *ServerRequiredSuite) TestPutAskGetForbidden(c *C) {
 219         log.Print("TestPutAndGet start")
 220
 221         kc := runProxy(c, []string{"keepproxy"}, "123abc", 29951)
 222         waitForListener()
 223         defer closeListener()
 224
 225         log.Print("keepclient created")
 226
 227         hash := fmt.Sprintf("%x", md5.Sum([]byte("bar")))
 228
 229         {
 230                 _, _, err := kc.Ask(hash)
 231                 c.Check(err, Equals, keepclient.BlockNotFound)
 232                 log.Print("Ask 1")
 233         }
 234
 235         {
 236                 hash2, rep, err := kc.PutB([]byte("bar"))
 237                 c.Check(hash2, Equals, "")
 238                 c.Check(rep, Equals, 0)
 239                 c.Check(err, Equals, keepclient.InsufficientReplicasError)
 240                 log.Print("PutB")
 241         }
 242
 243         {
 244                 blocklen, _, err := kc.Ask(hash)
 245                 c.Assert(err, Equals, keepclient.BlockNotFound)
 246                 c.Check(blocklen, Equals, int64(0))
 247                 log.Print("Ask 2")
 248         }
 249
 250         {
 251                 _, blocklen, _, err := kc.Get(hash)
 252                 c.Assert(err, Equals, keepclient.BlockNotFound)
 253                 c.Check(blocklen, Equals, int64(0))
 254                 log.Print("Get")
 255         }
 256
 257         log.Print("TestPutAndGetForbidden done")
 258 }
 259
 260 func (s *ServerRequiredSuite) TestGetDisabled(c *C) {
 261         log.Print("TestGetDisabled start")
 262
 263         kc := runProxy(c, []string{"keepproxy", "-no-get"}, "4axaw8zxe0qm22wa6urpp5nskcne8z88cvbupv653y1njyi05h", 29952)
 264         waitForListener()
 265         defer closeListener()
 266
 267         hash := fmt.Sprintf("%x", md5.Sum([]byte("baz")))
 268
 269         {
 270                 _, _, err := kc.Ask(hash)
 271                 c.Check(err, Equals, keepclient.BlockNotFound)
 272                 log.Print("Ask 1")
 273         }
 274
 275         {
 276                 hash2, rep, err := kc.PutB([]byte("baz"))
 277                 c.Check(hash2, Matches, fmt.Sprintf(`^%s\+3(\+.+)?$`, hash))
 278                 c.Check(rep, Equals, 2)
 279                 c.Check(err, Equals, nil)
 280                 log.Print("PutB")
 281         }
 282
 283         {
 284                 blocklen, _, err := kc.Ask(hash)
 285                 c.Assert(err, Equals, keepclient.BlockNotFound)
 286                 c.Check(blocklen, Equals, int64(0))
 287                 log.Print("Ask 2")
 288         }
 289
 290         {
 291                 _, blocklen, _, err := kc.Get(hash)
 292                 c.Assert(err, Equals, keepclient.BlockNotFound)
 293                 c.Check(blocklen, Equals, int64(0))
 294                 log.Print("Get")
 295         }
 296
 297         log.Print("TestGetDisabled done")
 298 }
 299
 300 func (s *ServerRequiredSuite) TestPutDisabled(c *C) {
 301         log.Print("TestPutDisabled start")
 302
 303         kc := runProxy(c, []string{"keepproxy", "-no-put"}, "4axaw8zxe0qm22wa6urpp5nskcne8z88cvbupv653y1njyi05h", 29953)
 304         waitForListener()
 305         defer closeListener()
 306
 307         {
 308                 hash2, rep, err := kc.PutB([]byte("quux"))
 309                 c.Check(hash2, Equals, "")
 310                 c.Check(rep, Equals, 0)
 311                 c.Check(err, Equals, keepclient.InsufficientReplicasError)
 312                 log.Print("PutB")
 313         }
 314
 315         log.Print("TestPutDisabled done")
 316 }