sdk/go/keepclient/collectionreader_test.go

   1 package keepclient
   2
   3 import (
   4         "crypto/md5"
   5         "crypto/rand"
   6         "fmt"
   7         "io"
   8         "io/ioutil"
   9         "net/http"
  10         "os"
  11         "strconv"
  12         "strings"
  13
  14         "git.curoverse.com/arvados.git/sdk/go/arvadosclient"
  15         "git.curoverse.com/arvados.git/sdk/go/arvadostest"
  16         check "gopkg.in/check.v1"
  17 )
  18
  19 var _ = check.Suite(&CollectionReaderUnit{})
  20
  21 type CollectionReaderUnit struct {
  22         arv     *arvadosclient.ArvadosClient
  23         kc      *KeepClient
  24         handler SuccessHandler
  25 }
  26
  27 func (s *CollectionReaderUnit) SetUpTest(c *check.C) {
  28         var err error
  29         s.arv, err = arvadosclient.MakeArvadosClient()
  30         c.Assert(err, check.IsNil)
  31         s.arv.ApiToken = arvadostest.ActiveToken
  32
  33         s.kc, err = MakeKeepClient(s.arv)
  34         c.Assert(err, check.IsNil)
  35
  36         s.handler = SuccessHandler{
  37                 disk: make(map[string][]byte),
  38                 lock: make(chan struct{}, 1),
  39                 ops:  new(int),
  40         }
  41         localRoots := make(map[string]string)
  42         for i, k := range RunSomeFakeKeepServers(s.handler, 4) {
  43                 localRoots[fmt.Sprintf("zzzzz-bi6l4-fakefakefake%03d", i)] = k.url
  44         }
  45         s.kc.SetServiceRoots(localRoots, localRoots, nil)
  46 }
  47
  48 type SuccessHandler struct {
  49         disk map[string][]byte
  50         lock chan struct{} // channel with buffer==1: full when an operation is in progress.
  51         ops  *int          // number of operations completed
  52 }
  53
  54 func (h SuccessHandler) ServeHTTP(resp http.ResponseWriter, req *http.Request) {
  55         switch req.Method {
  56         case "PUT":
  57                 buf, err := ioutil.ReadAll(req.Body)
  58                 if err != nil {
  59                         resp.WriteHeader(500)
  60                         return
  61                 }
  62                 pdh := fmt.Sprintf("%x+%d", md5.Sum(buf), len(buf))
  63                 h.lock <- struct{}{}
  64                 h.disk[pdh] = buf
  65                 if h.ops != nil {
  66                         (*h.ops)++
  67                 }
  68                 <-h.lock
  69                 resp.Write([]byte(pdh))
  70         case "GET":
  71                 pdh := req.URL.Path[1:]
  72                 h.lock <- struct{}{}
  73                 buf, ok := h.disk[pdh]
  74                 if h.ops != nil {
  75                         (*h.ops)++
  76                 }
  77                 <-h.lock
  78                 if !ok {
  79                         resp.WriteHeader(http.StatusNotFound)
  80                 } else {
  81                         resp.Write(buf)
  82                 }
  83         default:
  84                 resp.WriteHeader(http.StatusMethodNotAllowed)
  85         }
  86 }
  87
  88 type rdrTest struct {
  89         mt   string      // manifest text
  90         f    string      // filename
  91         want interface{} // error or string to expect
  92 }
  93
  94 func (s *CollectionReaderUnit) TestCollectionReaderContent(c *check.C) {
  95         s.kc.PutB([]byte("foo"))
  96         s.kc.PutB([]byte("bar"))
  97         s.kc.PutB([]byte("Hello world\n"))
  98         s.kc.PutB([]byte(""))
  99
 100         mt := arvadostest.PathologicalManifest
 101
 102         for _, testCase := range []rdrTest{
 103                 {mt: mt, f: "zzzz", want: os.ErrNotExist},
 104                 {mt: mt, f: "frob", want: os.ErrNotExist},
 105                 {mt: mt, f: "/segmented/frob", want: os.ErrNotExist},
 106                 {mt: mt, f: "./segmented/frob", want: os.ErrNotExist},
 107                 {mt: mt, f: "/f", want: os.ErrNotExist},
 108                 {mt: mt, f: "./f", want: os.ErrNotExist},
 109                 {mt: mt, f: "foo bar//baz", want: os.ErrNotExist},
 110                 {mt: mt, f: "foo/zero", want: ""},
 111                 {mt: mt, f: "zero@0", want: ""},
 112                 {mt: mt, f: "zero@1", want: ""},
 113                 {mt: mt, f: "zero@4", want: ""},
 114                 {mt: mt, f: "zero@9", want: ""},
 115                 {mt: mt, f: "f", want: "f"},
 116                 {mt: mt, f: "ooba", want: "ooba"},
 117                 {mt: mt, f: "overlapReverse/o", want: "o"},
 118                 {mt: mt, f: "overlapReverse/oo", want: "oo"},
 119                 {mt: mt, f: "overlapReverse/ofoo", want: "ofoo"},
 120                 {mt: mt, f: "foo bar/baz", want: "foo"},
 121                 {mt: mt, f: "segmented/frob", want: "frob"},
 122                 {mt: mt, f: "segmented/oof", want: "oof"},
 123         } {
 124                 c.Logf("%#v", testCase)
 125                 rdr, err := s.kc.CollectionFileReader(map[string]interface{}{"manifest_text": testCase.mt}, testCase.f)
 126                 switch want := testCase.want.(type) {
 127                 case error:
 128                         c.Check(rdr, check.IsNil)
 129                         c.Check(err, check.Equals, want)
 130                 case string:
 131                         buf := make([]byte, len(want))
 132                         n, err := io.ReadFull(rdr, buf)
 133                         c.Check(err, check.IsNil)
 134                         for i := 0; i < 4; i++ {
 135                                 c.Check(string(buf), check.Equals, want)
 136                                 n, err = rdr.Read(buf)
 137                                 c.Check(n, check.Equals, 0)
 138                                 c.Check(err, check.Equals, io.EOF)
 139                         }
 140
 141                         for a := len(want) - 2; a >= 0; a-- {
 142                                 for b := a + 1; b <= len(want); b++ {
 143                                         offset, err := rdr.Seek(int64(a), io.SeekStart)
 144                                         c.Logf("...a=%d, b=%d", a, b)
 145                                         c.Check(err, check.IsNil)
 146                                         c.Check(offset, check.Equals, int64(a))
 147                                         buf := make([]byte, b-a)
 148                                         n, err := io.ReadFull(rdr, buf)
 149                                         c.Check(n, check.Equals, b-a)
 150                                         c.Check(string(buf), check.Equals, want[a:b])
 151                                 }
 152                         }
 153                         offset, err := rdr.Seek(-1, io.SeekStart)
 154                         c.Check(err, check.NotNil)
 155                         c.Check(offset, check.Equals, int64(len(want)))
 156
 157                         c.Check(rdr.Close(), check.Equals, nil)
 158                 }
 159         }
 160 }
 161
 162 func (s *CollectionReaderUnit) TestCollectionReaderManyBlocks(c *check.C) {
 163         h := md5.New()
 164         buf := make([]byte, 4096)
 165         locs := make([]string, len(buf))
 166         filesize := 0
 167         for i := 0; i < len(locs); i++ {
 168                 _, err := io.ReadFull(rand.Reader, buf[:i])
 169                 c.Assert(err, check.IsNil)
 170                 h.Write(buf[:i])
 171                 locs[i], _, err = s.kc.PutB(buf[:i])
 172                 c.Assert(err, check.IsNil)
 173                 filesize += i
 174         }
 175         manifest := "./random " + strings.Join(locs, " ") + " 0:" + strconv.Itoa(filesize) + ":bytes.bin\n"
 176         dataMD5 := h.Sum(nil)
 177
 178         checkMD5 := md5.New()
 179         rdr, err := s.kc.CollectionFileReader(map[string]interface{}{"manifest_text": manifest}, "random/bytes.bin")
 180         c.Check(err, check.IsNil)
 181         _, err = io.Copy(checkMD5, rdr)
 182         c.Check(err, check.IsNil)
 183         _, err = rdr.Read(make([]byte, 1))
 184         c.Check(err, check.Equals, io.EOF)
 185         c.Check(checkMD5.Sum(nil), check.DeepEquals, dataMD5)
 186 }
 187
 188 func (s *CollectionReaderUnit) TestCollectionReaderCloseEarly(c *check.C) {
 189         s.kc.BlockCache = &BlockCache{}
 190         s.kc.PutB([]byte("foo"))
 191         s.kc.PutB([]byte("bar"))
 192         s.kc.PutB([]byte("baz"))
 193
 194         mt := ". "
 195         for i := 0; i < 300; i++ {
 196                 mt += "acbd18db4cc2f85cedef654fccc4a4d8+3 37b51d194a7513e45b56f6524f2d51f2+3 73feffa4b7f6bb68e44cf984c85f6e88+3 "
 197         }
 198         mt += "0:2700:foo900.txt\n"
 199
 200         // Grab the stub server's lock, ensuring our cfReader doesn't
 201         // get anything back from its first call to kc.Get() before we
 202         // have a chance to call Close().
 203         s.handler.lock <- struct{}{}
 204         opsBeforeRead := *s.handler.ops
 205
 206         rdr, err := s.kc.CollectionFileReader(map[string]interface{}{"manifest_text": mt}, "foo900.txt")
 207         c.Assert(err, check.IsNil)
 208
 209         firstReadDone := make(chan struct{})
 210         go func() {
 211                 n, err := rdr.Read(make([]byte, 3))
 212                 c.Check(n, check.Equals, 3)
 213                 c.Check(err, check.IsNil)
 214                 close(firstReadDone)
 215         }()
 216
 217         // Release the stub server's lock. The first GET operation will proceed.
 218         <-s.handler.lock
 219
 220         // Make sure our first read operation consumes the data
 221         // received from the first GET.
 222         <-firstReadDone
 223
 224         err = rdr.Close()
 225         c.Check(err, check.IsNil)
 226
 227         // Stub should have handled exactly one GET request.
 228         c.Check(*s.handler.ops, check.Equals, opsBeforeRead+1)
 229 }
 230
 231 func (s *CollectionReaderUnit) TestCollectionReaderDataError(c *check.C) {
 232         manifest := ". ffffffffffffffffffffffffffffffff+1 0:1:notfound.txt\n"
 233         buf := make([]byte, 1)
 234         rdr, err := s.kc.CollectionFileReader(map[string]interface{}{"manifest_text": manifest}, "notfound.txt")
 235         c.Check(err, check.IsNil)
 236         for i := 0; i < 2; i++ {
 237                 _, err = io.ReadFull(rdr, buf)
 238                 c.Check(err, check.NotNil)
 239                 c.Check(err, check.Not(check.Equals), io.EOF)
 240         }
 241         c.Check(rdr.Close(), check.IsNil)
 242 }