frostfs-s3-gw/api/handler/multipart_upload_test.go

package handler

import (
	"bytes"
	"encoding/xml"
	"net/http"
	"net/url"
	"strconv"
	"testing"
	"time"

	s3Errors "git.frostfs.info/TrueCloudLab/frostfs-s3-gw/api/errors"
	"github.com/stretchr/testify/require"
)

func TestPeriodicWriter(t *testing.T) {
	const dur = 100 * time.Millisecond
	const whitespaces = 8
	expected := []byte(xml.Header)
	for i := 0; i < whitespaces; i++ {
		expected = append(expected, []byte(" ")...)
	}

	t.Run("writes data", func(t *testing.T) {
		buf := bytes.NewBuffer(nil)
		stop := periodicXMLWriter(buf, dur)

		// N number of whitespaces + half durations to guarantee at least N writes in buffer
		time.Sleep(whitespaces*dur + dur/2)
		require.True(t, stop())
		require.Equal(t, expected, buf.Bytes())

		t.Run("no additional data after stop", func(t *testing.T) {
			time.Sleep(2 * dur)
			require.Equal(t, expected, buf.Bytes())
		})
	})

	t.Run("does not write data", func(t *testing.T) {
		buf := bytes.NewBuffer(nil)
		stop := periodicXMLWriter(buf, dur)
		time.Sleep(dur / 2)
		require.False(t, stop())
		require.Empty(t, buf.Bytes())

		t.Run("disabled", func(t *testing.T) {
			stop = periodicXMLWriter(buf, 0)
			require.False(t, stop())
			require.Empty(t, buf.Bytes())
		})
	})
}

func TestMultipartUploadInvalidPart(t *testing.T) {
	hc := prepareHandlerContext(t)

	bktName, objName := "bucket-to-upload-part", "object-multipart"
	createTestBucket(hc, bktName)
	partSize := 8 // less than min part size

	multipartUpload := createMultipartUpload(hc, bktName, objName, map[string]string{})
	etag1, _ := uploadPart(hc, bktName, objName, multipartUpload.UploadID, 1, partSize)
	etag2, _ := uploadPart(hc, bktName, objName, multipartUpload.UploadID, 2, partSize)
	w := completeMultipartUploadBase(hc, bktName, objName, multipartUpload.UploadID, []string{etag1, etag2})
	assertS3Error(hc.t, w, s3Errors.GetAPIError(s3Errors.ErrEntityTooSmall))
}

func TestMultipartReUploadPart(t *testing.T) {
	hc := prepareHandlerContext(t)

	bktName, objName := "bucket-to-upload-part", "object-multipart"
	bktInfo := createTestBucket(hc, bktName)
	partSizeLast := 8 // less than min part size
	partSizeFirst := 5 * 1024 * 1024

	uploadInfo := createMultipartUpload(hc, bktName, objName, map[string]string{})
	etag1, _ := uploadPart(hc, bktName, objName, uploadInfo.UploadID, 1, partSizeLast)
	etag2, _ := uploadPart(hc, bktName, objName, uploadInfo.UploadID, 2, partSizeFirst)

	list := listParts(hc, bktName, objName, uploadInfo.UploadID)
	require.Len(t, list.Parts, 2)
	require.Equal(t, etag1, list.Parts[0].ETag)
	require.Equal(t, etag2, list.Parts[1].ETag)

	w := completeMultipartUploadBase(hc, bktName, objName, uploadInfo.UploadID, []string{etag1, etag2})
	assertS3Error(hc.t, w, s3Errors.GetAPIError(s3Errors.ErrEntityTooSmall))

	etag1, data1 := uploadPart(hc, bktName, objName, uploadInfo.UploadID, 1, partSizeFirst)
	etag2, data2 := uploadPart(hc, bktName, objName, uploadInfo.UploadID, 2, partSizeLast)

	list = listParts(hc, bktName, objName, uploadInfo.UploadID)
	require.Len(t, list.Parts, 2)
	require.Equal(t, etag1, list.Parts[0].ETag)
	require.Equal(t, etag2, list.Parts[1].ETag)

	innerUploadInfo, err := hc.tree.GetMultipartUpload(hc.context, bktInfo, objName, uploadInfo.UploadID)
	require.NoError(t, err)
	treeParts, err := hc.tree.GetParts(hc.Context(), bktInfo, innerUploadInfo.ID)
	require.NoError(t, err)
	require.Len(t, treeParts, len(list.Parts))

	w = completeMultipartUploadBase(hc, bktName, objName, uploadInfo.UploadID, []string{etag1, etag2})
	assertStatus(hc.t, w, http.StatusOK)

	data, _ := getObject(hc, bktName, objName)
	equalDataSlices(t, append(data1, data2...), data)
}

