[#602] Fix removal non-empty bucket

Signed-off-by: Denis Kirillov <denis@nspcc.ru>
This commit is contained in:
Denis Kirillov 2022-07-21 12:05:47 +03:00 committed by Alex Vanin
parent 75afc0e4a6
commit 85eacdb970
4 changed files with 40 additions and 4 deletions

View file

@ -14,6 +14,20 @@ const (
emptyVersion = "" emptyVersion = ""
) )
func TestDeleteBucket(t *testing.T) {
tc := prepareHandlerContext(t)
bktName, objName := "bucket-for-removal", "object-to-delete"
_, objInfo := createVersionedBucketAndObject(t, tc, bktName, objName)
deleteMarker := deleteObject(t, tc, bktName, objName, emptyVersion)
deleteBucket(t, tc, bktName, http.StatusConflict)
deleteObject(t, tc, bktName, objName, objInfo.Version())
deleteBucket(t, tc, bktName, http.StatusConflict)
deleteObject(t, tc, bktName, objName, deleteMarker)
deleteBucket(t, tc, bktName, http.StatusNoContent)
}
func TestDeleteObject(t *testing.T) { func TestDeleteObject(t *testing.T) {
tc := prepareHandlerContext(t) tc := prepareHandlerContext(t)
@ -214,6 +228,12 @@ func deleteObject(t *testing.T, tc *handlerContext, bktName, objName, version st
return w.Header().Get(api.AmzVersionID) return w.Header().Get(api.AmzVersionID)
} }
func deleteBucket(t *testing.T, tc *handlerContext, bktName string, code int) {
w, r := prepareTestRequest(t, bktName, "", nil)
tc.Handler().DeleteBucketHandler(w, r)
assertStatus(t, w, code)
}
func checkNotFound(t *testing.T, tc *handlerContext, bktName, objName, version string) { func checkNotFound(t *testing.T, tc *handlerContext, bktName, objName, version string) {
query := make(url.Values) query := make(url.Values)
query.Add(api.QueryVersionID, version) query.Add(api.QueryVersionID, version)

View file

@ -593,11 +593,11 @@ func (n *layer) ResolveBucket(ctx context.Context, name string) (cid.ID, error)
} }
func (n *layer) DeleteBucket(ctx context.Context, p *DeleteBucketParams) error { func (n *layer) DeleteBucket(ctx context.Context, p *DeleteBucketParams) error {
objects, _, err := n.getLatestObjectsVersions(ctx, allObjectParams{Bucket: p.BktInfo, MaxKeys: 1}) nodeVersions, err := n.bucketNodeVersions(ctx, p.BktInfo, "")
if err != nil { if err != nil {
return err return err
} }
if len(objects) != 0 { if len(nodeVersions) != 0 {
return errors.GetAPIError(errors.ErrBucketNotEmpty) return errors.GetAPIError(errors.ErrBucketNotEmpty)
} }

View file

@ -16,6 +16,7 @@ import (
cid "github.com/nspcc-dev/neofs-sdk-go/container/id" cid "github.com/nspcc-dev/neofs-sdk-go/container/id"
"github.com/nspcc-dev/neofs-sdk-go/object" "github.com/nspcc-dev/neofs-sdk-go/object"
oid "github.com/nspcc-dev/neofs-sdk-go/object/id" oid "github.com/nspcc-dev/neofs-sdk-go/object/id"
"github.com/nspcc-dev/neofs-sdk-go/session"
"github.com/nspcc-dev/neofs-sdk-go/user" "github.com/nspcc-dev/neofs-sdk-go/user"
) )
@ -96,6 +97,12 @@ func (t *TestNeoFS) CreateContainer(_ context.Context, prm PrmContainerCreate) (
return id, nil return id, nil
} }
func (t *TestNeoFS) DeleteContainer(_ context.Context, cnrID cid.ID, _ *session.Container) error {
delete(t.containers, cnrID.EncodeToString())
return nil
}
func (t *TestNeoFS) Container(_ context.Context, id cid.ID) (*container.Container, error) { func (t *TestNeoFS) Container(_ context.Context, id cid.ID) (*container.Container, error) {
for k, v := range t.containers { for k, v := range t.containers {
if k == id.EncodeToString() { if k == id.EncodeToString() {

View file

@ -553,7 +553,7 @@ func (n *layer) initWorkerPool(ctx context.Context, size int, p allObjectParams,
return objCh, nil return objCh, nil
} }
func (n *layer) getAllObjectsVersions(ctx context.Context, bkt *data.BucketInfo, prefix, delimiter string) (map[string][]*data.ExtendedObjectInfo, error) { func (n *layer) bucketNodeVersions(ctx context.Context, bkt *data.BucketInfo, prefix string) ([]*data.NodeVersion, error) {
var err error var err error
cacheKey := cache.CreateObjectsListCacheKey(bkt.CID, prefix, false) cacheKey := cache.CreateObjectsListCacheKey(bkt.CID, prefix, false)
@ -562,13 +562,22 @@ func (n *layer) getAllObjectsVersions(ctx context.Context, bkt *data.BucketInfo,
if nodeVersions == nil { if nodeVersions == nil {
nodeVersions, err = n.treeService.GetAllVersionsByPrefix(ctx, bkt.CID, prefix) nodeVersions, err = n.treeService.GetAllVersionsByPrefix(ctx, bkt.CID, prefix)
if err != nil { if err != nil {
return nil, err return nil, fmt.Errorf("get all versions from tree service: %w", err)
} }
if err = n.listsCache.PutVersions(cacheKey, nodeVersions); err != nil { if err = n.listsCache.PutVersions(cacheKey, nodeVersions); err != nil {
n.log.Error("couldn't cache list of objects", zap.Error(err)) n.log.Error("couldn't cache list of objects", zap.Error(err))
} }
} }
return nodeVersions, nil
}
func (n *layer) getAllObjectsVersions(ctx context.Context, bkt *data.BucketInfo, prefix, delimiter string) (map[string][]*data.ExtendedObjectInfo, error) {
nodeVersions, err := n.bucketNodeVersions(ctx, bkt, prefix)
if err != nil {
return nil, err
}
versions := make(map[string][]*data.ExtendedObjectInfo, len(nodeVersions)) versions := make(map[string][]*data.ExtendedObjectInfo, len(nodeVersions))
for _, nodeVersion := range nodeVersions { for _, nodeVersion := range nodeVersions {