Something went wrong. Try again.
This repository has no description
Something went wrong. Try again.
7.9 kB · 214 lines
Go
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215package s3
import ( "context" "fmt" "sync" "testing" "time"
"github.com/aws/aws-sdk-go-v2/aws" "github.com/aws/aws-sdk-go-v2/service/s3" "github.com/aws/aws-sdk-go-v2/service/s3/types" "github.com/stretchr/testify/require")
// fakeCopyClient is an in-memory stand-in for the subset of *s3.Client that// Copy uses. HeadObject reports a configurable size/content-type (so the// >5 GiB multipart path can be exercised without allocating gigabytes), and// the multipart ops record the ranges they were asked to copy.type fakeCopyClient struct { headSize int64 headContentType string headErr error
copyDelay time.Duration failPart int32 // if >0, UploadPartCopy for this part number returns an error
srcBytes []byte // for the small-object CopyObject path dst map[string][]byte // CopyObject writes here when non-nil
mu sync.Mutex copyObjectCalls int createCalls int createdType string partRanges map[int32][2]int64 completedParts []types.CompletedPart aborted bool inFlight int maxInFlight int}
func (f *fakeCopyClient) HeadObject(context.Context, *s3.HeadObjectInput, ...func(*s3.Options)) (*s3.HeadObjectOutput, error) { if f.headErr != nil { return nil, f.headErr } out := &s3.HeadObjectOutput{ContentLength: aws.Int64(f.headSize)} if f.headContentType != "" { out.ContentType = aws.String(f.headContentType) } return out, nil}
func (f *fakeCopyClient) CopyObject(_ context.Context, in *s3.CopyObjectInput, _ ...func(*s3.Options)) (*s3.CopyObjectOutput, error) { f.mu.Lock() defer f.mu.Unlock() f.copyObjectCalls++ if f.dst != nil { f.dst[aws.ToString(in.Key)] = f.srcBytes } return &s3.CopyObjectOutput{}, nil}
func (f *fakeCopyClient) CreateMultipartUpload(_ context.Context, in *s3.CreateMultipartUploadInput, _ ...func(*s3.Options)) (*s3.CreateMultipartUploadOutput, error) { f.mu.Lock() defer f.mu.Unlock() f.createCalls++ f.createdType = aws.ToString(in.ContentType) return &s3.CreateMultipartUploadOutput{UploadId: aws.String("test-upload-id")}, nil}
func (f *fakeCopyClient) UploadPartCopy(_ context.Context, in *s3.UploadPartCopyInput, _ ...func(*s3.Options)) (*s3.UploadPartCopyOutput, error) { f.mu.Lock() f.inFlight++ if f.inFlight > f.maxInFlight { f.maxInFlight = f.inFlight } f.mu.Unlock()
if f.copyDelay > 0 { time.Sleep(f.copyDelay) }
num := aws.ToInt32(in.PartNumber) var start, end int64 if _, err := fmt.Sscanf(aws.ToString(in.CopySourceRange), "bytes=%d-%d", &start, &end); err != nil { return nil, fmt.Errorf("bad CopySourceRange %q: %w", aws.ToString(in.CopySourceRange), err) }
f.mu.Lock() defer f.mu.Unlock() f.inFlight-- if f.failPart > 0 && num == f.failPart { return nil, fmt.Errorf("simulated failure on part %d", num) } f.partRanges[num] = [2]int64{start, end} return &s3.UploadPartCopyOutput{ CopyPartResult: &types.CopyPartResult{ETag: aws.String(fmt.Sprintf("etag-%d", num))}, }, nil}
func (f *fakeCopyClient) CompleteMultipartUpload(_ context.Context, in *s3.CompleteMultipartUploadInput, _ ...func(*s3.Options)) (*s3.CompleteMultipartUploadOutput, error) { f.mu.Lock() defer f.mu.Unlock() f.completedParts = in.MultipartUpload.Parts return &s3.CompleteMultipartUploadOutput{}, nil}
func (f *fakeCopyClient) AbortMultipartUpload(context.Context, *s3.AbortMultipartUploadInput, ...func(*s3.Options)) (*s3.AbortMultipartUploadOutput, error) { f.mu.Lock() defer f.mu.Unlock() f.aborted = true return &s3.AbortMultipartUploadOutput{}, nil}
// TestCopySmallObjectUsesCopyObject verifies sub-5-GiB objects take the// single-call CopyObject path and land byte-for-byte at the destination.func TestCopySmallObjectUsesCopyObject(t *testing.T) { src := []byte("hello world, a small object") fake := &fakeCopyClient{ headSize: int64(len(src)), headContentType: "video/mp4", srcBytes: src, dst: map[string][]byte{}, partRanges: map[int32][2]int64{}, } require.NoError(t, copyObject(context.Background(), fake, "bucket", "src", "dst")) require.Equal(t, 1, fake.copyObjectCalls) require.Equal(t, 0, fake.createCalls, "small object must not open a multipart upload") require.Equal(t, src, fake.dst["dst"])}
// TestCopyAtThresholdUsesCopyObject pins the boundary: an object exactly at// maxCopyObjectSize still copies in one shot.func TestCopyAtThresholdUsesCopyObject(t *testing.T) { fake := &fakeCopyClient{headSize: maxCopyObjectSize, partRanges: map[int32][2]int64{}} require.NoError(t, copyObject(context.Background(), fake, "bucket", "src", "dst")) require.Equal(t, 1, fake.copyObjectCalls) require.Equal(t, 0, fake.createCalls)}
// TestCopyLargeObjectUsesMultipart verifies an over-5-GiB object is copied// with a multipart UploadPartCopy whose parts tile the source exactly,// preserve content type, finish in ascending order, and run concurrently.func TestCopyLargeObjectUsesMultipart(t *testing.T) { const size = int64(12) * 1024 * 1024 * 1024 // 12 GiB -> spans many parts fake := &fakeCopyClient{ headSize: size, headContentType: "video/mp4", copyDelay: 5 * time.Millisecond, // force overlap so concurrency is observable partRanges: map[int32][2]int64{}, } require.NoError(t, copyObject(context.Background(), fake, "bucket", "src", "dst"))
require.Equal(t, 0, fake.copyObjectCalls, "large object must not use single CopyObject") require.Equal(t, 1, fake.createCalls) require.Equal(t, "video/mp4", fake.createdType, "content type must be preserved")
wantParts := int((size + copyPartSize - 1) / copyPartSize) require.Len(t, fake.partRanges, wantParts) require.Len(t, fake.completedParts, wantParts)
// Parts must tile [0,size) contiguously with no gaps or overlaps, and // every part except the last is exactly copyPartSize. var covered int64 for n := int32(1); n <= int32(wantParts); n++ { r, ok := fake.partRanges[n] require.Truef(t, ok, "missing part %d", n) require.Equalf(t, covered, r[0], "part %d start should continue from previous end", n) require.GreaterOrEqual(t, r[1], r[0]) if int(n) < wantParts { require.Equalf(t, int64(copyPartSize), r[1]-r[0]+1, "non-final part %d should be a full part", n) } covered = r[1] + 1 } require.Equal(t, size, covered, "parts must cover the whole object")
// CompleteMultipartUpload requires ascending part numbers. for i := 1; i < len(fake.completedParts); i++ { require.Less(t, aws.ToInt32(fake.completedParts[i-1].PartNumber), aws.ToInt32(fake.completedParts[i].PartNumber)) } require.Greater(t, fake.maxInFlight, 1, "expected part copies to run concurrently") require.False(t, fake.aborted)}
// TestCopyLargeObjectAbortsOnPartError verifies a failed part copy aborts// the multipart upload (so S3 doesn't retain orphaned parts) and never// completes it.func TestCopyLargeObjectAbortsOnPartError(t *testing.T) { const size = int64(12) * 1024 * 1024 * 1024 fake := &fakeCopyClient{ headSize: size, partRanges: map[int32][2]int64{}, failPart: 3, } err := copyObject(context.Background(), fake, "bucket", "src", "dst") require.Error(t, err) require.Contains(t, err.Error(), "upload part copy 3") require.True(t, fake.aborted, "a failed part must abort the upload") require.Empty(t, fake.completedParts, "must not complete after a part failure")}
// TestCopyHeadErrorPropagates verifies a missing source surfaces the// HeadObject error (which blob.Move sniffs as NotFound for idempotency)// before any copy is attempted.func TestCopyHeadErrorPropagates(t *testing.T) { fake := &fakeCopyClient{headErr: fmt.Errorf("api error NotFound: object missing")} err := copyObject(context.Background(), fake, "bucket", "missing", "dst") require.Error(t, err) require.Contains(t, err.Error(), "NotFound") require.Equal(t, 0, fake.copyObjectCalls) require.Equal(t, 0, fake.createCalls)}