func TestListMultipartUploads(t *testing.T) {
	hc := prepareHandlerContext(t)

	bktName := "bucket-to-list-uploads"
	createTestBucket(hc, bktName)

	objName1 := "/my/object/name"
	uploadInfo1 := createMultipartUpload(hc, bktName, objName1, map[string]string{})
	objName2 := "/my/object2"
	uploadInfo2 := createMultipartUpload(hc, bktName, objName2, map[string]string{})
	objName3 := "/zzz/object/name3"
	uploadInfo3 := createMultipartUpload(hc, bktName, objName3, map[string]string{})

	t.Run("check upload key", func(t *testing.T) {
		listUploads := listAllMultipartUploads(hc, bktName)
		require.Len(t, listUploads.Uploads, 3)
		for i, upload := range []*InitiateMultipartUploadResponse{uploadInfo1, uploadInfo2, uploadInfo3} {
			require.Equal(t, upload.UploadID, listUploads.Uploads[i].UploadID)
			require.Equal(t, upload.Key, listUploads.Uploads[i].Key)
		}
	})

	t.Run("check max uploads", func(t *testing.T) {
		listUploads := listMultipartUploadsBase(hc, bktName, "", "", "", "", 2)
		require.Len(t, listUploads.Uploads, 2)
		require.Equal(t, uploadInfo1.UploadID, listUploads.Uploads[0].UploadID)
		require.Equal(t, uploadInfo2.UploadID, listUploads.Uploads[1].UploadID)
	})

	t.Run("check prefix", func(t *testing.T) {
		listUploads := listMultipartUploadsBase(hc, bktName, "/my", "", "", "", -1)
		require.Len(t, listUploads.Uploads, 2)
		require.Equal(t, uploadInfo1.UploadID, listUploads.Uploads[0].UploadID)
		require.Equal(t, uploadInfo2.UploadID, listUploads.Uploads[1].UploadID)
	})

	t.Run("check markers", func(t *testing.T) {
		t.Run("check only key-marker", func(t *testing.T) {
			listUploads := listMultipartUploadsBase(hc, bktName, "", "", "", objName2, -1)
			require.Len(t, listUploads.Uploads, 1)
			// If upload-id-marker is not specified, only the keys lexicographically greater than the specified key-marker will be included in the list.
			require.Equal(t, uploadInfo3.UploadID, listUploads.Uploads[0].UploadID)
		})

		t.Run("check only upload-id-marker", func(t *testing.T) {
			uploadIDMarker := uploadInfo1.UploadID
			if uploadIDMarker > uploadInfo2.UploadID {
				uploadIDMarker = uploadInfo2.UploadID
			}
			listUploads := listMultipartUploadsBase(hc, bktName, "", "", uploadIDMarker, "", -1)
			// If key-marker is not specified, the upload-id-marker parameter is ignored.
			require.Len(t, listUploads.Uploads, 3)
		})

		t.Run("check key-marker along with upload-id-marker", func(t *testing.T) {
			uploadIDMarker := "00000000-0000-0000-0000-000000000000"

			listUploads := listMultipartUploadsBase(hc, bktName, "", "", uploadIDMarker, objName3, -1)
			require.Len(t, listUploads.Uploads, 1)
			// If upload-id-marker is specified, any multipart uploads for a key equal to the key-marker might also be included,
			// provided those multipart uploads have upload IDs lexicographically greater than the specified upload-id-marker.
			require.Equal(t, uploadInfo3.UploadID, listUploads.Uploads[0].UploadID)
		})
	})
}

func listAllMultipartUploads(hc *handlerContext, bktName string) *ListMultipartUploadsResponse {
	return listMultipartUploadsBase(hc, bktName, "", "", "", "", -1)
}

func listMultipartUploadsBase(hc *handlerContext, bktName, prefix, delimiter, uploadIDMarker, keyMarker string, maxUploads int) *ListMultipartUploadsResponse {
	query := make(url.Values)
	query.Set(prefixQueryName, prefix)
	query.Set(delimiterQueryName, delimiter)
	query.Set(uploadIDMarkerQueryName, uploadIDMarker)
	query.Set(keyMarkerQueryName, keyMarker)
	if maxUploads != -1 {
		query.Set(maxUploadsQueryName, strconv.Itoa(maxUploads))
	}

	w, r := prepareTestRequestWithQuery(hc, bktName, "", query, nil)

	hc.Handler().ListMultipartUploadsHandler(w, r)
	listPartsResponse := &ListMultipartUploadsResponse{}
	readResponse(hc.t, w, http.StatusOK, listPartsResponse)

	return listPartsResponse
}

func listParts(hc *handlerContext, bktName, objName string, uploadID string) *ListPartsResponse {
	return listPartsBase(hc, bktName, objName, false, uploadID)
}

func listPartsBase(hc *handlerContext, bktName, objName string, encrypted bool, uploadID string) *ListPartsResponse {
	query := make(url.Values)
	query.Set(uploadIDQuery, uploadID)

	w, r := prepareTestRequestWithQuery(hc, bktName, objName, query, nil)
	if encrypted {
		setEncryptHeaders(r)
	}

	hc.Handler().ListPartsHandler(w, r)
	listPartsResponse := &ListPartsResponse{}
	readResponse(hc.t, w, http.StatusOK, listPartsResponse)

	return listPartsResponse
}