1// SPDX-License-Identifier: AGPL-3.0-or-later
2
3// Package blobtest provides the shared BlobStore contract suite.
4package blobtest
5
6import (
7 "bytes"
8 "context"
9 "crypto/sha256"
10 "encoding/hex"
11 "errors"
12 "io"
13 "strings"
14 "testing"
15
16 "bnema.dev/ermite/internal/domain"
17 "bnema.dev/ermite/internal/ports"
18 "github.com/stretchr/testify/require"
19)
20
21// Option adapts Run to a provider quirk.
22type Option func(*options)
23
24type options struct{ exactKeyUploadList bool }
25
26// ExactKeyUploadList is for providers (MinIO) that list unfinished multipart
27// uploads only when the prefix is a whole object key: the suite then lists by
28// key instead of by directory prefix.
29func ExactKeyUploadList() Option { return func(o *options) { o.exactKeyUploadList = true } }
30
31// Run checks the blob contract against a fresh store for each subtest.
32func Run(t *testing.T, newStore func(*testing.T) ports.BlobStore, opts ...Option) {
33 ctx := context.Background()
34 var o options
35 for _, opt := range opts {
36 opt(&o)
37 }
38 // uploadPrefix is the ListMultipart prefix for key: dir, or key itself
39 // on a provider that lists by whole key only.
40 uploadPrefix := func(dir, key string) string {
41 if o.exactKeyUploadList {
42 return key
43 }
44 return dir
45 }
46 t.Run("roundtrip", func(t *testing.T) {
47 s := newStore(t)
48 require.NoError(t, s.Check(ctx))
49 data := []byte("abcdefghij")
50 info, e := s.Put(ctx, "ci/roundtrip", bytes.NewReader(data), int64(len(data)))
51 require.NoError(t, e)
52 h := sha256.Sum256(data)
53 require.Equal(t, "sha256:"+hex.EncodeToString(h[:]), info.Digest)
54 info, e = s.Stat(ctx, "ci/roundtrip")
55 require.NoError(t, e)
56 require.EqualValues(t, len(data), info.Size)
57 require.False(t, info.ModTime.IsZero())
58 for _, r := range []struct {
59 off, len int64
60 want string
61 }{{0, -1, "abcdefghij"}, {0, 3, "abc"}, {3, 4, "defg"}, {9, -1, "j"}, {3, 0, ""}} {
62 rc, meta, e := s.Open(ctx, "ci/roundtrip", r.off, r.len)
63 require.NoError(t, e)
64 b, e := io.ReadAll(rc)
65 require.NoError(t, e)
66 require.NoError(t, rc.Close())
67 require.Equal(t, r.want, string(b))
68 require.EqualValues(t, 10, meta.Size)
69 }
70 for _, off := range []int64{10, 11} {
71 _, _, e := s.Open(ctx, "ci/roundtrip", off, -1)
72 require.ErrorIs(t, e, domain.ErrOutOfRange)
73 }
74 _, e = s.Put(ctx, "ci/roundtrip", strings.NewReader("new"), 3)
75 require.NoError(t, e)
76 rc, _, e := s.Open(ctx, "ci/roundtrip", 0, -1)
77 require.NoError(t, e)
78 b, e := io.ReadAll(rc)
79 require.NoError(t, e)
80 _ = rc.Close()
81 require.Equal(t, "new", string(b))
82 require.NoError(t, s.Delete(ctx, "ci/roundtrip"))
83 require.NoError(t, s.Delete(ctx, "ci/roundtrip"))
84 _, e = s.Stat(ctx, "ci/roundtrip")
85 require.ErrorIs(t, e, domain.ErrNotFound)
86 _, _, e = s.Open(ctx, "ci/absent", 0, -1)
87 require.ErrorIs(t, e, domain.ErrNotFound)
88 })
89 t.Run("list_and_reserved_names", func(t *testing.T) {
90 s := newStore(t)
91 keys := []string{"a.parts/b", "a/.tmp-object", "ci/z", "ci/a"}
92 for _, k := range keys {
93 _, e := s.Put(ctx, k, strings.NewReader(k), int64(len(k)))
94 require.NoError(t, e)
95 r, _, e := s.Open(ctx, k, 0, -1)
96 require.NoError(t, e)
97 b, e := io.ReadAll(r)
98 _ = r.Close()
99 require.NoError(t, e)
100 require.Equal(t, k, string(b))
101 }
102 var got []string
103 require.NoError(t, s.List(ctx, "", func(i ports.BlobInfo) error { got = append(got, i.Key); return nil }))
104 require.Equal(t, []string{"a.parts/b", "a/.tmp-object", "ci/a", "ci/z"}, got)
105 got = nil
106 require.NoError(t, s.List(ctx, "ci/", func(i ports.BlobInfo) error { got = append(got, i.Key); return nil }))
107 require.Equal(t, []string{"ci/a", "ci/z"}, got)
108 stop := errors.New("stop")
109 calls := 0
110 require.ErrorIs(t, s.List(ctx, "ci/", func(ports.BlobInfo) error { calls++; return stop }), stop)
111 require.Equal(t, 1, calls)
112 })
113 t.Run("invalid", func(t *testing.T) {
114 s := newStore(t)
115 for _, key := range []string{"", "/a", "a//b", "a/../b", "a/./b", "é", strings.Repeat("a", 513)} {
116 _, e := s.Put(ctx, key, strings.NewReader("x"), 1)
117 require.ErrorIs(t, e, domain.ErrInvalid)
118 _, e = s.Stat(ctx, key)
119 require.ErrorIs(t, e, domain.ErrInvalid)
120 _, _, e = s.Open(ctx, key, 0, -1)
121 require.ErrorIs(t, e, domain.ErrInvalid)
122 require.ErrorIs(t, s.Delete(ctx, key), domain.ErrInvalid)
123 _, e = s.NewMultipart(ctx, key)
124 require.ErrorIs(t, e, domain.ErrInvalid)
125 require.ErrorIs(t, s.AbortMultipart(ctx, key, "id"), domain.ErrInvalid)
126 }
127 _, e := s.Put(ctx, "ci/negative", strings.NewReader(""), -1)
128 require.ErrorIs(t, e, domain.ErrInvalid)
129 })
130 t.Run("multipart", func(t *testing.T) {
131 s := newStore(t)
132 u, e := s.NewMultipart(ctx, "ci/multi")
133 require.NoError(t, e)
134 var all []byte
135 for n := 1; n <= 4; n++ {
136 size := ports.BlobMinPartSize
137 if n == 4 {
138 size = 1024
139 }
140 data := bytes.Repeat([]byte{byte(n)}, size)
141 all = append(all, data...)
142 require.NoError(t, u.UploadPart(ctx, n, bytes.NewReader(data), int64(size)))
143 }
144 info, e := u.Complete(ctx)
145 require.NoError(t, e)
146 require.Empty(t, info.Digest)
147 r, _, e := s.Open(ctx, "ci/multi", 0, -1)
148 require.NoError(t, e)
149 b, e := io.ReadAll(r)
150 _ = r.Close()
151 require.NoError(t, e)
152 require.Equal(t, all, b)
153 })
154 t.Run("abort", func(t *testing.T) {
155 s := newStore(t)
156 for _, byID := range []bool{false, true} {
157 u, e := s.NewMultipart(ctx, "ci/abort")
158 require.NoError(t, e)
159 require.NoError(t, u.UploadPart(ctx, 1, strings.NewReader("x"), 1))
160 var uploads []ports.MultipartInfo
161 require.NoError(t, s.ListMultipart(ctx, uploadPrefix("ci/", "ci/abort"), func(i ports.MultipartInfo) error { uploads = append(uploads, i); return nil }))
162 require.Len(t, uploads, 1)
163 require.Equal(t, u.ID(), uploads[0].ID)
164 require.Equal(t, "ci/abort", uploads[0].Key)
165 require.False(t, uploads[0].Started.IsZero())
166 if byID {
167 require.NoError(t, s.AbortMultipart(ctx, "ci/abort", u.ID()))
168 } else {
169 require.NoError(t, u.Abort(ctx))
170 }
171 _, e = s.Stat(ctx, "ci/abort")
172 require.ErrorIs(t, e, domain.ErrNotFound)
173 uploads = nil
174 require.NoError(t, s.ListMultipart(ctx, uploadPrefix("", "ci/abort"), func(i ports.MultipartInfo) error { uploads = append(uploads, i); return nil }))
175 require.Empty(t, uploads)
176 }
177 })
178 t.Run("undersized_part", func(t *testing.T) {
179 s := newStore(t)
180 u, e := s.NewMultipart(ctx, "ci/small")
181 require.NoError(t, e)
182 require.NoError(t, u.UploadPart(ctx, 1, strings.NewReader("x"), 1))
183 require.NoError(t, u.UploadPart(ctx, 2, strings.NewReader("y"), 1))
184 _, e = u.Complete(ctx)
185 require.ErrorIs(t, e, domain.ErrInvalid)
186 _, e = s.Stat(ctx, "ci/small")
187 require.ErrorIs(t, e, domain.ErrNotFound)
188 })
189 t.Run("failed_put", func(t *testing.T) {
190 s := newStore(t)
191 _, e := s.Put(ctx, "ci/fail", io.MultiReader(strings.NewReader("abc"), failReader{}), 10)
192 require.Error(t, e)
193 _, e = s.Stat(ctx, "ci/fail")
194 require.ErrorIs(t, e, domain.ErrNotFound)
195 for _, v := range []struct {
196 data string
197 size int64
198 }{{"a", 2}, {"abc", 2}} {
199 _, e = s.Put(ctx, "ci/size", strings.NewReader(v.data), v.size)
200 require.ErrorIs(t, e, domain.ErrInvalid)
201 _, e = s.Stat(ctx, "ci/size")
202 require.ErrorIs(t, e, domain.ErrNotFound)
203 }
204 })
205 t.Run("large_put", func(t *testing.T) {
206 s := newStore(t)
207 data := bytes.Repeat([]byte("large"), ports.BlobPartSize/5+100)
208 info, e := s.Put(ctx, "ci/large", bytes.NewBuffer(data), int64(len(data)))
209 require.NoError(t, e)
210 h := sha256.Sum256(data)
211 require.Equal(t, "sha256:"+hex.EncodeToString(h[:]), info.Digest)
212 r, _, e := s.Open(ctx, "ci/large", 0, -1)
213 require.NoError(t, e)
214 got, e := io.ReadAll(r)
215 require.NoError(t, e)
216 require.NoError(t, r.Close())
217 require.Equal(t, data, got)
218 for _, size := range []int64{int64(len(data) - 1), int64(len(data) + 1)} {
219 _, e = s.Put(ctx, "ci/badlarge", bytes.NewBuffer(data), size)
220 require.ErrorIs(t, e, domain.ErrInvalid)
221 _, e = s.Stat(ctx, "ci/badlarge")
222 require.ErrorIs(t, e, domain.ErrNotFound)
223 }
224 require.NoError(t, s.ListMultipart(ctx, uploadPrefix("", "ci/badlarge"), func(ports.MultipartInfo) error { t.Error("unfinished upload after Put"); return nil }))
225 })
226
227 t.Run("cancel_put", func(t *testing.T) {
228 s := newStore(t)
229 c, cancel := context.WithCancel(ctx)
230 defer cancel()
231 r := &cancelReader{cancel: cancel}
232 _, e := s.Put(c, "ci/cancel", r, 1024)
233 require.ErrorIs(t, e, context.Canceled)
234 _, e = s.Stat(ctx, "ci/cancel")
235 require.ErrorIs(t, e, domain.ErrNotFound)
236 })
237}
238
239type failReader struct{}
240
241func (failReader) Read([]byte) (int, error) { return 0, errors.New("reader failed") }
242
243type cancelReader struct {
244 cancel context.CancelFunc
245 done bool
246}
247
248func (r *cancelReader) Read(p []byte) (int, error) {
249 if r.done {
250 return 0, context.Canceled
251 }
252 r.done = true
253 r.cancel()
254 p[0] = 'x'
255 return 1, nil
256}