seaweedfs/weed/shell/command_s3_clean_uploads_test.go

package shell

import (
	"context"
	"errors"
	"io"
	"testing"

	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
	"github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
	"google.golang.org/grpc"
	"google.golang.org/grpc/metadata"
)

type stubFilerClient struct {
	client filer_pb.SeaweedFilerClient
}

func (s stubFilerClient) WithFilerClient(_ bool, fn func(filer_pb.SeaweedFilerClient) error) error {
	return fn(s.client)
}

func (s stubFilerClient) AdjustedUrl(location *filer_pb.Location) string { return "" }

func (s stubFilerClient) GetDataCenter() string { return "" }

type stubFilerSvc struct {
	filer_pb.SeaweedFilerClient
	lookupResp *filer_pb.LookupDirectoryEntryResponse
	lookupErr  error
	listResp   []*filer_pb.Entry
	listDir    string
	listErr    error
	lookupReq  *filer_pb.LookupDirectoryEntryRequest
	listReq    *filer_pb.ListEntriesRequest
}

func (s *stubFilerSvc) LookupDirectoryEntry(ctx context.Context, req *filer_pb.LookupDirectoryEntryRequest, opts ...grpc.CallOption) (*filer_pb.LookupDirectoryEntryResponse, error) {
	s.lookupReq = req
	return s.lookupResp, s.lookupErr
}

func (s *stubFilerSvc) ListEntries(ctx context.Context, req *filer_pb.ListEntriesRequest, opts ...grpc.CallOption) (filer_pb.SeaweedFiler_ListEntriesClient, error) {
	s.listReq = req
	if s.listErr != nil {
		return nil, s.listErr
	}
	if s.listDir != "" && req.Directory != s.listDir {
		return &stubListStream{}, nil
	}
	return &stubListStream{entries: s.listResp}, nil
}

type stubListStream struct {
	entries []*filer_pb.Entry
	idx     int
}

func (s *stubListStream) Header() (metadata.MD, error) { return nil, nil }
func (s *stubListStream) Trailer() metadata.MD         { return nil }
func (s *stubListStream) CloseSend() error             { return nil }
func (s *stubListStream) Context() context.Context     { return context.Background() }
func (s *stubListStream) SendMsg(m any) error          { return nil }
func (s *stubListStream) RecvMsg(m any) error          { return nil }

func (s *stubListStream) Recv() (*filer_pb.ListEntriesResponse, error) {
	if s.idx >= len(s.entries) {
		return nil, io.EOF
	}
	entry := s.entries[s.idx]
	s.idx++
	return &filer_pb.ListEntriesResponse{Entry: entry}, nil
}

func uploadEntry(uploadId, objectKey string) *filer_pb.Entry {
	return &filer_pb.Entry{
		Name:     uploadId,
		Extended: map[string][]byte{s3_constants.ExtMultipartObjectKey: []byte(objectKey)},
	}
}

func completedEntry(uploadId string) *filer_pb.Entry {
	return &filer_pb.Entry{
		Extended: map[string][]byte{s3_constants.SeaweedFSUploadId: []byte(uploadId)},
	}
}

func TestUploadCompletedVersionPaths(t *testing.T) {
	for _, tc := range []struct {
		key       string
		objectDir string
		name      string
	}{
		{key: "obj", objectDir: "/buckets/b", name: "obj"},
		{key: "dir/obj", objectDir: "/buckets/b/dir", name: "obj"},
		{key: "dir/", objectDir: "/buckets/b/dir", name: "dir"},
	} {
		t.Run(tc.key, func(t *testing.T) {
			svc := &stubFilerSvc{
				lookupErr: filer_pb.ErrNotFound,
				listDir:   tc.objectDir + "/" + tc.name + s3_constants.VersionsFolder,
				listResp:  []*filer_pb.Entry{completedEntry("up1")},
			}
			c := &commandS3CleanUploads{}
			completed, err := c.uploadCompleted(stubFilerClient{client: svc}, "/buckets/b", uploadEntry("up1", tc.key))
			if err != nil || !completed {
				t.Fatalf("got (%v, %v), want (true, nil)", completed, err)
			}
			if svc.lookupReq.Directory != tc.objectDir || svc.lookupReq.Name != tc.name {
				t.Fatalf("lookup = %s/%s, want %s/%s", svc.lookupReq.Directory, svc.lookupReq.Name, tc.objectDir, tc.name)
			}
		})
	}
}

func TestUploadCompleted(t *testing.T) {
	c := &commandS3CleanUploads{}

	t.Run("object entry carries the upload id", func(t *testing.T) {
		fc := stubFilerClient{client: &stubFilerSvc{
			lookupResp: &filer_pb.LookupDirectoryEntryResponse{Entry: completedEntry("up1")},
		}}
		completed, err := c.uploadCompleted(fc, "/buckets/b", uploadEntry("up1", "obj"))
		if err != nil || !completed {
			t.Fatalf("got (%v, %v), want (true, nil)", completed, err)
		}
	})

	t.Run("object entry has a different upload id", func(t *testing.T) {
		fc := stubFilerClient{client: &stubFilerSvc{
			lookupResp: &filer_pb.LookupDirectoryEntryResponse{Entry: completedEntry("up2")},
		}}
		completed, err := c.uploadCompleted(fc, "/buckets/b", uploadEntry("up1", "obj"))
		if err != nil || completed {
			t.Fatalf("got (%v, %v), want (false, nil)", completed, err)
		}
	})

	t.Run("a version file carries the upload id", func(t *testing.T) {
		fc := stubFilerClient{client: &stubFilerSvc{
			lookupErr: filer_pb.ErrNotFound,
			listResp: []*filer_pb.Entry{
				completedEntry("up0"),
				completedEntry("up1"),
			},
		}}
		completed, err := c.uploadCompleted(fc, "/buckets/b", uploadEntry("up1", "obj"))
		if err != nil || !completed {
			t.Fatalf("got (%v, %v), want (true, nil)", completed, err)
		}
	})

	t.Run("no object and no versions", func(t *testing.T) {
		fc := stubFilerClient{client: &stubFilerSvc{
			lookupErr: filer_pb.ErrNotFound,
		}}
		completed, err := c.uploadCompleted(fc, "/buckets/b", uploadEntry("up1", "obj"))
		if err != nil || completed {
			t.Fatalf("got (%v, %v), want (false, nil)", completed, err)
		}
	})

	t.Run("upload dir without an object key", func(t *testing.T) {
		completed, err := c.uploadCompleted(stubFilerClient{}, "/buckets/b", &filer_pb.Entry{Name: "up1"})
		if err != nil || completed {
			t.Fatalf("got (%v, %v), want (false, nil)", completed, err)
		}
	})

	t.Run("lookup error propagates", func(t *testing.T) {
		fc := stubFilerClient{client: &stubFilerSvc{
			lookupErr: errors.New("store unavailable"),
		}}
		completed, err := c.uploadCompleted(fc, "/buckets/b", uploadEntry("up1", "obj"))
		if err == nil || completed {
			t.Fatalf("got (%v, %v), want (false, error)", completed, err)
		}
	})

	t.Run("versions list not found means no completed object", func(t *testing.T) {
		fc := stubFilerClient{client: &stubFilerSvc{
			lookupErr: filer_pb.ErrNotFound,
			listErr:   filer_pb.ErrNotFound,
		}}
		completed, err := c.uploadCompleted(fc, "/buckets/b", uploadEntry("up1", "obj"))
		if err != nil || completed {
			t.Fatalf("got (%v, %v), want (false, nil)", completed, err)
		}
	})

	t.Run("trailing slash key resolves inside its directory", func(t *testing.T) {
		svc := &stubFilerSvc{
			lookupResp: &filer_pb.LookupDirectoryEntryResponse{Entry: completedEntry("up1")},
		}
		completed, err := c.uploadCompleted(stubFilerClient{client: svc}, "/buckets/b", uploadEntry("up1", "dir/"))
		if err != nil || !completed {
			t.Fatalf("got (%v, %v), want (true, nil)", completed, err)
		}
		if svc.lookupReq.Directory != "/buckets/b/dir" || svc.lookupReq.Name != "dir" {
			t.Fatalf("lookup = %s/%s, want /buckets/b/dir/dir", svc.lookupReq.Directory, svc.lookupReq.Name)
		}
	})
}