1 // Copyright (C) The Arvados Authors. All rights reserved.
3 // SPDX-License-Identifier: Apache-2.0
18 "git.curoverse.com/arvados.git/sdk/go/arvadosclient"
19 "git.curoverse.com/arvados.git/sdk/go/arvadostest"
20 check "gopkg.in/check.v1"
23 var _ = check.Suite(&CollectionReaderUnit{})
25 type CollectionReaderUnit struct {
26 arv *arvadosclient.ArvadosClient
28 handler SuccessHandler
31 func (s *CollectionReaderUnit) SetUpTest(c *check.C) {
33 s.arv, err = arvadosclient.MakeArvadosClient()
34 c.Assert(err, check.IsNil)
35 s.arv.ApiToken = arvadostest.ActiveToken
37 s.kc, err = MakeKeepClient(s.arv)
38 c.Assert(err, check.IsNil)
40 s.handler = SuccessHandler{
41 disk: make(map[string][]byte),
42 lock: make(chan struct{}, 1),
45 localRoots := make(map[string]string)
46 for i, k := range RunSomeFakeKeepServers(s.handler, 4) {
47 localRoots[fmt.Sprintf("zzzzz-bi6l4-fakefakefake%03d", i)] = k.url
49 s.kc.SetServiceRoots(localRoots, localRoots, nil)
52 type SuccessHandler struct {
53 disk map[string][]byte
54 lock chan struct{} // channel with buffer==1: full when an operation is in progress.
55 ops *int // number of operations completed
58 func (h SuccessHandler) ServeHTTP(resp http.ResponseWriter, req *http.Request) {
61 buf, err := ioutil.ReadAll(req.Body)
66 pdh := fmt.Sprintf("%x+%d", md5.Sum(buf), len(buf))
73 resp.Write([]byte(pdh))
75 pdh := req.URL.Path[1:]
77 buf, ok := h.disk[pdh]
83 resp.WriteHeader(http.StatusNotFound)
88 resp.WriteHeader(http.StatusMethodNotAllowed)
93 mt string // manifest text
95 want interface{} // error or string to expect
98 func (s *CollectionReaderUnit) TestCollectionReaderContent(c *check.C) {
99 s.kc.PutB([]byte("foo"))
100 s.kc.PutB([]byte("bar"))
101 s.kc.PutB([]byte("Hello world\n"))
102 s.kc.PutB([]byte(""))
104 mt := arvadostest.PathologicalManifest
106 for _, testCase := range []rdrTest{
107 {mt: mt, f: "zzzz", want: os.ErrNotExist},
108 {mt: mt, f: "frob", want: os.ErrNotExist},
109 {mt: mt, f: "/segmented/frob", want: "frob"},
110 {mt: mt, f: "./segmented/frob", want: "frob"},
111 {mt: mt, f: "/f", want: "f"},
112 {mt: mt, f: "./f", want: "f"},
113 {mt: mt, f: "foo bar//baz", want: "foo"},
114 {mt: mt, f: "foo/zero", want: ""},
115 {mt: mt, f: "zero@0", want: ""},
116 {mt: mt, f: "zero@1", want: ""},
117 {mt: mt, f: "zero@4", want: ""},
118 {mt: mt, f: "zero@9", want: ""},
119 {mt: mt, f: "f", want: "f"},
120 {mt: mt, f: "ooba", want: "ooba"},
121 {mt: mt, f: "overlapReverse/o", want: "o"},
122 {mt: mt, f: "overlapReverse/oo", want: "oo"},
123 {mt: mt, f: "overlapReverse/ofoo", want: "ofoo"},
124 {mt: mt, f: "foo bar/baz", want: "foo"},
125 {mt: mt, f: "segmented/frob", want: "frob"},
126 {mt: mt, f: "segmented/oof", want: "oof"},
128 c.Logf("%#v", testCase)
129 rdr, err := s.kc.CollectionFileReader(map[string]interface{}{"manifest_text": testCase.mt}, testCase.f)
130 switch want := testCase.want.(type) {
132 c.Check(rdr, check.IsNil)
133 c.Check(err, check.Equals, want)
135 buf := make([]byte, len(want))
136 n, err := io.ReadFull(rdr, buf)
137 c.Check(err, check.IsNil)
138 for i := 0; i < 4; i++ {
139 c.Check(string(buf), check.Equals, want)
140 n, err = rdr.Read(buf)
141 c.Check(n, check.Equals, 0)
142 c.Check(err, check.Equals, io.EOF)
145 for a := len(want) - 2; a >= 0; a-- {
146 for b := a + 1; b <= len(want); b++ {
147 offset, err := rdr.Seek(int64(a), io.SeekStart)
148 c.Logf("...a=%d, b=%d", a, b)
149 c.Check(err, check.IsNil)
150 c.Check(offset, check.Equals, int64(a))
151 buf := make([]byte, b-a)
152 n, err := io.ReadFull(rdr, buf)
153 c.Check(n, check.Equals, b-a)
154 c.Check(string(buf), check.Equals, want[a:b])
157 offset, err := rdr.Seek(-1, io.SeekStart)
158 c.Check(err, check.NotNil)
159 c.Check(offset, check.Equals, int64(len(want)))
161 c.Check(rdr.Close(), check.Equals, nil)
166 func (s *CollectionReaderUnit) TestCollectionReaderManyBlocks(c *check.C) {
169 buf := make([]byte, 4096)
170 locs := make([]string, len(buf))
172 for i := 0; i < len(locs); i++ {
173 _, err := rand.Read(buf[:i])
175 locs[i], _, err = s.kc.PutB(buf[:i])
176 c.Assert(err, check.IsNil)
178 testdata = append(testdata, buf[:i]...)
180 manifest := "./random " + strings.Join(locs, " ") + " 0:" + strconv.Itoa(filesize) + ":bytes.bin\n"
181 dataMD5 := h.Sum(nil)
183 checkMD5 := md5.New()
184 rdr, err := s.kc.CollectionFileReader(map[string]interface{}{"manifest_text": manifest}, "random/bytes.bin")
185 c.Assert(err, check.IsNil)
188 _, err = io.Copy(checkMD5, rdr)
189 c.Check(err, check.IsNil)
190 _, err = rdr.Read(make([]byte, 1))
191 c.Check(err, check.Equals, io.EOF)
192 c.Check(checkMD5.Sum(nil), check.DeepEquals, dataMD5)
194 size, err := rdr.Seek(0, io.SeekEnd)
195 c.Check(err, check.IsNil)
196 buf = make([]byte, len(testdata))
199 for i := 0; i < 16; i++ {
200 offset := rand.Intn(len(buf) - 1)
201 count := rand.Intn(len(buf) - offset)
202 if rand.Intn(2) == 0 {
203 curPos, err = rdr.Seek(int64(offset)-curPos, io.SeekCurrent)
205 curPos, err = rdr.Seek(int64(offset), io.SeekStart)
207 c.Check(curPos, check.Equals, int64(offset))
209 n, err := rdr.Read(buf[offset : offset+count])
210 c.Assert(err, check.IsNil)
211 c.Assert(n > 0, check.Equals, true)
215 curPos, err = rdr.Seek(0, io.SeekCurrent)
216 c.Check(curPos, check.Equals, int64(offset))
218 c.Check(md5.Sum(buf), check.DeepEquals, md5.Sum(testdata))
219 c.Check(buf[:1000], check.DeepEquals, testdata[:1000])
221 curPos, err = rdr.Seek(size+12345, io.SeekCurrent)
222 c.Check(err, check.IsNil)
223 c.Check(curPos, check.Equals, size)
225 curPos, err = rdr.Seek(8-size, io.SeekCurrent)
226 c.Check(err, check.IsNil)
227 c.Check(curPos, check.Equals, int64(8))
229 curPos, err = rdr.Seek(-9, io.SeekCurrent)
230 c.Check(err, check.NotNil)
231 c.Check(curPos, check.Equals, int64(8))
234 func (s *CollectionReaderUnit) TestCollectionReaderCloseEarly(c *check.C) {
235 s.kc.BlockCache = &BlockCache{}
236 s.kc.PutB([]byte("foo"))
237 s.kc.PutB([]byte("bar"))
238 s.kc.PutB([]byte("baz"))
241 for i := 0; i < 300; i++ {
242 mt += "acbd18db4cc2f85cedef654fccc4a4d8+3 37b51d194a7513e45b56f6524f2d51f2+3 73feffa4b7f6bb68e44cf984c85f6e88+3 "
244 mt += "0:2700:foo900.txt\n"
246 // Grab the stub server's lock, ensuring our cfReader doesn't
247 // get anything back from its first call to kc.Get() before we
248 // have a chance to call Close().
249 s.handler.lock <- struct{}{}
250 opsBeforeRead := *s.handler.ops
252 rdr, err := s.kc.CollectionFileReader(map[string]interface{}{"manifest_text": mt}, "foo900.txt")
253 c.Assert(err, check.IsNil)
255 firstReadDone := make(chan struct{})
257 n, err := rdr.Read(make([]byte, 3))
258 c.Check(n, check.Equals, 3)
259 c.Check(err, check.IsNil)
263 // Release the stub server's lock. The first GET operation will proceed.
266 // Make sure our first read operation consumes the data
267 // received from the first GET.
271 c.Check(err, check.IsNil)
273 // Stub should have handled exactly one GET request.
274 c.Check(*s.handler.ops, check.Equals, opsBeforeRead+1)
277 func (s *CollectionReaderUnit) TestCollectionReaderDataError(c *check.C) {
278 manifest := ". ffffffffffffffffffffffffffffffff+1 0:1:notfound.txt\n"
279 buf := make([]byte, 1)
280 rdr, err := s.kc.CollectionFileReader(map[string]interface{}{"manifest_text": manifest}, "notfound.txt")
281 c.Check(err, check.IsNil)
282 for i := 0; i < 2; i++ {
283 _, err = io.ReadFull(rdr, buf)
284 c.Check(err, check.NotNil)
285 c.Check(err, check.Not(check.Equals), io.EOF)
287 c.Check(rdr.Close(), check.IsNil)