Merge pull request #1113 from versity/fix/listobjects-dir-objs

fix: Fixes ListObjects direcotry objects listing issue. Fixes DeleteO…

Author: benmcclelland

backend/posix/posix.go

@@ -3212,7 +3212,10 @@ func (p *Posix) removeParents(bucket, object string) {
 	// this with a special attribute to indicate these. stop
 	// at either the bucket or the first parent we encounter
 	// with the attribute, whichever comes first.
-	objPath := object
+
+	// Remove the last path separator for the directory objects
+	// to correctly detect the parent in the loop
+	objPath := strings.TrimSuffix(object, "/")
 	for {
 		parent := filepath.Dir(objPath)
 		if parent == string(filepath.Separator) || parent == "." {

backend/walk.go

@@ -88,6 +88,7 @@ func Walk(ctx context.Context, fileSystem fs.FS, prefix, delimiter, marker strin
 		// so we can skip a directory without an early return
 		var skipflag error
 		if d.IsDir() {
+			fmt.Println("path: ", path)
 			// If prefix is defined and the directory does not match prefix,
 			// do not descend into the directory because nothing will
 			// match this prefix. Make sure to append the / at the end of
@@ -108,13 +109,7 @@ func Walk(ctx context.Context, fileSystem fs.FS, prefix, delimiter, marker strin
 				strings.Contains(strings.TrimPrefix(path+"/", prefix), delimiter) {
 				skipflag = fs.SkipDir
 			} else {
-				// TODO: can we do better here rather than a second readdir
-				// per directory?
-				ents, err := fs.ReadDir(fileSystem, path)
-				if err != nil {
-					return fmt.Errorf("readdir %q: %w", path, err)
-				}
-				if len(ents) == 0 && delimiter == "" {
+				if delimiter == "" {
 					dirobj, err := getObj(path+"/", d)
 					if err == ErrSkipObj {
 						return skipflag
@@ -127,6 +122,13 @@ func Walk(ctx context.Context, fileSystem fs.FS, prefix, delimiter, marker strin
 					return skipflag
 				}
 
+				// TODO: can we do better here rather than a second readdir
+				// per directory?
+				ents, err := fs.ReadDir(fileSystem, path)
+				if err != nil {
+					return fmt.Errorf("readdir %q: %w", path, err)
+				}
+
 				if len(ents) != 0 {
 					return skipflag
 				}

tests/integration/group-tests.go

@@ -213,6 +213,7 @@ func TestListObjects(s *S3Conf) {
 	ListObjects_max_keys_none(s)
 	ListObjects_marker_not_from_obj_list(s)
 	ListObjects_list_all_objs(s)
+	ListObjects_nested_dir_file_objs(s)
 	//TODO: remove the condition after implementing checksums in azure
 	if !s.azureTests {
 		ListObjects_with_checksum(s)
@@ -246,6 +247,7 @@ func TestDeleteObject(s *S3Conf) {
 	DeleteObject_non_existing_object(s)
 	DeleteObject_directory_object_noslash(s)
 	DeleteObject_non_existing_dir_object(s)
+	DeleteObject_directory_object(s)
 	DeleteObject_success(s)
 	DeleteObject_success_status_code(s)
 }
@@ -876,6 +878,7 @@ func GetIntTests() IntTests {
 		"ListObjects_max_keys_none":                                               ListObjects_max_keys_none,
 		"ListObjects_marker_not_from_obj_list":                                    ListObjects_marker_not_from_obj_list,
 		"ListObjects_list_all_objs":                                               ListObjects_list_all_objs,
+		"ListObjects_nested_dir_file_objs":                                        ListObjects_nested_dir_file_objs,
 		"ListObjects_with_checksum":                                               ListObjects_with_checksum,
 		"ListObjectsV2_start_after":                                               ListObjectsV2_start_after,
 		"ListObjectsV2_both_start_after_and_continuation_token":                   ListObjectsV2_both_start_after_and_continuation_token,
@@ -892,6 +895,7 @@ func GetIntTests() IntTests {
 		"DeleteObject_directory_object_noslash":                                   DeleteObject_directory_object_noslash,
 		"DeleteObject_name_too_long":                                              DeleteObject_name_too_long,
 		"DeleteObject_non_existing_dir_object":                                    DeleteObject_non_existing_dir_object,
+		"DeleteObject_directory_object":                                           DeleteObject_directory_object,
 		"DeleteObject_success":                                                    DeleteObject_success,
 		"DeleteObject_success_status_code":                                        DeleteObject_success_status_code,
 		"DeleteObjects_empty_input":                                               DeleteObjects_empty_input,

tests/integration/tests.go

@@ -4771,6 +4771,44 @@ func ListObjects_list_all_objs(s *S3Conf) error {
 	})
 }
 
+func ListObjects_nested_dir_file_objs(s *S3Conf) error {
+	testName := "ListObjects_nested_dir_file_objs"
+	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
+		contents, err := putObjects(s3client, []string{"foo/bar/", "foo/bar/baz", "foo/bar/quxx"}, bucket)
+		if err != nil {
+			return err
+		}
+
+		ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
+		res, err := s3client.ListObjects(ctx, &s3.ListObjectsInput{
+			Bucket: &bucket,
+		})
+		cancel()
+		if err != nil {
+			return err
+		}
+
+		if !compareObjects(contents, res.Contents) {
+			return fmt.Errorf("expected the objects list to be %+v, instead got %+v", contents, res.Contents)
+		}
+
+		// Clean up the nested objects to avoid `ErrDirectoryNotEmpty` error on teardown
+		for _, obj := range []string{"foo/bar/baz", "foo/bar/quxx"} {
+			ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
+			_, err := s3client.DeleteObject(ctx, &s3.DeleteObjectInput{
+				Bucket: &bucket,
+				Key:    &obj,
+			})
+			cancel()
+			if err != nil {
+				return err
+			}
+		}
+
+		return nil
+	})
+}
+
 func ListObjectsV2_start_after(s *S3Conf) error {
 	testName := "ListObjectsV2_start_after"
 	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
@@ -5391,6 +5429,25 @@ func DeleteObject_non_existing_dir_object(s *S3Conf) error {
 	})
 }
 
+func DeleteObject_directory_object(s *S3Conf) error {
+	testName := "DeleteObject_directory_object"
+	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
+		obj := "foo/bar/"
+		_, err := putObjects(s3client, []string{obj}, bucket)
+		if err != nil {
+			return err
+		}
+
+		ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
+		_, err = s3client.DeleteObject(ctx, &s3.DeleteObjectInput{
+			Bucket: &bucket,
+			Key:    &obj,
+		})
+		cancel()
+		return err
+	})
+}
+
 func DeleteObject_success(s *S3Conf) error {
 	testName := "DeleteObject_success"
 	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {