1// SPDX-License-Identifier: AGPL-3.0-or-later
2
3package usecase
4
5import (
6 "bytes"
7 "context"
8 "crypto/sha256"
9 "encoding/hex"
10 "errors"
11 "io"
12 "strconv"
13 "strings"
14 "testing"
15 "time"
16
17 "github.com/stretchr/testify/assert"
18 "github.com/stretchr/testify/mock"
19 "github.com/stretchr/testify/require"
20
21 "bnema.dev/ermite/internal/domain"
22 "bnema.dev/ermite/internal/ports"
23 "bnema.dev/ermite/internal/ports/mocks"
24)
25
26// artifactBlobMock uses generated port mocks with byte storage in callbacks.
27// Real-disk end-to-end behavior is exercised by internal/app's artifact rig.
28func artifactBlobMock(t *testing.T, failPart bool) (*mocks.BlobStore, *int, *int) {
29 t.Helper()
30 b := mocks.NewBlobStore(t)
31 objects := map[string][]byte{}
32 active := map[string]ports.MultipartInfo{}
33 puts, multiparts := new(int), new(int)
34 b.EXPECT().Put(mock.Anything, mock.Anything, mock.Anything, mock.Anything).RunAndReturn(func(_ context.Context, key string, r io.Reader, size int64) (ports.BlobInfo, error) {
35 *puts++
36 data, err := io.ReadAll(r)
37 if err != nil {
38 return ports.BlobInfo{}, err
39 }
40 if int64(len(data)) != size {
41 return ports.BlobInfo{}, domain.ErrInvalid
42 }
43 objects[key] = data
44 return ports.BlobInfo{Key: key, Size: size}, nil
45 }).Maybe()
46 b.EXPECT().Open(mock.Anything, mock.Anything, mock.Anything, mock.Anything).RunAndReturn(func(_ context.Context, key string, offset, length int64) (io.ReadCloser, ports.BlobInfo, error) {
47 data, ok := objects[key]
48 if !ok {
49 return nil, ports.BlobInfo{}, domain.ErrNotFound
50 }
51 info := ports.BlobInfo{Key: key, Size: int64(len(data))}
52 if offset >= info.Size {
53 return nil, info, domain.ErrOutOfRange
54 }
55 end := info.Size
56 if length >= 0 {
57 end = min(end, offset+length)
58 }
59 return io.NopCloser(bytes.NewReader(data[offset:end])), info, nil
60 }).Maybe()
61 b.EXPECT().Stat(mock.Anything, mock.Anything).RunAndReturn(func(_ context.Context, key string) (ports.BlobInfo, error) {
62 data, ok := objects[key]
63 if !ok {
64 return ports.BlobInfo{}, domain.ErrNotFound
65 }
66 return ports.BlobInfo{Key: key, Size: int64(len(data))}, nil
67 }).Maybe()
68 b.EXPECT().Delete(mock.Anything, mock.Anything).RunAndReturn(func(_ context.Context, key string) error { delete(objects, key); return nil }).Maybe()
69 b.EXPECT().List(mock.Anything, mock.Anything, mock.Anything).RunAndReturn(func(_ context.Context, prefix string, fn func(ports.BlobInfo) error) error {
70 for key, data := range objects {
71 if strings.HasPrefix(key, prefix) {
72 if err := fn(ports.BlobInfo{Key: key, Size: int64(len(data))}); err != nil {
73 return err
74 }
75 }
76 }
77 return nil
78 }).Maybe()
79 b.EXPECT().ListMultipart(mock.Anything, mock.Anything, mock.Anything).RunAndReturn(func(_ context.Context, prefix string, fn func(ports.MultipartInfo) error) error {
80 for key, m := range active {
81 if strings.HasPrefix(key, prefix) {
82 if err := fn(m); err != nil {
83 return err
84 }
85 }
86 }
87 return nil
88 }).Maybe()
89 b.EXPECT().AbortMultipart(mock.Anything, mock.Anything, mock.Anything).RunAndReturn(func(_ context.Context, key, id string) error { delete(active, key); return nil }).Maybe()
90 b.EXPECT().NewMultipart(mock.Anything, mock.Anything).RunAndReturn(func(_ context.Context, key string) (ports.MultipartUpload, error) {
91 *multiparts++
92 active[key] = ports.MultipartInfo{Key: key, ID: "upload"}
93 u := mocks.NewMultipartUpload(t)
94 var data []byte
95 u.EXPECT().UploadPart(mock.Anything, mock.Anything, mock.Anything, mock.Anything).RunAndReturn(func(_ context.Context, _ int, r io.Reader, size int64) error {
96 if failPart {
97 return errors.New("part failed")
98 }
99 part, err := io.ReadAll(r)
100 data = append(data, part...)
101 return err
102 }).Maybe()
103 u.EXPECT().Complete(mock.Anything).RunAndReturn(func(context.Context) (ports.BlobInfo, error) {
104 objects[key] = data
105 delete(active, key)
106 return ports.BlobInfo{Key: key, Size: int64(len(data))}, nil
107 }).Maybe()
108 u.EXPECT().Abort(mock.Anything).RunAndReturn(func(context.Context) error { delete(active, key); return nil }).Maybe()
109 return u, nil
110 }).Maybe()
111 return b, puts, multiparts
112}
113
114func TestArtifactBlobAssembly(t *testing.T) {
115 for _, size := range []int{0, 1, ports.BlobPartSize, ports.BlobPartSize + 1, 2*ports.BlobPartSize + 17} {
116 t.Run(strconv.Itoa(size), func(t *testing.T) {
117 ctx := t.Context()
118 b, puts, multiparts := artifactBlobMock(t, false)
119 c := &CI{cfg: CIConfig{Blobs: b}}
120 a := domain.CIArtifact{RunID: 1, ID: 2, Store: domain.CIArtifactStoreBlob}
121 data := bytes.Repeat([]byte("x"), size)
122 // Parts straddle block boundaries, with an empty block in the list.
123 cut := min(size, 3<<20)
124 _, err := b.Put(ctx, a.BlockKey("a"), bytes.NewReader(data[:cut]), int64(cut))
125 require.NoError(t, err)
126 _, err = b.Put(ctx, a.BlockKey("b"), bytes.NewReader(data[cut:]), int64(size-cut))
127 require.NoError(t, err)
128 _, err = b.Put(ctx, a.BlockKey("empty"), bytes.NewReader(nil), 0)
129 require.NoError(t, err)
130 *puts = 0
131 n, digest, err := c.assembleBlobs(ctx, a, []string{"a", "empty", "b"}, int64(size))
132 require.NoError(t, err)
133 assert.EqualValues(t, size, n)
134 sum := sha256.Sum256(data)
135 assert.Equal(t, "sha256:"+hex.EncodeToString(sum[:]), digest)
136 if size <= ports.BlobPartSize {
137 assert.Equal(t, 1, *puts)
138 assert.Zero(t, *multiparts)
139 } else {
140 assert.Zero(t, *puts)
141 assert.Equal(t, 1, *multiparts)
142 }
143 if size > 0 {
144 r, _, err := b.Open(ctx, a.BlobKey(), 0, -1)
145 require.NoError(t, err)
146 got, err := io.ReadAll(r)
147 require.NoError(t, err)
148 require.NoError(t, r.Close())
149 assert.Equal(t, data, got)
150 } else {
151 info, err := b.Stat(ctx, a.BlobKey())
152 require.NoError(t, err)
153 assert.Zero(t, info.Size)
154 }
155 // No legacy reader is needed for a blob download, including EOF resumes.
156 a.Size = int64(size)
157 var got []byte
158 require.NoError(t, c.streamArtifact(ctx, a, 0, func(ch ports.ArtifactChunk) error { got = append(got, ch.Data...); return nil }))
159 assert.Equal(t, data, append([]byte{}, got...))
160 require.NoError(t, c.streamArtifact(ctx, a, a.Size, func(ports.ArtifactChunk) error { return nil }))
161 c.removeArtifactFiles(ctx, []domain.CIArtifact{a})
162 _, err = b.Stat(ctx, a.BlobKey())
163 assert.ErrorIs(t, err, domain.ErrNotFound)
164 require.NoError(t, b.List(ctx, a.BlobKey()+".blocks/", func(ports.BlobInfo) error { t.Error("block survived deletion"); return nil }))
165 })
166 }
167}
168
169func TestArtifactBlobAssemblyFailureAborts(t *testing.T) {
170 for _, failPart := range []bool{false, true} {
171 t.Run(map[bool]string{false: "missing block", true: "failed part"}[failPart], func(t *testing.T) {
172 b, _, multiparts := artifactBlobMock(t, failPart)
173 c := &CI{cfg: CIConfig{Blobs: b}}
174 a := domain.CIArtifact{RunID: 1, ID: 2}
175 data := bytes.Repeat([]byte("a"), ports.BlobPartSize)
176 _, err := b.Put(t.Context(), a.BlockKey("a"), bytes.NewReader(data), int64(len(data)))
177 require.NoError(t, err)
178 _, _, err = c.assembleBlobs(t.Context(), a, []string{"a", "missing"}, ports.BlobPartSize+1)
179 require.Error(t, err)
180 assert.Equal(t, 1, *multiparts)
181 _, err = b.Stat(t.Context(), a.BlobKey())
182 assert.ErrorIs(t, err, domain.ErrNotFound)
183 require.NoError(t, b.ListMultipart(t.Context(), "ci/", func(ports.MultipartInfo) error { t.Error("multipart survived failure"); return nil }))
184 })
185 }
186}
187
188func TestArtifactPutUnsizedBound(t *testing.T) {
189 b, _, _ := artifactBlobMock(t, false)
190 c := &CI{cfg: CIConfig{Blobs: b}}
191 for _, size := range []int{0, 1, domain.CIArtifactMaxBlockSize, domain.CIArtifactMaxBlockSize + 1} {
192 n, err := c.putUnsized(t.Context(), "ci/1/2", bytes.NewReader(make([]byte, size)))
193 if size > domain.CIArtifactMaxBlockSize {
194 assert.ErrorIs(t, err, domain.ErrInvalid)
195 assert.Contains(t, err.Error(), "a block has at most 8 MiB")
196 } else {
197 require.NoError(t, err)
198 assert.EqualValues(t, size, n)
199 }
200 }
201 ctx, cancel := context.WithCancel(t.Context())
202 cancel()
203 _, err := c.putUnsized(ctx, "ci/1/cancelled", bytes.NewReader([]byte("x")))
204 assert.ErrorIs(t, err, context.Canceled)
205 _, err = b.Stat(t.Context(), "ci/1/cancelled")
206 assert.ErrorIs(t, err, domain.ErrNotFound)
207}
208
209func TestArtifactLegacyStream(t *testing.T) {
210 a := domain.CIArtifact{RunID: 1, ID: 2, Size: 6}
211 legacy := mocks.NewCIArtifactFiles(t)
212 legacy.EXPECT().Open(mock.Anything, artifactRef(a), int64(2)).RunAndReturn(func(context.Context, ports.CIArtifactRef, int64) (io.ReadCloser, error) {
213 return io.NopCloser(bytes.NewReader([]byte("gacy"))), nil
214 }).Twice()
215 c := &CI{cfg: CIConfig{ArtifactFiles: legacy}}
216 for _, store := range []domain.CIArtifactStore{domain.CIArtifactStoreLegacy, "disk"} {
217 a.Store = store
218 var got []byte
219 require.NoError(t, c.streamArtifact(t.Context(), a, 2, func(ch ports.ArtifactChunk) error { got = append(got, ch.Data...); return nil }))
220 assert.Equal(t, "gacy", string(got))
221 }
222 c.cfg.ArtifactFiles = nil
223 assert.ErrorIs(t, c.streamArtifact(t.Context(), a, 0, func(ports.ArtifactChunk) error { return nil }), domain.ErrNotFound)
224}
225
226func TestArtifactBlobOrphanSweep(t *testing.T) {
227 ctx := t.Context()
228 b := mocks.NewBlobStore(t)
229 s := mocks.NewCIArtifactStore(t)
230 c := &CI{cfg: CIConfig{Blobs: b, Artifacts: s, Clock: stillClock{}}}
231 now := c.now()
232 old := now.Add(-25 * time.Hour)
233 pending := domain.CIArtifact{RunID: 1, ID: 2, Store: domain.CIArtifactStoreBlob, State: domain.CIArtifactPending}
234 committed := pending
235 committed.ID = 3
236 committed.State = domain.CIArtifactCommitted
237 missing := pending
238 missing.ID = 4
239 b.EXPECT().List(mock.Anything, "ci/", mock.Anything).RunAndReturn(func(_ context.Context, _ string, fn func(ports.BlobInfo) error) error {
240 for _, info := range []ports.BlobInfo{
241 {Key: missing.BlobKey(), ModTime: now}, // young: no lookup, no deletion
242 {Key: missing.BlobKey(), ModTime: old},
243 {Key: pending.BlockKey("a"), ModTime: old},
244 {Key: committed.BlockKey("a"), ModTime: old},
245 {Key: committed.BlobKey(), ModTime: old},
246 } {
247 require.NoError(t, fn(info))
248 }
249 return nil
250 })
251 s.EXPECT().GetArtifact(mock.Anything, missing.ID).Return(domain.CIArtifact{}, domain.ErrNotFound).Once()
252 s.EXPECT().GetArtifact(mock.Anything, pending.ID).Return(pending, nil).Once()
253 s.EXPECT().GetArtifact(mock.Anything, committed.ID).Return(committed, nil).Twice()
254 b.EXPECT().Delete(mock.Anything, missing.BlobKey()).Return(nil).Once()
255 b.EXPECT().Delete(mock.Anything, committed.BlockKey("a")).Return(nil).Once()
256 b.EXPECT().ListMultipart(mock.Anything, "ci/", mock.Anything).RunAndReturn(func(_ context.Context, _ string, fn func(ports.MultipartInfo) error) error {
257 require.NoError(t, fn(ports.MultipartInfo{Key: missing.BlobKey(), ID: "old", Started: old}))
258 require.NoError(t, fn(ports.MultipartInfo{Key: missing.BlobKey(), ID: "young", Started: now}))
259 return nil
260 })
261 b.EXPECT().AbortMultipart(mock.Anything, missing.BlobKey(), "old").Return(nil).Once()
262 c.sweepArtifactBlobOrphans(ctx)
263}
264
265func TestArtifactRemoveRunsWithoutLegacyReader(t *testing.T) {
266 ctx := t.Context()
267 b, _, _ := artifactBlobMock(t, false)
268 c := &CI{cfg: CIConfig{Blobs: b}}
269 a := domain.CIArtifact{RunID: 1, ID: 2}
270 other := a
271 other.RunID = 2
272 for _, key := range []string{a.BlobKey(), a.BlockKey("a"), other.BlobKey()} {
273 _, err := b.Put(ctx, key, bytes.NewReader([]byte("x")), 1)
274 require.NoError(t, err)
275 }
276 _, err := b.NewMultipart(ctx, a.BlobKey())
277 require.NoError(t, err)
278 require.NoError(t, c.removeArtifactRuns(ctx, []int64{1}))
279 _, err = b.Stat(ctx, a.BlobKey())
280 assert.ErrorIs(t, err, domain.ErrNotFound)
281 _, err = b.Stat(ctx, a.BlockKey("a"))
282 assert.ErrorIs(t, err, domain.ErrNotFound)
283 _, err = b.Stat(ctx, other.BlobKey())
284 require.NoError(t, err)
285 require.NoError(t, b.ListMultipart(ctx, "ci/", func(ports.MultipartInfo) error { t.Error("upload survived run deletion"); return nil }))
286}
287
288func TestArtifactBlobIDs(t *testing.T) {
289 a := domain.CIArtifact{RunID: 7, ID: 42}
290 for _, key := range []string{a.BlobKey(), a.BlockKey("abcdef.0123")} {
291 run, id, block, ok := artifactBlobIDs(key)
292 assert.True(t, ok)
293 assert.EqualValues(t, 7, run)
294 assert.EqualValues(t, 42, id)
295 assert.Equal(t, strings.Contains(key, ".blocks/"), block)
296 }
297 for _, key := range []string{"release/1", "ci/1/2", a.BlobKey() + ".blocks/", a.BlobKey() + "/wrong"} {
298 _, _, _, ok := artifactBlobIDs(key)
299 assert.False(t, ok, key)
300 }
301}