1// SPDX-License-Identifier: AGPL-3.0-or-later
2
3package usecase_test
4
5import (
6 "bytes"
7 "context"
8 "crypto/sha256"
9 "errors"
10 "fmt"
11 "io"
12 "sync"
13 "testing"
14 "time"
15
16 "bnema.dev/ermite/internal/domain"
17 "bnema.dev/ermite/internal/ports"
18 "bnema.dev/ermite/internal/ports/mocks"
19 "bnema.dev/ermite/internal/usecase"
20 "github.com/stretchr/testify/mock"
21 "github.com/stretchr/testify/require"
22)
23
24func (r *releaseRig) upload(t *testing.T, rel domain.Release, n string, data []byte) domain.ReleaseAsset {
25 t.Helper()
26 a, err := r.svc.UploadAsset(r.ctx, domain.Root(), r.repo, ports.UploadAssetParams{ReleaseID: rel.ID, Name: n, Size: int64(len(data))}, bytes.NewReader(data))
27 require.NoError(t, err)
28 return a
29}
30func TestReleaseAssetRoundTrip(t *testing.T) {
31 r := newReleaseRig(t, ports.ReleaseLimits{AssetMax: 10, Quota: 10})
32 rel := r.create(t, "v1", false)
33 a := r.upload(t, rel, "binary", []byte("0123456789"))
34 require.Equal(t, fmt.Sprintf("sha256:%x", sha256.Sum256([]byte("0123456789"))), a.Digest)
35 usage, err := r.svc.Usage(r.ctx, domain.Anonymous(), r.repo)
36 require.NoError(t, err)
37 require.EqualValues(t, 10, usage.Used)
38 for _, tc := range []struct {
39 offset, length int64
40 want string
41 err error
42 }{{0, -1, "0123456789", nil}, {3, 4, "3456", nil}, {9, 1, "9", nil}, {10, -1, "", domain.ErrOutOfRange}, {11, -1, "", domain.ErrOutOfRange}} {
43 var got bytes.Buffer
44 metadata := false
45 err := r.svc.DownloadAsset(r.ctx, domain.Anonymous(), r.repo, ports.AssetRef{Latest: true, Name: a.Name}, tc.offset, tc.length, func(c ports.AssetChunk) error {
46 if c.Asset != nil {
47 require.False(t, metadata)
48 metadata = true
49 require.Equal(t, a.ID, c.Asset.ID)
50 } else {
51 require.True(t, metadata)
52 got.Write(c.Data)
53 }
54 return nil
55 })
56 require.ErrorIs(t, err, tc.err)
57 require.Equal(t, tc.want, got.String())
58 }
59 _, err = r.svc.UploadAsset(r.ctx, domain.Root(), r.repo, ports.UploadAssetParams{ReleaseID: rel.ID, Name: "large", Size: 11}, bytes.NewReader(nil))
60 require.ErrorIs(t, err, domain.ErrResourceExhausted)
61 _, err = r.svc.UploadAsset(r.ctx, domain.Root(), r.repo, ports.UploadAssetParams{ReleaseID: rel.ID, Name: "quota", Size: 1}, bytes.NewReader([]byte("x")))
62 require.ErrorIs(t, err, domain.ErrResourceExhausted)
63 _, err = r.svc.UploadAsset(r.ctx, domain.Root(), r.repo, ports.UploadAssetParams{ReleaseID: rel.ID, Name: a.Name, Size: 0}, bytes.NewReader(nil))
64 require.ErrorIs(t, err, domain.ErrAlreadyExists)
65 replacement, err := r.svc.UploadAsset(r.ctx, domain.Root(), r.repo, ports.UploadAssetParams{ReleaseID: rel.ID, Name: a.Name, Size: 2, Replace: true}, bytes.NewReader([]byte("hi")))
66 require.NoError(t, err)
67 require.NotEqual(t, a.ID, replacement.ID)
68 _, err = r.blobs.Stat(r.ctx, a.BlobKey())
69 require.ErrorIs(t, err, domain.ErrNotFound)
70 renamed := "new"
71 label := "label"
72 replacement, err = r.svc.UpdateAsset(r.ctx, domain.Root(), r.repo, replacement.ID, ports.UpdateAssetParams{Name: &renamed, Label: &label})
73 require.NoError(t, err)
74 require.Equal(t, label, replacement.Label)
75 list, err := r.svc.ListAssets(r.ctx, domain.Anonymous(), r.repo, rel.ID)
76 require.NoError(t, err)
77 require.Len(t, list, 1)
78 require.Equal(t, renamed, list[0].Name)
79 require.NoError(t, r.svc.DeleteAsset(r.ctx, domain.Root(), r.repo, replacement.ID))
80 usage, err = r.svc.Usage(r.ctx, domain.Root(), r.repo)
81 require.NoError(t, err)
82 require.Zero(t, usage.Used)
83}
84func TestReleaseAssetCountBoundary(t *testing.T) {
85 r := newReleaseRig(t, ports.ReleaseLimits{})
86 rel := r.create(t, "v1", false)
87 // Seed 999 records in one batch, then exercise the exact public boundary.
88 used, v, err := r.store.Releases().ReleaseUsage(r.ctx, r.repo)
89 require.NoError(t, err)
90 w := ports.ReleaseWrite{Repo: r.repo, Used: used, UsageVersion: v}
91 for range 999 {
92 id, err := r.store.Releases().NextReleaseID(r.ctx)
93 require.NoError(t, err)
94 w.PutAssets = append(w.PutAssets, domain.ReleaseAsset{ID: id, ReleaseID: rel.ID, Repo: r.repo, Name: fmt.Sprint(id), State: domain.ReleaseAssetUploaded})
95 }
96 require.NoError(t, r.store.Releases().ApplyReleases(r.ctx, w))
97 r.upload(t, rel, "thousandth", nil)
98 _, err = r.svc.UploadAsset(r.ctx, domain.Root(), r.repo, ports.UploadAssetParams{ReleaseID: rel.ID, Name: "1001"}, bytes.NewReader(nil))
99 require.ErrorIs(t, err, domain.ErrResourceExhausted)
100}
101func TestReleaseAssetBadBodies(t *testing.T) {
102 bodyErr := errors.New("reader failed")
103 for _, size := range []int64{4, 9 << 20} {
104 for _, kind := range []string{"short", "long", "error", "cancel"} {
105 t.Run(fmt.Sprint(size, "/", kind), func(t *testing.T) {
106 r := newReleaseRig(t, ports.ReleaseLimits{})
107 rel := r.create(t, "v1", false)
108 var body io.Reader
109 switch kind {
110 case "short":
111 body = bytes.NewReader(make([]byte, size-1))
112 case "long":
113 body = bytes.NewReader(make([]byte, size+1))
114 case "error":
115 body = errReader{bodyErr}
116 case "cancel":
117 body = errReader{context.Canceled}
118 }
119 _, err := r.svc.UploadAsset(r.ctx, domain.Root(), r.repo, ports.UploadAssetParams{ReleaseID: rel.ID, Name: "bad", Size: size}, body)
120 switch kind {
121 case "error":
122 require.ErrorIs(t, err, bodyErr)
123 case "cancel":
124 require.ErrorIs(t, err, context.Canceled)
125 default:
126 require.ErrorIs(t, err, domain.ErrInvalid)
127 }
128 all, err := r.store.Releases().AllAssets(r.ctx)
129 require.NoError(t, err)
130 require.Empty(t, all)
131 u, err := r.svc.Usage(r.ctx, domain.Root(), r.repo)
132 require.NoError(t, err)
133 require.Zero(t, u.Used)
134 require.NoError(t, r.blobs.List(r.ctx, "release/", func(ports.BlobInfo) error { t.Error("blob left behind"); return nil }))
135 require.NoError(t, r.blobs.ListMultipart(r.ctx, "release/", func(ports.MultipartInfo) error { t.Error("multipart left behind"); return nil }))
136 })
137 }
138 }
139}
140
141type releaseBlobSpy struct {
142 ports.BlobStore
143 put, multipart int
144 parts []int64
145}
146
147func (s *releaseBlobSpy) Put(ctx context.Context, key string, b io.Reader, size int64) (ports.BlobInfo, error) {
148 s.put++
149 return s.BlobStore.Put(ctx, key, b, size)
150}
151func (s *releaseBlobSpy) NewMultipart(ctx context.Context, key string) (ports.MultipartUpload, error) {
152 s.multipart++
153 u, e := s.BlobStore.NewMultipart(ctx, key)
154 return &releasePartSpy{MultipartUpload: u, spy: s}, e
155}
156
157type releasePartSpy struct {
158 ports.MultipartUpload
159 spy *releaseBlobSpy
160}
161
162func (s *releasePartSpy) UploadPart(ctx context.Context, n int, b io.Reader, size int64) error {
163 s.spy.parts = append(s.spy.parts, size)
164 return s.MultipartUpload.UploadPart(ctx, n, b, size)
165}
166func TestReleaseAssetUploadPaths(t *testing.T) {
167 for _, size := range []int{1 << 20, 9 << 20} {
168 t.Run(fmt.Sprint(size), func(t *testing.T) {
169 r := newReleaseRig(t, ports.ReleaseLimits{})
170 rel := r.create(t, "v1", false)
171 spy := &releaseBlobSpy{BlobStore: r.blobs}
172 bus := mocks.NewEventBus(t)
173 bus.EXPECT().Publish(mock.Anything, mock.Anything).Return().Maybe()
174 r.svc = usecase.NewReleases(r.store.Releases(), spy, r.store, r.reader, r.tags, r.pushes, bus, r.clock, ports.ReleaseLimits{})
175 data := bytes.Repeat([]byte("z"), size)
176 a := r.upload(t, rel, "blob", data)
177 require.Equal(t, fmt.Sprintf("sha256:%x", sha256.Sum256(data)), a.Digest)
178 if size == 1<<20 {
179 require.Equal(t, 1, spy.put)
180 require.Zero(t, spy.multipart)
181 } else {
182 require.Zero(t, spy.put)
183 require.Equal(t, 1, spy.multipart)
184 require.Equal(t, []int64{8 << 20, 1 << 20}, spy.parts)
185 }
186 })
187 }
188}
189func TestReleaseAssetStaleReplacementAndReleaseDelete(t *testing.T) {
190 r := newReleaseRig(t, ports.ReleaseLimits{})
191 rel := r.create(t, "v1", false)
192 a := r.upload(t, rel, "pending", []byte("old"))
193 current, err := r.store.Releases().GetAsset(r.ctx, a.ID)
194 require.NoError(t, err)
195 current.State = domain.ReleaseAssetUploading
196 current.UpdatedAt = now.Add(-61 * time.Second)
197 used, v, err := r.store.Releases().ReleaseUsage(r.ctx, r.repo)
198 require.NoError(t, err)
199 require.NoError(t, r.store.Releases().ApplyReleases(r.ctx, ports.ReleaseWrite{Repo: r.repo, Used: used, UsageVersion: v, PutAssets: []domain.ReleaseAsset{current}}))
200 b := r.upload(t, rel, "pending", []byte("new"))
201 require.NotEqual(t, a.ID, b.ID)
202 require.NoError(t, r.svc.DeleteRelease(r.ctx, domain.Root(), r.repo, rel.ID))
203 _, err = r.blobs.Stat(r.ctx, b.BlobKey())
204 require.ErrorIs(t, err, domain.ErrNotFound)
205 u, err := r.svc.Usage(r.ctx, domain.Root(), r.repo)
206 require.NoError(t, err)
207 require.Zero(t, u.Used)
208}
209
210type releaseBlockingReader struct {
211 entered chan<- struct{}
212 release <-chan struct{}
213}
214
215func (b releaseBlockingReader) Read([]byte) (int, error) {
216 b.entered <- struct{}{}
217 <-b.release
218 return 0, io.EOF
219}
220func TestReleaseAssetConcurrencyCap(t *testing.T) {
221 r := newReleaseRig(t, ports.ReleaseLimits{})
222 rel := r.create(t, "v1", false)
223 entered := make(chan struct{}, 16)
224 release := make(chan struct{})
225 var wg sync.WaitGroup
226 errs := make(chan error, 16)
227 for i := range 16 {
228 wg.Go(func() {
229 _, err := r.svc.UploadAsset(r.ctx, domain.Root(), r.repo, ports.UploadAssetParams{ReleaseID: rel.ID, Name: fmt.Sprint(i), Size: 1}, releaseBlockingReader{entered, release})
230 errs <- err
231 })
232 }
233 // Every reader has a reservation and holds its slot before the 17th call.
234 for range 16 {
235 <-entered
236 }
237 _, err := r.svc.UploadAsset(r.ctx, domain.Root(), r.repo, ports.UploadAssetParams{ReleaseID: rel.ID, Name: "excess", Size: 0}, bytes.NewReader(nil))
238 require.ErrorIs(t, err, domain.ErrResourceExhausted)
239 close(release)
240 wg.Wait()
241 close(errs)
242 for err := range errs {
243 require.ErrorIs(t, err, domain.ErrInvalid)
244 }
245}
246
247type errReader struct{ err error }
248
249func (r errReader) Read([]byte) (int, error) { return 0, r.err }
250
251func TestReleaseAssetDraftAccessAndRenameCollision(t *testing.T) {
252 r := newReleaseRig(t, ports.ReleaseLimits{})
253 rel := r.create(t, "draft", true)
254 a := r.upload(t, rel, "a", []byte("abc"))
255 b := r.upload(t, rel, "b", nil)
256 _, err := r.svc.GetAsset(r.ctx, domain.Anonymous(), r.repo, ports.AssetRef{ID: a.ID})
257 require.ErrorIs(t, err, domain.ErrNotFound)
258 _, err = r.svc.ListAssets(r.ctx, domain.Anonymous(), r.repo, rel.ID)
259 require.ErrorIs(t, err, domain.ErrNotFound)
260 _, err = r.svc.UpdateAsset(r.ctx, domain.Root(), r.repo, b.ID, ports.UpdateAssetParams{Name: &a.Name})
261 require.ErrorIs(t, err, domain.ErrAlreadyExists)
262 got, err := r.svc.GetAsset(r.ctx, domain.Root(), r.repo, ports.AssetRef{Tag: rel.Tag, Name: a.Name})
263 require.NoError(t, err)
264 require.Equal(t, a.ID, got.ID)
265 assets, err := r.svc.AssetsOf(r.ctx, r.repo)
266 require.NoError(t, err)
267 require.Len(t, assets, 2)
268 require.NoError(t, r.store.Delete(r.ctx, r.repo))
269 r.svc.OnRepoDeleted(r.ctx, r.repo, assets)
270 _, err = r.blobs.Stat(r.ctx, a.BlobKey())
271 require.ErrorIs(t, err, domain.ErrNotFound)
272 all, err := r.store.Releases().AllAssets(r.ctx)
273 require.NoError(t, err)
274 require.Empty(t, all)
275}
276
277func TestReleaseAssetQuotaAcrossReleasesAndWrongRepository(t *testing.T) {
278 r := newReleaseRig(t, ports.ReleaseLimits{AssetMax: 3, Quota: 5})
279 a := r.create(t, "a", false)
280 b := r.create(t, "b", false)
281 r.upload(t, a, "a", []byte("abc"))
282 r.upload(t, b, "b", []byte("de"))
283 _, err := r.svc.UploadAsset(r.ctx, domain.Root(), r.repo, ports.UploadAssetParams{ReleaseID: b.ID, Name: "extra", Size: 1}, bytes.NewReader([]byte("x")))
284 require.ErrorIs(t, err, domain.ErrResourceExhausted)
285 other := name(t, "other")
286 require.NoError(t, r.store.Create(r.ctx, domain.Repo{Name: other, Visibility: domain.VisibilityPublic, CreatedAt: now}))
287 _, err = r.svc.UploadAsset(r.ctx, domain.Root(), other, ports.UploadAssetParams{ReleaseID: b.ID, Name: "wrong"}, bytes.NewReader(nil))
288 require.ErrorIs(t, err, domain.ErrPreconditionFailed)
289}
290
291func TestReleaseAssetDownloadZeroBytes(t *testing.T) {
292 r := newReleaseRig(t, ports.ReleaseLimits{AssetMax: 10, Quota: 10})
293 rel := r.create(t, "v1", false)
294 a := r.upload(t, rel, "empty", nil)
295 var chunks []ports.AssetChunk
296 err := r.svc.DownloadAsset(r.ctx, domain.Anonymous(), r.repo, ports.AssetRef{ID: a.ID}, 0, -1, func(c ports.AssetChunk) error {
297 chunks = append(chunks, c)
298 return nil
299 })
300 require.NoError(t, err)
301 require.Len(t, chunks, 1)
302 require.Equal(t, a.ID, chunks[0].Asset.ID)
303
304 // A missing blob stays an error.
305 require.NoError(t, r.blobs.Delete(r.ctx, a.BlobKey()))
306 err = r.svc.DownloadAsset(r.ctx, domain.Anonymous(), r.repo, ports.AssetRef{ID: a.ID}, 0, -1, func(ports.AssetChunk) error { return nil })
307 require.ErrorIs(t, err, domain.ErrNotFound)
308}