diff --git a/backend/posix/posix.go b/backend/posix/posix.go
index a331a3ff..0e45d369 100644
--- a/backend/posix/posix.go
+++ b/backend/posix/posix.go
@@ -152,7 +152,7 @@ func New(rootdir string, meta meta.MetadataStorer, opts PosixOpts) (*Posix, erro
 			return nil, fmt.Errorf("versioning path should be a directory")
 		}
 
-		fmt.Printf("bucket versioning enabled with directory: %v\n", verioningdirAbs)
+		fmt.Printf("Bucket versioning enabled with directory: %v\n", verioningdirAbs)
 	}
 
 	return &Posix{
@@ -500,6 +500,16 @@ func (p *Posix) GetBucketVersioning(_ context.Context, bucket string) (*s3.GetBu
 	return &s3.GetBucketVersioningOutput{}, nil
 }
 
+// Returns the specified bucket versioning status
+func (p *Posix) isBucketVersioningEnabled(ctx context.Context, bucket string) (bool, error) {
+	res, err := p.GetBucketVersioning(ctx, bucket)
+	if err != nil {
+		return false, err
+	}
+
+	return res.Status == types.BucketVersioningStatusEnabled, nil
+}
+
 // Generates the object version path in the versioning directory
 func (p *Posix) genObjVersionPath(bucket, key string) string {
 	return filepath.Join(p.versioningDir, bucket, genObjVersionKey(key))
@@ -522,7 +532,7 @@ func (p *Posix) createObjVersion(bucket, key string, size int64, acc auth.Accoun
 
 	var versionId string
 	data, err := p.meta.RetrieveAttribute(bucket, key, versionIdKey)
-	if err != nil {
+	if err == nil {
 		versionId = string(data)
 	} else {
 		versionId = ulid.Make().String()
@@ -651,7 +661,7 @@ func (p *Posix) isObjDeleteMarker(bucket, object string) (bool, error) {
 // Converts the file to object version. Finds all the object versions,
 // delete markers from the versioning directory and returns
 func (p *Posix) fileToObjVersions(bucket string) backend.GetVersionsFunc {
-	return func(path, versionIdMarker string, availableObjCount int, d fs.DirEntry) (*backend.ObjVersionFuncResult, error) {
+	return func(path, versionIdMarker string, pastVersionIdMarker *bool, availableObjCount int, d fs.DirEntry) (*backend.ObjVersionFuncResult, error) {
 		var objects []types.ObjectVersion
 		var delMarkers []types.DeleteMarkerEntry
 		// if the number of available objects is 0, return truncated response
@@ -663,8 +673,44 @@ func (p *Posix) fileToObjVersions(bucket string) backend.GetVersionsFunc {
 			}, nil
 		}
 		if d.IsDir() {
-			//TODO: directory objects can't have versions, but they are listed in object versions result?
-			return nil, backend.ErrSkipObj
+			// directory object only happens if directory empty
+			// check to see if this is a directory object by checking etag
+			etagBytes, err := p.meta.RetrieveAttribute(bucket, path, etagkey)
+			if errors.Is(err, meta.ErrNoSuchKey) || errors.Is(err, fs.ErrNotExist) {
+				return nil, backend.ErrSkipObj
+			}
+			if err != nil {
+				return nil, fmt.Errorf("get etag: %w", err)
+			}
+			etag := string(etagBytes)
+
+			fi, err := d.Info()
+			if errors.Is(err, fs.ErrNotExist) {
+				return nil, backend.ErrSkipObj
+			}
+			if err != nil {
+				return nil, fmt.Errorf("get fileinfo: %w", err)
+			}
+
+			key := path + "/"
+			// Directory objects don't contain data
+			size := int64(0)
+			versionId := "null"
+
+			objects = append(objects, types.ObjectVersion{
+				ETag:         &etag,
+				Key:          &key,
+				LastModified: backend.GetTimePtr(fi.ModTime()),
+				IsLatest:     getBoolPtr(true),
+				Size:         &size,
+				VersionId:    &versionId,
+			})
+
+			return &backend.ObjVersionFuncResult{
+				ObjectVersions: objects,
+				DelMarkers:     delMarkers,
+				Truncated:      availableObjCount == 1,
+			}, nil
 		}
 
 		// file object, get object info and fill out object data
@@ -685,47 +731,58 @@ func (p *Posix) fileToObjVersions(bucket string) backend.GetVersionsFunc {
 		if err == nil {
 			versionId = string(versionIdBytes)
 		}
-
-		fi, err := d.Info()
-		if errors.Is(err, fs.ErrNotExist) {
-			return nil, backend.ErrSkipObj
-		}
-		if err != nil {
-			return nil, fmt.Errorf("get fileinfo: %w", err)
+		if versionId == versionIdMarker {
+			*pastVersionIdMarker = true
 		}
+		if *pastVersionIdMarker {
+			fi, err := d.Info()
+			if errors.Is(err, fs.ErrNotExist) {
+				return nil, backend.ErrSkipObj
+			}
+			if err != nil {
+				return nil, fmt.Errorf("get fileinfo: %w", err)
+			}
 
-		size := fi.Size()
+			size := fi.Size()
 
-		isDel, err := p.isObjDeleteMarker(bucket, path)
-		if err != nil {
-			return nil, err
-		}
+			isDel, err := p.isObjDeleteMarker(bucket, path)
+			if err != nil {
+				return nil, err
+			}
 
-		if isDel {
-			delMarkers = append(delMarkers, types.DeleteMarkerEntry{
-				IsLatest:     getBoolPtr(true),
-				VersionId:    &versionId,
-				LastModified: backend.GetTimePtr(fi.ModTime()),
-				Key:          &path,
-			})
-		} else {
-			objects = append(objects, types.ObjectVersion{
-				ETag:         &etag,
-				Key:          &path,
-				LastModified: backend.GetTimePtr(fi.ModTime()),
-				Size:         &size,
-				VersionId:    &versionId,
-				IsLatest:     getBoolPtr(true),
-			})
+			if isDel {
+				delMarkers = append(delMarkers, types.DeleteMarkerEntry{
+					IsLatest:     getBoolPtr(true),
+					VersionId:    &versionId,
+					LastModified: backend.GetTimePtr(fi.ModTime()),
+					Key:          &path,
+				})
+			} else {
+				objects = append(objects, types.ObjectVersion{
+					ETag:         &etag,
+					Key:          &path,
+					LastModified: backend.GetTimePtr(fi.ModTime()),
+					Size:         &size,
+					VersionId:    &versionId,
+					IsLatest:     getBoolPtr(true),
+				})
+			}
+
+			availableObjCount--
+			if availableObjCount == 0 {
+				return &backend.ObjVersionFuncResult{
+					ObjectVersions:      objects,
+					DelMarkers:          delMarkers,
+					Truncated:           true,
+					NextVersionIdMarker: versionId,
+				}, nil
+			}
 		}
 
-		availableObjCount--
-		if availableObjCount == 0 {
+		if !p.versioningEnabled() {
 			return &backend.ObjVersionFuncResult{
-				ObjectVersions:      objects,
-				DelMarkers:          delMarkers,
-				Truncated:           true,
-				NextVersionIdMarker: versionId,
+				ObjectVersions: objects,
+				DelMarkers:     delMarkers,
 			}, nil
 		}
 
@@ -763,6 +820,13 @@ func (p *Posix) fileToObjVersions(bucket string) backend.GetVersionsFunc {
 			versionId := f.Name()
 			size := f.Size()
 
+			if !*pastVersionIdMarker {
+				if versionId == versionIdMarker {
+					*pastVersionIdMarker = true
+				}
+				continue
+			}
+
 			etagBytes, err := p.meta.RetrieveAttribute(versionPath, versionId, etagkey)
 			if errors.Is(err, fs.ErrNotExist) {
 				return nil, backend.ErrSkipObj
@@ -782,8 +846,9 @@ func (p *Posix) fileToObjVersions(bucket string) backend.GetVersionsFunc {
 			if isDel {
 				delMarkers = append(delMarkers, types.DeleteMarkerEntry{
 					VersionId:    &versionId,
-					LastModified: backend.GetTimePtr(fi.ModTime()),
+					LastModified: backend.GetTimePtr(f.ModTime()),
 					Key:          &path,
+					IsLatest:     getBoolPtr(false),
 				})
 			} else {
 				objects = append(objects, types.ObjectVersion{
@@ -792,6 +857,7 @@ func (p *Posix) fileToObjVersions(bucket string) backend.GetVersionsFunc {
 					LastModified: backend.GetTimePtr(f.ModTime()),
 					Size:         &size,
 					VersionId:    &versionId,
+					IsLatest:     getBoolPtr(false),
 				})
 			}
 
@@ -1870,6 +1936,11 @@ func (p *Posix) PutObject(ctx context.Context, po *s3.PutObjectInput) (s3respons
 		}, nil
 	}
 
+	vEnabled, err := p.isBucketVersioningEnabled(ctx, *po.Bucket)
+	if err != nil {
+		return s3response.PutObjectOutput{}, err
+	}
+
 	// object is file
 	d, err := os.Stat(name)
 	if err == nil && d.IsDir() {
@@ -1877,7 +1948,7 @@ func (p *Posix) PutObject(ctx context.Context, po *s3.PutObjectInput) (s3respons
 	}
 
 	// if the versioninng is enabled first create the file object version
-	if p.versioningEnabled() && err == nil {
+	if p.versioningEnabled() && vEnabled && err == nil {
 		_, err := p.createObjVersion(*po.Bucket, *po.Key, d.Size(), acct)
 		if err != nil {
 			return s3response.PutObjectOutput{}, fmt.Errorf("create object version: %w", err)
@@ -1987,7 +2058,7 @@ func (p *Posix) PutObject(ctx context.Context, po *s3.PutObjectInput) (s3respons
 
 	// if the versioning is enabled, generate a new versionID for the object
 	var versionID string
-	if p.versioningEnabled() {
+	if p.versioningEnabled() && vEnabled {
 		versionID = ulid.Make().String()
 
 		if err := p.meta.StoreAttribute(*po.Bucket, *po.Key, versionIdKey, []byte(versionID)); err != nil {
@@ -2011,6 +2082,7 @@ func (p *Posix) DeleteObject(ctx context.Context, input *s3.DeleteObjectInput) (
 
 	bucket := *input.Bucket
 	object := *input.Key
+	isDir := strings.HasSuffix(object, "/")
 
 	_, err := os.Stat(bucket)
 	if errors.Is(err, fs.ErrNotExist) {
@@ -2022,17 +2094,36 @@ func (p *Posix) DeleteObject(ctx context.Context, input *s3.DeleteObjectInput) (
 
 	objpath := filepath.Join(bucket, object)
 
-	if p.versioningEnabled() {
-		if *input.VersionId == "" {
+	vEnabled, err := p.isBucketVersioningEnabled(ctx, bucket)
+	if err != nil {
+		return nil, err
+	}
+
+	// Directory objects can't have versions
+	if !isDir && p.versioningEnabled() && vEnabled {
+		if getString(input.VersionId) == "" {
 			// if the versionId is not specified, make the current version a delete marker
-			_, err := os.Stat(objpath)
+			fi, err := os.Stat(objpath)
 			if err != nil {
 				return nil, s3err.GetAPIError(s3err.ErrNoSuchKey)
 			}
 
+			acct, ok := ctx.Value("account").(auth.Account)
+			if !ok {
+				acct = auth.Account{}
+			}
+
+			// Creates a new version in the versioning directory
+			_, err = p.createObjVersion(bucket, object, fi.Size(), acct)
+			if err != nil {
+				return nil, err
+			}
+
+			// Mark the object as a delete marker
 			if err := p.meta.StoreAttribute(bucket, object, deleteMarkerKey, []byte{}); err != nil {
 				return nil, fmt.Errorf("set delete marker: %w", err)
 			}
+			// Generate & set a unique versionId for the delete marker
 			versionId := ulid.Make().String()
 			if err := p.meta.StoreAttribute(bucket, object, versionIdKey, []byte(versionId)); err != nil {
 				return nil, fmt.Errorf("set versionId: %w", err)
@@ -2042,7 +2133,6 @@ func (p *Posix) DeleteObject(ctx context.Context, input *s3.DeleteObjectInput) (
 				VersionId: &versionId,
 			}, nil
 		} else {
-			delMarker := true
 			versionPath := p.genObjVersionPath(bucket, object)
 
 			vId, err := p.meta.RetrieveAttribute(bucket, object, versionIdKey)
@@ -2054,7 +2144,12 @@ func (p *Posix) DeleteObject(ctx context.Context, input *s3.DeleteObjectInput) (
 				// if the specified VersionId is the same as in the latest version,
 				// remove the latest version, find the latest version from the versioning
 				// directory and move to the place of the deleted object, to make it the latest
-				err := os.Remove(objpath)
+
+				isDelMarker, err := p.isObjDeleteMarker(bucket, object)
+				if err != nil {
+					return nil, err
+				}
+				err = os.Remove(objpath)
 				if err != nil {
 					return nil, fmt.Errorf("remove obj version: %w", err)
 				}
@@ -2062,7 +2157,7 @@ func (p *Posix) DeleteObject(ctx context.Context, input *s3.DeleteObjectInput) (
 				ents, err := os.ReadDir(versionPath)
 				if errors.Is(err, fs.ErrNotExist) {
 					return &s3.DeleteObjectOutput{
-						DeleteMarker: &delMarker,
+						DeleteMarker: &isDelMarker,
 						VersionId:    input.VersionId,
 					}, nil
 				}
@@ -2072,7 +2167,7 @@ func (p *Posix) DeleteObject(ctx context.Context, input *s3.DeleteObjectInput) (
 
 				if len(ents) == 0 {
 					return &s3.DeleteObjectOutput{
-						DeleteMarker: &delMarker,
+						DeleteMarker: &isDelMarker,
 						VersionId:    input.VersionId,
 					}, nil
 				}
@@ -2128,15 +2223,17 @@ func (p *Posix) DeleteObject(ctx context.Context, input *s3.DeleteObjectInput) (
 				}
 
 				return &s3.DeleteObjectOutput{
-					DeleteMarker: &delMarker,
+					DeleteMarker: &isDelMarker,
 					VersionId:    input.VersionId,
 				}, nil
 			}
 
+			isDelMarker, _ := p.isObjDeleteMarker(versionPath, *input.VersionId)
+
 			err = os.Remove(filepath.Join(versionPath, *input.VersionId))
 			if errors.Is(err, fs.ErrNotExist) {
 				return &s3.DeleteObjectOutput{
-					DeleteMarker: &delMarker,
+					DeleteMarker: &isDelMarker,
 					VersionId:    input.VersionId,
 				}, nil
 			}
@@ -2145,11 +2242,10 @@ func (p *Posix) DeleteObject(ctx context.Context, input *s3.DeleteObjectInput) (
 			}
 
 			return &s3.DeleteObjectOutput{
-				DeleteMarker: &delMarker,
+				DeleteMarker: &isDelMarker,
 				VersionId:    input.VersionId,
 			}, nil
 		}
-
 	}
 
 	fi, err := os.Stat(objpath)
@@ -2199,7 +2295,7 @@ func (p *Posix) removeParents(bucket, object string) error {
 	for {
 		parent := filepath.Dir(objPath)
 
-		if parent == "." {
+		if parent == string(filepath.Separator) || parent == "." {
 			// stop removing parents if we hit the bucket directory.
 			break
 		}
@@ -2228,15 +2324,22 @@ func (p *Posix) DeleteObjects(ctx context.Context, input *s3.DeleteObjectsInput)
 	for _, obj := range input.Delete.Objects {
 		//TODO: Make the delete operation concurrent
 		res, err := p.DeleteObject(ctx, &s3.DeleteObjectInput{
-			Bucket: input.Bucket,
-			Key:    obj.Key,
+			Bucket:    input.Bucket,
+			Key:       obj.Key,
+			VersionId: obj.VersionId,
 		})
 		if err == nil {
-			delResult = append(delResult, types.DeletedObject{
+			delEntity := types.DeletedObject{
 				Key:          obj.Key,
-				VersionId:    res.VersionId,
 				DeleteMarker: res.DeleteMarker,
-			})
+			}
+			if delEntity.DeleteMarker != nil && *delEntity.DeleteMarker {
+				delEntity.DeleteMarkerVersionId = res.VersionId
+			} else {
+				delEntity.VersionId = res.VersionId
+			}
+
+			delResult = append(delResult, delEntity)
 		} else {
 			serr, ok := err.(s3err.APIError)
 			if ok {
@@ -2310,6 +2413,9 @@ func (p *Posix) GetObject(_ context.Context, input *s3.GetObjectInput) (*s3.GetO
 
 	fi, err := os.Stat(objPath)
 	if errors.Is(err, fs.ErrNotExist) {
+		if *input.VersionId != "" {
+			return nil, s3err.GetAPIError(s3err.ErrInvalidVersionId)
+		}
 		return nil, s3err.GetAPIError(s3err.ErrNoSuchKey)
 	}
 	if errors.Is(err, syscall.ENAMETOOLONG) {
@@ -2508,6 +2614,14 @@ func (p *Posix) HeadObject(ctx context.Context, input *s3.HeadObjectInput) (*s3.
 		}, nil
 	}
 
+	_, err := os.Stat(bucket)
+	if errors.Is(err, fs.ErrNotExist) {
+		return nil, s3err.GetAPIError(s3err.ErrNoSuchBucket)
+	}
+	if err != nil {
+		return nil, fmt.Errorf("stat bucket: %w", err)
+	}
+
 	if *input.VersionId != "" {
 		vId, err := p.meta.RetrieveAttribute(bucket, object, versionIdKey)
 		if errors.Is(err, fs.ErrNotExist) {
@@ -2527,14 +2641,6 @@ func (p *Posix) HeadObject(ctx context.Context, input *s3.HeadObjectInput) (*s3.
 		}
 	}
 
-	_, err := os.Stat(bucket)
-	if errors.Is(err, fs.ErrNotExist) {
-		return nil, s3err.GetAPIError(s3err.ErrNoSuchBucket)
-	}
-	if err != nil {
-		return nil, fmt.Errorf("stat bucket: %w", err)
-	}
-
 	objPath := filepath.Join(bucket, object)
 
 	fi, err := os.Stat(objPath)
@@ -2663,7 +2769,11 @@ func (p *Posix) CopyObject(ctx context.Context, input *s3.CopyObjectInput) (*s3.
 	copySource := cSplitted[0]
 	var srcVersionId string
 	if len(cSplitted) > 1 {
-		srcVersionId = cSplitted[1]
+		versionIdParts := strings.Split(cSplitted[1], "=")
+		if len(versionIdParts) != 2 || versionIdParts[0] != "versionId" {
+			return nil, s3err.GetAPIError(s3err.ErrInvalidRequest)
+		}
+		srcVersionId = versionIdParts[1]
 	}
 
 	srcBucket, srcObject, ok := strings.Cut(copySource, "/")
@@ -2681,9 +2791,28 @@ func (p *Posix) CopyObject(ctx context.Context, input *s3.CopyObjectInput) (*s3.
 		return nil, fmt.Errorf("stat bucket: %w", err)
 	}
 
+	vEnabled, err := p.isBucketVersioningEnabled(ctx, srcBucket)
+	if err != nil {
+		return nil, err
+	}
+
 	if srcVersionId != "" {
-		srcBucket = filepath.Join(p.versioningDir, srcBucket)
-		srcObject = filepath.Join(genObjVersionKey(srcObject), srcVersionId)
+		if !p.versioningEnabled() || !vEnabled {
+			return nil, s3err.GetAPIError(s3err.ErrInvalidVersionId)
+		}
+		vId, err := p.meta.RetrieveAttribute(srcBucket, srcObject, versionIdKey)
+		if errors.Is(err, fs.ErrNotExist) {
+			return nil, s3err.GetAPIError(s3err.ErrNoSuchKey)
+		}
+		if err != nil && !errors.Is(err, meta.ErrNoSuchKey) {
+			return nil, fmt.Errorf("get src object version id: %w", err)
+		}
+
+		if string(vId) != srcVersionId {
+			srcBucket = filepath.Join(p.versioningDir, srcBucket)
+			srcObject = filepath.Join(genObjVersionKey(srcObject), srcVersionId)
+		}
+
 	}
 
 	_, err = os.Stat(dstBucket)
@@ -2697,6 +2826,9 @@ func (p *Posix) CopyObject(ctx context.Context, input *s3.CopyObjectInput) (*s3.
 	objPath := filepath.Join(srcBucket, srcObject)
 	f, err := os.Open(objPath)
 	if errors.Is(err, fs.ErrNotExist) {
+		if p.versioningEnabled() && vEnabled {
+			return nil, s3err.GetAPIError(s3err.ErrNoSuchVersion)
+		}
 		return nil, s3err.GetAPIError(s3err.ErrNoSuchKey)
 	}
 	if errors.Is(err, syscall.ENAMETOOLONG) {
@@ -2865,6 +2997,12 @@ func (p *Posix) fileToObj(bucket string) backend.GetObjFunc {
 			}, nil
 		}
 
+		// If the object is a delete marker, skip
+		isDel, _ := p.isObjDeleteMarker(bucket, path)
+		if isDel {
+			return s3response.Object{}, backend.ErrSkipObj
+		}
+
 		// file object, get object info and fill out object data
 		etagBytes, err := p.meta.RetrieveAttribute(bucket, path, etagkey)
 		if errors.Is(err, fs.ErrNotExist) {
diff --git a/backend/walk.go b/backend/walk.go
index 8df944eb..0cf36f65 100644
--- a/backend/walk.go
+++ b/backend/walk.go
@@ -263,7 +263,7 @@ type ObjVersionFuncResult struct {
 	Truncated           bool
 }
 
-type GetVersionsFunc func(path, versionIdMarker string, availableObjCount int, d fs.DirEntry) (*ObjVersionFuncResult, error)
+type GetVersionsFunc func(path, versionIdMarker string, pastVersionIdMarker *bool, availableObjCount int, d fs.DirEntry) (*ObjVersionFuncResult, error)
 
 // WalkVersions walks the supplied fs.FS and returns results compatible with
 // ListObjectVersions action response
@@ -280,6 +280,8 @@ func WalkVersions(ctx context.Context, fileSystem fs.FS, prefix, delimiter, keyM
 	var nextVersionIdMarker string
 	var truncated bool
 
+	pastVersionIdMarker := versionIdMarker == ""
+
 	err := fs.WalkDir(fileSystem, ".", func(path string, d fs.DirEntry, err error) error {
 		if err != nil {
 			return err
@@ -295,6 +297,15 @@ func WalkVersions(ctx context.Context, fileSystem fs.FS, prefix, delimiter, keyM
 			return fs.SkipDir
 		}
 
+		if !pastMarker {
+			if path == keyMarker {
+				pastMarker = true
+			}
+			if path < keyMarker {
+				return nil
+			}
+		}
+
 		if d.IsDir() {
 			// If prefix is defined and the directory does not match prefix,
 			// do not descend into the directory because nothing will
@@ -309,18 +320,23 @@ func WalkVersions(ctx context.Context, fileSystem fs.FS, prefix, delimiter, keyM
 				return fs.SkipDir
 			}
 
-			// skip directory objects, as they can't have versions
-			return nil
-		}
-
-		if !pastMarker {
-			if path == keyMarker {
-				pastMarker = true
+			res, err := getObj(path, versionIdMarker, &pastVersionIdMarker, max-len(objects)-len(delMarkers)-len(cpmap), d)
+			if err == ErrSkipObj {
 				return nil
 			}
-			if path < keyMarker {
-				return nil
+			if err != nil {
+				return fmt.Errorf("directory to object %q: %w", path, err)
 			}
+			objects = append(objects, res.ObjectVersions...)
+			delMarkers = append(delMarkers, res.DelMarkers...)
+			if res.Truncated {
+				truncated = true
+				nextMarker = path
+				nextVersionIdMarker = res.NextVersionIdMarker
+				return fs.SkipAll
+			}
+
+			return nil
 		}
 
 		// If object doesn't have prefix, don't include in results.
@@ -331,7 +347,7 @@ func WalkVersions(ctx context.Context, fileSystem fs.FS, prefix, delimiter, keyM
 		if delimiter == "" {
 			// If no delimiter specified, then all files with matching
 			// prefix are included in results
-			res, err := getObj(path, versionIdMarker, max-len(objects)-len(delMarkers)-len(cpmap), d)
+			res, err := getObj(path, versionIdMarker, &pastVersionIdMarker, max-len(objects)-len(delMarkers)-len(cpmap), d)
 			if err == ErrSkipObj {
 				return nil
 			}
@@ -374,7 +390,7 @@ func WalkVersions(ctx context.Context, fileSystem fs.FS, prefix, delimiter, keyM
 		suffix := strings.TrimPrefix(path, prefix)
 		before, _, found := strings.Cut(suffix, delimiter)
 		if !found {
-			res, err := getObj(path, versionIdMarker, max-len(objects)-len(delMarkers)-len(cpmap), d)
+			res, err := getObj(path, versionIdMarker, &pastVersionIdMarker, max-len(objects)-len(delMarkers)-len(cpmap), d)
 			if err == ErrSkipObj {
 				return nil
 			}
diff --git a/cmd/versitygw/test.go b/cmd/versitygw/test.go
index b6c9ec24..b139f471 100644
--- a/cmd/versitygw/test.go
+++ b/cmd/versitygw/test.go
@@ -22,20 +22,21 @@ import (
 )
 
 var (
-	awsID           string
-	awsSecret       string
-	endpoint        string
-	prefix          string
-	dstBucket       string
-	partSize        int64
-	objSize         int64
-	concurrency     int
-	files           int
-	totalReqs       int
-	upload          bool
-	download        bool
-	pathStyle       bool
-	checksumDisable bool
+	awsID             string
+	awsSecret         string
+	endpoint          string
+	prefix            string
+	dstBucket         string
+	partSize          int64
+	objSize           int64
+	concurrency       int
+	files             int
+	totalReqs         int
+	upload            bool
+	download          bool
+	pathStyle         bool
+	checksumDisable   bool
+	versioningEnabled bool
 )
 
 func testCommand() *cli.Command {
@@ -87,6 +88,14 @@ func initTestCommands() []*cli.Command {
 			Usage:       "Tests the full flow of gateway.",
 			Description: `Runs all the available tests to test the full flow of the gateway.`,
 			Action:      getAction(integration.TestFullFlow),
+			Flags: []cli.Flag{
+				&cli.BoolFlag{
+					Name:        "versioning-enabled",
+					Usage:       "Test the bucket object versioning, if the versioning is enabled",
+					Destination: &versioningEnabled,
+					Aliases:     []string{"vs"},
+				},
+			},
 		},
 		{
 			Name:   "posix",
@@ -276,6 +285,9 @@ func getAction(tf testFunc) func(*cli.Context) error {
 		if debug {
 			opts = append(opts, integration.WithDebug())
 		}
+		if versioningEnabled {
+			opts = append(opts, integration.WithVersioningEnabled())
+		}
 
 		s := integration.NewS3Conf(opts...)
 		tf(s)
diff --git a/go.mod b/go.mod
index fc64b5c8..0e471212 100644
--- a/go.mod
+++ b/go.mod
@@ -16,6 +16,7 @@ require (
 	github.com/google/uuid v1.6.0
 	github.com/hashicorp/vault-client-go v0.4.3
 	github.com/nats-io/nats.go v1.37.0
+	github.com/oklog/ulid/v2 v2.1.0
 	github.com/pkg/xattr v0.4.10
 	github.com/segmentio/kafka-go v0.4.47
 	github.com/smira/go-statsd v1.3.3
@@ -45,7 +46,6 @@ require (
 	github.com/mitchellh/go-homedir v1.1.0 // indirect
 	github.com/nats-io/nkeys v0.4.7 // indirect
 	github.com/nats-io/nuid v1.0.1 // indirect
-	github.com/oklog/ulid/v2 v2.1.0 // indirect
 	github.com/pierrec/lz4/v4 v4.1.21 // indirect
 	github.com/pkg/browser v0.0.0-20240102092130-5ac0b6a4141c // indirect
 	github.com/ryanuber/go-glob v1.0.0 // indirect
diff --git a/runtests.sh b/runtests.sh
index abb63209..83d45116 100755
--- a/runtests.sh
+++ b/runtests.sh
@@ -5,9 +5,11 @@ rm -rf /tmp/gw
 mkdir /tmp/gw
 rm -rf /tmp/covdata
 mkdir /tmp/covdata
+rm -rf /tmp/versioningdir
+mkdir /tmp/versioningdir
 
 # run server in background
-GOCOVERDIR=/tmp/covdata ./versitygw -a user -s pass --iam-dir /tmp/gw posix /tmp/gw &
+GOCOVERDIR=/tmp/covdata ./versitygw -a user -s pass --iam-dir /tmp/gw posix --versioning-dir /tmp/versioningdir /tmp/gw &
 GW_PID=$!
 
 # wait a second for server to start up
@@ -21,7 +23,7 @@ fi
 
 # run tests
 # full flow tests
-if ! ./versitygw test -a user -s pass -e http://127.0.0.1:7070 full-flow; then
+if ! ./versitygw test -a user -s pass -e http://127.0.0.1:7070 full-flow -vs; then
 	echo "full flow tests failed"
 	kill $GW_PID
 	exit 1
diff --git a/s3api/controllers/base.go b/s3api/controllers/base.go
index 499cbb82..969fdd57 100644
--- a/s3api/controllers/base.go
+++ b/s3api/controllers/base.go
@@ -495,6 +495,15 @@ func (c S3ApiController) GetActions(ctx *fiber.Ctx) error {
 	utils.SetMetaHeaders(ctx, res.Metadata)
 	// Set other response headers
 	utils.SetResponseHeaders(ctx, hdrs)
+	// Set version id header
+	if getstring(res.VersionId) != "" {
+		utils.SetResponseHeaders(ctx, []utils.CustomHeader{
+			{
+				Key:   "x-amz-version-id",
+				Value: getstring(res.VersionId),
+			},
+		})
+	}
 
 	status := http.StatusOK
 	if acceptRange != "" {
@@ -2945,6 +2954,7 @@ func (c S3ApiController) HeadObject(ctx *fiber.Ctx) error {
 			Value: getstring(res.VersionId),
 		})
 	}
+
 	utils.SetResponseHeaders(ctx, headers)
 
 	return SendResponse(ctx, nil,
diff --git a/s3err/s3err.go b/s3err/s3err.go
index 53950075..14fa70ae 100644
--- a/s3err/s3err.go
+++ b/s3err/s3err.go
@@ -133,6 +133,7 @@ const (
 	ErrInvalidMetadataDirective
 	ErrKeyTooLong
 	ErrInvalidVersionId
+	ErrNoSuchVersion
 
 	// Non-AWS errors
 	ErrExistingObjectIsDirectory
@@ -531,6 +532,11 @@ var errorCodeResponse = map[ErrorCode]APIError{
 		Description:    "Your key is too long.",
 		HTTPStatusCode: http.StatusBadRequest,
 	},
+	ErrNoSuchVersion: {
+		Code:           "NoSuchVersion",
+		Description:    "The specified version does not exist.",
+		HTTPStatusCode: http.StatusNotFound,
+	},
 
 	// non aws errors
 	ErrExistingObjectIsDirectory: {
diff --git a/tests/integration/group-tests.go b/tests/integration/group-tests.go
index 12c7ef2d..ad4fd811 100644
--- a/tests/integration/group-tests.go
+++ b/tests/integration/group-tests.go
@@ -469,6 +469,9 @@ func TestFullFlow(s *S3Conf) {
 	TestGetObjectLegalHold(s)
 	TestWORMProtection(s)
 	TestAccessControl(s)
+	if s.versioningEnabled {
+		TestVersioning(s)
+	}
 }
 
 func TestPosix(s *S3Conf) {
@@ -503,6 +506,34 @@ func TestAccessControl(s *S3Conf) {
 	AccessControl_copy_object_with_starting_slash_for_user(s)
 }
 
+func TestVersioning(s *S3Conf) {
+	PutBucketVersioning_non_existing_bucket(s)
+	PutBucketVersioning_invalid_status(s)
+	PutBucketVersioning_success(s)
+	GetBucketVersioning_non_existing_bucket(s)
+	GetBucketVersioning_success(s)
+	Versioning_PutObject_success(s)
+	Versioning_CopyObject_success(s)
+	Versioning_CopyObject_non_existing_version_id(s)
+	Versioning_CopyObject_from_an_object_version(s)
+	Versioning_HeadObject_invalid_versionId(s)
+	Versioning_HeadObject_success(s)
+	Versioning_HeadObject_delete_marker(s)
+	Versioning_GetObject_invalid_versionId(s)
+	Versioning_GetObject_success(s)
+	Versioning_GetObject_delete_marker(s)
+	Versioning_DeleteObject_delete_object_version(s)
+	Versioning_DeleteObject_delete_a_delete_marker(s)
+	Versioning_DeleteObjects_success(s)
+	Versioning_DeleteObjects_delete_deleteMarkers(s)
+	// ListObjectVersions
+	ListObjectVersions_non_existing_bucket(s)
+	ListObjectVersions_list_single_object_versions(s)
+	ListObjectVersions_list_multiple_object_versions(s)
+	ListObjectVersions_multiple_object_versions_truncated(s)
+	ListObjectVersions_with_delete_markers(s)
+}
+
 type IntTests map[string]func(s *S3Conf) error
 
 func GetIntTests() IntTests {
@@ -812,5 +843,29 @@ func GetIntTests() IntTests {
 		"AccessControl_root_PutBucketAcl":                                     AccessControl_root_PutBucketAcl,
 		"AccessControl_user_PutBucketAcl_with_policy_access":                  AccessControl_user_PutBucketAcl_with_policy_access,
 		"AccessControl_copy_object_with_starting_slash_for_user":              AccessControl_copy_object_with_starting_slash_for_user,
+		"PutBucketVersioning_non_existing_bucket":                             PutBucketVersioning_non_existing_bucket,
+		"PutBucketVersioning_invalid_status":                                  PutBucketVersioning_invalid_status,
+		"PutBucketVersioning_success":                                         PutBucketVersioning_success,
+		"GetBucketVersioning_non_existing_bucket":                             GetBucketVersioning_non_existing_bucket,
+		"GetBucketVersioning_success":                                         GetBucketVersioning_success,
+		"Versioning_PutObject_success":                                        Versioning_PutObject_success,
+		"Versioning_CopyObject_success":                                       Versioning_CopyObject_success,
+		"Versioning_CopyObject_non_existing_version_id":                       Versioning_CopyObject_non_existing_version_id,
+		"Versioning_CopyObject_from_an_object_version":                        Versioning_CopyObject_from_an_object_version,
+		"Versioning_HeadObject_invalid_versionId":                             Versioning_HeadObject_invalid_versionId,
+		"Versioning_HeadObject_success":                                       Versioning_HeadObject_success,
+		"Versioning_HeadObject_delete_marker":                                 Versioning_HeadObject_delete_marker,
+		"Versioning_GetObject_invalid_versionId":                              Versioning_GetObject_invalid_versionId,
+		"Versioning_GetObject_success":                                        Versioning_GetObject_success,
+		"Versioning_GetObject_delete_marker":                                  Versioning_GetObject_delete_marker,
+		"Versioning_DeleteObject_delete_object_version":                       Versioning_DeleteObject_delete_object_version,
+		"Versioning_DeleteObject_delete_a_delete_marker":                      Versioning_DeleteObject_delete_a_delete_marker,
+		"Versioning_DeleteObjects_success":                                    Versioning_DeleteObjects_success,
+		"Versioning_DeleteObjects_delete_deleteMarkers":                       Versioning_DeleteObjects_delete_deleteMarkers,
+		"ListObjectVersions_non_existing_bucket":                              ListObjectVersions_non_existing_bucket,
+		"ListObjectVersions_list_single_object_versions":                      ListObjectVersions_list_single_object_versions,
+		"ListObjectVersions_list_multiple_object_versions":                    ListObjectVersions_list_multiple_object_versions,
+		"ListObjectVersions_multiple_object_versions_truncated":               ListObjectVersions_multiple_object_versions_truncated,
+		"ListObjectVersions_with_delete_markers":                              ListObjectVersions_with_delete_markers,
 	}
 }
diff --git a/tests/integration/s3conf.go b/tests/integration/s3conf.go
index c547e8d2..49c362db 100644
--- a/tests/integration/s3conf.go
+++ b/tests/integration/s3conf.go
@@ -31,15 +31,16 @@ import (
 )
 
 type S3Conf struct {
-	awsID           string
-	awsSecret       string
-	awsRegion       string
-	endpoint        string
-	checksumDisable bool
-	pathStyle       bool
-	PartSize        int64
-	Concurrency     int
-	debug           bool
+	awsID             string
+	awsSecret         string
+	awsRegion         string
+	endpoint          string
+	checksumDisable   bool
+	pathStyle         bool
+	PartSize          int64
+	Concurrency       int
+	debug             bool
+	versioningEnabled bool
 }
 
 func NewS3Conf(opts ...Option) *S3Conf {
@@ -80,6 +81,9 @@ func WithConcurrency(c int) Option {
 func WithDebug() Option {
 	return func(s *S3Conf) { s.debug = true }
 }
+func WithVersioningEnabled() Option {
+	return func(s *S3Conf) { s.versioningEnabled = true }
+}
 
 func (c *S3Conf) getCreds() credentials.StaticCredentialsProvider {
 	// TODO support token/IAM
diff --git a/tests/integration/tests.go b/tests/integration/tests.go
index 016d6b37..86d30044 100644
--- a/tests/integration/tests.go
+++ b/tests/integration/tests.go
@@ -2985,7 +2985,7 @@ func HeadObject_non_existing_dir_object(s *S3Conf) error {
 			"key2": "val2",
 		}
 
-		_, _, err := putObjectWithData(dataLen, &s3.PutObjectInput{
+		_, err := putObjectWithData(dataLen, &s3.PutObjectInput{
 			Bucket:   &bucket,
 			Key:      &obj,
 			Metadata: meta,
@@ -3018,7 +3018,7 @@ func HeadObject_with_contenttype(s *S3Conf) error {
 		contentType := "text/plain"
 		contentEncoding := "gzip"
 
-		_, _, err := putObjectWithData(dataLen, &s3.PutObjectInput{
+		_, err := putObjectWithData(dataLen, &s3.PutObjectInput{
 			Bucket:          &bucket,
 			Key:             &obj,
 			ContentType:     &contentType,
@@ -3075,7 +3075,7 @@ func HeadObject_success(s *S3Conf) error {
 		}
 		ctype := defaultContentType
 
-		_, _, err := putObjectWithData(dataLen, &s3.PutObjectInput{
+		_, err := putObjectWithData(dataLen, &s3.PutObjectInput{
 			Bucket:      &bucket,
 			Key:         &obj,
 			Metadata:    meta,
@@ -3234,7 +3234,7 @@ func GetObject_invalid_ranges(s *S3Conf) error {
 	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
 		dataLength, obj := int64(1234567), "my-obj"
 
-		_, _, err := putObjectWithData(dataLength, &s3.PutObjectInput{
+		_, err := putObjectWithData(dataLength, &s3.PutObjectInput{
 			Bucket: &bucket,
 			Key:    &obj,
 		}, s3client)
@@ -3291,7 +3291,7 @@ func GetObject_with_meta(s *S3Conf) error {
 			"key2": "val2",
 		}
 
-		_, _, err := putObjectWithData(0, &s3.PutObjectInput{Bucket: &bucket, Key: &obj, Metadata: meta}, s3client)
+		_, err := putObjectWithData(0, &s3.PutObjectInput{Bucket: &bucket, Key: &obj, Metadata: meta}, s3client)
 		if err != nil {
 			return err
 		}
@@ -3320,7 +3320,7 @@ func GetObject_success(s *S3Conf) error {
 		dataLength, obj := int64(1234567), "my-obj"
 		ctype := defaultContentType
 
-		csum, _, err := putObjectWithData(dataLength, &s3.PutObjectInput{
+		r, err := putObjectWithData(dataLength, &s3.PutObjectInput{
 			Bucket:      &bucket,
 			Key:         &obj,
 			ContentType: &ctype,
@@ -3354,7 +3354,7 @@ func GetObject_success(s *S3Conf) error {
 		}
 		defer out.Body.Close()
 		outCsum := sha256.Sum256(bdy)
-		if outCsum != csum {
+		if outCsum != r.csum {
 			return fmt.Errorf("invalid object data")
 		}
 		return nil
@@ -3368,7 +3368,7 @@ func GetObject_directory_success(s *S3Conf) error {
 	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
 		dataLength, obj := int64(0), "my-dir/"
 
-		_, _, err := putObjectWithData(dataLength, &s3.PutObjectInput{
+		_, err := putObjectWithData(dataLength, &s3.PutObjectInput{
 			Bucket: &bucket,
 			Key:    &obj,
 		}, s3client)
@@ -3405,7 +3405,7 @@ func GetObject_by_range_success(s *S3Conf) error {
 	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
 		dataLength, obj := int64(1234567), "my-obj"
 
-		_, data, err := putObjectWithData(dataLength, &s3.PutObjectInput{
+		r, err := putObjectWithData(dataLength, &s3.PutObjectInput{
 			Bucket: &bucket,
 			Key:    &obj,
 		}, s3client)
@@ -3438,7 +3438,7 @@ func GetObject_by_range_success(s *S3Conf) error {
 		}
 
 		// bytes range is inclusive, go range for second value is not
-		if !isEqual(b, data[100:201]) {
+		if !isEqual(b, r.data[100:201]) {
 			return fmt.Errorf("data mismatch of range")
 		}
 
@@ -3462,7 +3462,7 @@ func GetObject_by_range_success(s *S3Conf) error {
 		}
 
 		// bytes range is inclusive, go range for second value is not
-		if !isEqual(b, data[100:]) {
+		if !isEqual(b, r.data[100:]) {
 			return fmt.Errorf("data mismatch of range")
 		}
 		return nil
@@ -3473,7 +3473,7 @@ func GetObject_by_range_resp_status(s *S3Conf) error {
 	testName := "GetObject_by_range_resp_status"
 	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
 		obj, dLen := "my-obj", int64(4000)
-		_, _, err := putObjectWithData(dLen, &s3.PutObjectInput{
+		_, err := putObjectWithData(dLen, &s3.PutObjectInput{
 			Bucket: &bucket,
 			Key:    &obj,
 		}, s3client)
@@ -3521,7 +3521,7 @@ func GetObject_non_existing_dir_object(s *S3Conf) error {
 	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
 		dataLength, obj := int64(1234567), "my-obj"
 
-		_, _, err := putObjectWithData(dataLength, &s3.PutObjectInput{
+		_, err := putObjectWithData(dataLength, &s3.PutObjectInput{
 			Bucket: &bucket,
 			Key:    &obj,
 		}, s3client)
@@ -4308,9 +4308,11 @@ func DeleteObjects_success(s *S3Conf) error {
 		}
 
 		delObjects := []types.ObjectIdentifier{}
+		delResult := []types.DeletedObject{}
 		for _, key := range objToDel {
 			k := key
 			delObjects = append(delObjects, types.ObjectIdentifier{Key: &k})
+			delResult = append(delResult, types.DeletedObject{Key: &k})
 		}
 		ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
 		out, err := s3client.DeleteObjects(ctx, &s3.DeleteObjectsInput{
@@ -4331,7 +4333,7 @@ func DeleteObjects_success(s *S3Conf) error {
 			return fmt.Errorf("expected 2 errors, instead got %v", len(out.Errors))
 		}
 
-		if !compareDelObjects(objToDel, out.Deleted) {
+		if !compareDelObjects(delResult, out.Deleted) {
 			return fmt.Errorf("unexpected deleted output")
 		}
 
@@ -4559,7 +4561,7 @@ func CopyObject_CopySource_starting_with_slash(s *S3Conf) error {
 			return err
 		}
 
-		csum, _, err := putObjectWithData(dataLength, &s3.PutObjectInput{
+		r, err := putObjectWithData(dataLength, &s3.PutObjectInput{
 			Bucket: &bucket,
 			Key:    &obj,
 		}, s3client)
@@ -4588,7 +4590,7 @@ func CopyObject_CopySource_starting_with_slash(s *S3Conf) error {
 			return err
 		}
 		if *out.ContentLength != dataLength {
-			return fmt.Errorf("expected content-length %v, instead got %v", dataLength, out.ContentLength)
+			return fmt.Errorf("expected content-length %v, instead got %v", dataLength, *out.ContentLength)
 		}
 
 		defer out.Body.Close()
@@ -4598,7 +4600,7 @@ func CopyObject_CopySource_starting_with_slash(s *S3Conf) error {
 			return err
 		}
 		outCsum := sha256.Sum256(bdy)
-		if outCsum != csum {
+		if outCsum != r.csum {
 			return fmt.Errorf("invalid object data")
 		}
 
@@ -4620,7 +4622,7 @@ func CopyObject_non_existing_dir_object(s *S3Conf) error {
 			return err
 		}
 
-		_, _, err = putObjectWithData(dataLength, &s3.PutObjectInput{
+		_, err = putObjectWithData(dataLength, &s3.PutObjectInput{
 			Bucket: &bucket,
 			Key:    &obj,
 		}, s3client)
@@ -4660,7 +4662,7 @@ func CopyObject_success(s *S3Conf) error {
 			return err
 		}
 
-		csum, _, err := putObjectWithData(dataLength, &s3.PutObjectInput{
+		r, err := putObjectWithData(dataLength, &s3.PutObjectInput{
 			Bucket: &bucket,
 			Key:    &obj,
 		}, s3client)
@@ -4689,7 +4691,7 @@ func CopyObject_success(s *S3Conf) error {
 			return err
 		}
 		if *out.ContentLength != dataLength {
-			return fmt.Errorf("expected content-length %v, instead got %v", dataLength, out.ContentLength)
+			return fmt.Errorf("expected content-length %v, instead got %v", dataLength, *out.ContentLength)
 		}
 
 		bdy, err := io.ReadAll(out.Body)
@@ -4698,7 +4700,7 @@ func CopyObject_success(s *S3Conf) error {
 		}
 		defer out.Body.Close()
 		outCsum := sha256.Sum256(bdy)
-		if outCsum != csum {
+		if outCsum != r.csum {
 			return fmt.Errorf("invalid object data")
 		}
 
@@ -5724,7 +5726,7 @@ func UploadPartCopy_success(s *S3Conf) error {
 			return err
 		}
 		objSize := 5 * 1024 * 1024
-		_, _, err = putObjectWithData(int64(objSize), &s3.PutObjectInput{
+		_, err = putObjectWithData(int64(objSize), &s3.PutObjectInput{
 			Bucket: &srcBucket,
 			Key:    &srcObj,
 		}, s3client)
@@ -5793,7 +5795,7 @@ func UploadPartCopy_by_range_invalid_range(s *S3Conf) error {
 			return err
 		}
 		objSize := 5 * 1024 * 1024
-		_, _, err = putObjectWithData(int64(objSize), &s3.PutObjectInput{
+		_, err = putObjectWithData(int64(objSize), &s3.PutObjectInput{
 			Bucket: &srcBucket,
 			Key:    &srcObj,
 		}, s3client)
@@ -5839,7 +5841,7 @@ func UploadPartCopy_greater_range_than_obj_size(s *S3Conf) error {
 			return err
 		}
 		srcObjSize := 5 * 1024 * 1024
-		_, _, err = putObjectWithData(int64(srcObjSize), &s3.PutObjectInput{
+		_, err = putObjectWithData(int64(srcObjSize), &s3.PutObjectInput{
 			Bucket: &srcBucket,
 			Key:    &srcObj,
 		}, s3client)
@@ -5885,7 +5887,7 @@ func UploadPartCopy_by_range_success(s *S3Conf) error {
 			return err
 		}
 		objSize := 5 * 1024 * 1024
-		_, _, err = putObjectWithData(int64(objSize), &s3.PutObjectInput{
+		_, err = putObjectWithData(int64(objSize), &s3.PutObjectInput{
 			Bucket: &srcBucket,
 			Key:    &srcObj,
 		}, s3client)
@@ -10114,7 +10116,7 @@ func PutObject_overwrite_file_obj(s *S3Conf) error {
 func PutObject_dir_obj_with_data(s *S3Conf) error {
 	testName := "PutObject_dir_obj_with_data"
 	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
-		_, _, err := putObjectWithData(int64(20), &s3.PutObjectInput{
+		_, err := putObjectWithData(int64(20), &s3.PutObjectInput{
 			Bucket: &bucket,
 			Key:    getPtr("obj/"),
 		}, s3client)
@@ -10155,6 +10157,26 @@ func PutObject_name_too_long(s *S3Conf) error {
 	})
 }
 
+// Versioning tests
+func PutBucketVersioning_non_existing_bucket(s *S3Conf) error {
+	testName := "PutBucketVersioning_non_existing_bucket"
+	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
+		ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
+		_, err := s3client.PutBucketVersioning(ctx, &s3.PutBucketVersioningInput{
+			Bucket: getPtr(getBucketName()),
+			VersioningConfiguration: &types.VersioningConfiguration{
+				Status: types.BucketVersioningStatusEnabled,
+			},
+		})
+		cancel()
+		if err := checkApiErr(err, s3err.GetAPIError(s3err.ErrNoSuchBucket)); err != nil {
+			return err
+		}
+
+		return nil
+	})
+}
+
 func HeadObject_name_too_long(s *S3Conf) error {
 	testName := "HeadObject_name_too_long"
 	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
@@ -10164,6 +10186,291 @@ func HeadObject_name_too_long(s *S3Conf) error {
 			Key:    getPtr(genRandString(300)),
 		})
 		cancel()
+		if err := checkApiErr(err, s3err.GetAPIError(s3err.ErrMalformedXML)); err != nil {
+			return err
+		}
+
+		return nil
+	})
+}
+
+func PutBucketVersioning_invalid_status(s *S3Conf) error {
+	testName := "PutBucketVersioning_invalid_status"
+	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
+		ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
+		_, err := s3client.PutBucketVersioning(ctx, &s3.PutBucketVersioningInput{
+			Bucket: &bucket,
+			VersioningConfiguration: &types.VersioningConfiguration{
+				Status: types.BucketVersioningStatus("invalid_status"),
+			},
+		})
+		cancel()
+		if err := checkApiErr(err, s3err.GetAPIError(s3err.ErrMalformedXML)); err != nil {
+			return err
+		}
+
+		return nil
+	})
+}
+
+func PutBucketVersioning_success(s *S3Conf) error {
+	testName := "PutBucketVersioning_success"
+	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
+		ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
+		_, err := s3client.PutBucketVersioning(ctx, &s3.PutBucketVersioningInput{
+			Bucket: &bucket,
+			VersioningConfiguration: &types.VersioningConfiguration{
+				Status: types.BucketVersioningStatusEnabled,
+			},
+		})
+		cancel()
+		if err != nil {
+			return err
+		}
+
+		return nil
+	})
+}
+
+func GetBucketVersioning_non_existing_bucket(s *S3Conf) error {
+	testName := "GetBucketVersioning_non_existing_bucket"
+	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
+		ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
+		_, err := s3client.GetBucketVersioning(ctx, &s3.GetBucketVersioningInput{
+			Bucket: getPtr(getBucketName()),
+		})
+		cancel()
+		if err := checkApiErr(err, s3err.GetAPIError(s3err.ErrNoSuchBucket)); err != nil {
+			return err
+		}
+
+		return nil
+	})
+}
+
+func GetBucketVersioning_success(s *S3Conf) error {
+	testName := "GetBucketVersioning_success"
+	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
+		ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
+		res, err := s3client.GetBucketVersioning(ctx, &s3.GetBucketVersioningInput{
+			Bucket: &bucket,
+		})
+		cancel()
+		if err != nil {
+			return err
+		}
+
+		if res.Status != types.BucketVersioningStatusEnabled {
+			return fmt.Errorf("expected bucket versioning status to be %v, instead got %v", types.BucketVersioningStatusEnabled, res.Status)
+		}
+		return nil
+	}, withVersioning())
+}
+
+func Versioning_PutObject_success(s *S3Conf) error {
+	testName := "Versioning_PutObject_success"
+	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
+		ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
+		res, err := s3client.PutObject(ctx, &s3.PutObjectInput{
+			Bucket: &bucket,
+			Key:    getPtr("my-obj"),
+		})
+		cancel()
+		if err != nil {
+			return err
+		}
+
+		if res.VersionId == nil || *res.VersionId == "" {
+			return fmt.Errorf("expected the versionId to be returned")
+		}
+
+		return nil
+	}, withVersioning())
+}
+
+func Versioning_CopyObject_success(s *S3Conf) error {
+	testName := "Versioning_CopyObject_success"
+	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
+		dstObj := "dst-obj"
+		srcBucket, srcObj := getBucketName(), "src-obj"
+
+		if err := setup(s, srcBucket); err != nil {
+			return err
+		}
+
+		dstObjVersions, err := createObjVersions(s3client, bucket, dstObj, 1)
+		if err != nil {
+			return err
+		}
+
+		srcObjLen := int64(2345)
+		_, err = putObjectWithData(srcObjLen, &s3.PutObjectInput{
+			Bucket: &srcBucket,
+			Key:    &srcObj,
+		}, s3client)
+		if err != nil {
+			return err
+		}
+
+		ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
+		out, err := s3client.CopyObject(ctx, &s3.CopyObjectInput{
+			Bucket:     &bucket,
+			Key:        &dstObj,
+			CopySource: getPtr(fmt.Sprintf("%v/%v", srcBucket, srcObj)),
+		})
+		cancel()
+		if err != nil {
+			return err
+		}
+
+		if err := teardown(s, srcBucket); err != nil {
+			return err
+		}
+
+		if out.VersionId == nil || *out.VersionId == "" {
+			return fmt.Errorf("expected non empty versionId in the result")
+		}
+
+		dstObjVersions[0].IsLatest = getBoolPtr(false)
+		versions := append([]types.ObjectVersion{
+			{
+				ETag:      out.CopyObjectResult.ETag,
+				IsLatest:  getBoolPtr(true),
+				Key:       &dstObj,
+				Size:      &srcObjLen,
+				VersionId: out.VersionId,
+			},
+		}, dstObjVersions...)
+
+		ctx, cancel = context.WithTimeout(context.Background(), shortTimeout)
+		res, err := s3client.ListObjectVersions(ctx, &s3.ListObjectVersionsInput{
+			Bucket: &bucket,
+		})
+		cancel()
+		if err != nil {
+			return err
+		}
+
+		if !compareVersions(versions, res.Versions) {
+			return fmt.Errorf("expected the resulting versions to be %v, instead got %v", versions, res.Versions)
+		}
+
+		return nil
+	}, withVersioning())
+}
+
+func Versioning_CopyObject_non_existing_version_id(s *S3Conf) error {
+	testName := "Versioning_CopyObject_non_existing_version_id"
+	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
+		dstBucket, dstObj := getBucketName(), "my-obj"
+		srcObj := "my-obj"
+
+		if err := setup(s, dstBucket); err != nil {
+			return err
+		}
+
+		_, err := createObjVersions(s3client, bucket, srcObj, 1)
+		if err != nil {
+			return err
+		}
+
+		ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
+		_, err = s3client.CopyObject(ctx, &s3.CopyObjectInput{
+			Bucket:     &dstBucket,
+			Key:        &dstObj,
+			CopySource: getPtr(fmt.Sprintf("%v/%v?versionId=invalid_versionId", bucket, srcObj)),
+		})
+		cancel()
+		if err := checkApiErr(err, s3err.GetAPIError(s3err.ErrNoSuchVersion)); err != nil {
+			return err
+		}
+
+		if err := teardown(s, dstBucket); err != nil {
+			return err
+		}
+
+		return nil
+	}, withVersioning())
+}
+
+func Versioning_CopyObject_from_an_object_version(s *S3Conf) error {
+	testName := "Versioning_CopyObject_from_an_object_version"
+	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
+		srcBucket, srcObj, dstObj := getBucketName(), "my-obj", "my-dst-obj"
+		if err := setup(s, srcBucket, withVersioning()); err != nil {
+			return err
+		}
+
+		srcObjVersions, err := createObjVersions(s3client, srcBucket, srcObj, 1)
+		if err != nil {
+			return err
+		}
+		srcObjVersion := srcObjVersions[0]
+
+		ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
+		out, err := s3client.CopyObject(ctx, &s3.CopyObjectInput{
+			Bucket:     &bucket,
+			Key:        &dstObj,
+			CopySource: getPtr(fmt.Sprintf("%v/%v?versionId=%v", srcBucket, srcObj, *srcObjVersion.VersionId)),
+		})
+		cancel()
+		if err != nil {
+			return err
+		}
+
+		if err := teardown(s, srcBucket); err != nil {
+			return err
+		}
+
+		if out.VersionId == nil || *out.VersionId == "" {
+			return fmt.Errorf("expected non empty versionId")
+		}
+		if *out.CopySourceVersionId != *srcObjVersion.VersionId {
+			return fmt.Errorf("expected the SourceVersionId to be %v, instead got %v", *srcObjVersion.VersionId, *out.CopySourceVersionId)
+		}
+
+		ctx, cancel = context.WithTimeout(context.Background(), shortTimeout)
+		res, err := s3client.HeadObject(ctx, &s3.HeadObjectInput{
+			Bucket:    &bucket,
+			Key:       &dstObj,
+			VersionId: out.VersionId,
+		})
+		cancel()
+		if err != nil {
+			return err
+		}
+
+		if *res.ContentLength != *srcObjVersion.Size {
+			return fmt.Errorf("expected the copied object size to be %v, instead got %v", *srcObjVersion.Size, *res.ContentLength)
+		}
+		if *res.VersionId != *out.VersionId {
+			return fmt.Errorf("expected the copied object versionId to be %v, instead got %v", *out.VersionId, *res.VersionId)
+		}
+
+		return nil
+	}, withVersioning())
+}
+
+func Versioning_HeadObject_invalid_versionId(s *S3Conf) error {
+	testName := "Versioning_HeadObject_invalid_versionId"
+	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
+		dLen := int64(2000)
+		obj := "my-obj"
+		_, err := putObjectWithData(dLen, &s3.PutObjectInput{
+			Bucket: &bucket,
+			Key:    &obj,
+		}, s3client)
+		if err != nil {
+			return err
+		}
+
+		ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
+		_, err = s3client.HeadObject(ctx, &s3.HeadObjectInput{
+			Bucket:    &bucket,
+			Key:       &obj,
+			VersionId: getPtr("invalid_version_id"),
+		})
+		cancel()
 		if err := checkSdkApiErr(err, "BadRequest"); err != nil {
 			return err
 		}
@@ -10183,3 +10490,688 @@ func DeleteObject_name_too_long(s *S3Conf) error {
 		return err
 	})
 }
+
+func Versioning_HeadObject_success(s *S3Conf) error {
+	testName := "Versioning_HeadObject_success"
+	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
+		dLen := int64(2000)
+		obj := "my-obj"
+		r, err := putObjectWithData(dLen, &s3.PutObjectInput{
+			Bucket: &bucket,
+			Key:    &obj,
+		}, s3client)
+		if err != nil {
+			return err
+		}
+
+		ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
+		out, err := s3client.HeadObject(ctx, &s3.HeadObjectInput{
+			Bucket:    &bucket,
+			Key:       &obj,
+			VersionId: r.res.VersionId,
+		})
+		cancel()
+		if err != nil {
+			return err
+		}
+
+		if *out.ContentLength != dLen {
+			return fmt.Errorf("expected the object content-length to be %v, instead got %v", dLen, *out.ContentLength)
+		}
+		if *out.VersionId != *r.res.VersionId {
+			return fmt.Errorf("expected the versionId to be %v, instead got %v", *r.res.VersionId, *out.VersionId)
+		}
+
+		return nil
+	}, withVersioning())
+}
+
+func Versioning_HeadObject_delete_marker(s *S3Conf) error {
+	testName := "Versioning_HeadObject_delete_marker"
+	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
+		dLen := int64(2000)
+		obj := "my-obj"
+		_, err := putObjectWithData(dLen, &s3.PutObjectInput{
+			Bucket: &bucket,
+			Key:    &obj,
+		}, s3client)
+		if err != nil {
+			return err
+		}
+
+		ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
+		out, err := s3client.DeleteObject(ctx, &s3.DeleteObjectInput{
+			Bucket: &bucket,
+			Key:    &obj,
+		})
+		cancel()
+		if err != nil {
+			return err
+		}
+
+		if out.VersionId == nil || *out.VersionId == "" {
+			return fmt.Errorf("expected non empty versionId")
+		}
+
+		ctx, cancel = context.WithTimeout(context.Background(), shortTimeout)
+		_, err = s3client.HeadObject(ctx, &s3.HeadObjectInput{
+			Bucket:    &bucket,
+			Key:       &obj,
+			VersionId: out.VersionId,
+		})
+		cancel()
+		if err := checkSdkApiErr(err, "MethodNotAllowed"); err != nil {
+			return err
+		}
+
+		return nil
+	}, withVersioning())
+}
+
+func Versioning_GetObject_invalid_versionId(s *S3Conf) error {
+	testName := "Versioning_GetObject_invalid_versionId"
+	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
+		dLen := int64(2000)
+		obj := "my-obj"
+		_, err := putObjectWithData(dLen, &s3.PutObjectInput{
+			Bucket: &bucket,
+			Key:    &obj,
+		}, s3client)
+		if err != nil {
+			return err
+		}
+
+		ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
+		_, err = s3client.GetObject(ctx, &s3.GetObjectInput{
+			Bucket:    &bucket,
+			Key:       &obj,
+			VersionId: getPtr("invalid_version_id"),
+		})
+		cancel()
+		if err := checkApiErr(err, s3err.GetAPIError(s3err.ErrInvalidVersionId)); err != nil {
+			return err
+		}
+
+		return nil
+	}, withVersioning())
+}
+
+func Versioning_GetObject_success(s *S3Conf) error {
+	testName := "Versioning_GetObject_success"
+	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
+		dLen := int64(2000)
+		obj := "my-obj"
+		r, err := putObjectWithData(dLen, &s3.PutObjectInput{
+			Bucket: &bucket,
+			Key:    &obj,
+		}, s3client)
+		if err != nil {
+			return err
+		}
+
+		ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
+		out, err := s3client.GetObject(ctx, &s3.GetObjectInput{
+			Bucket:    &bucket,
+			Key:       &obj,
+			VersionId: r.res.VersionId,
+		})
+		cancel()
+		if err != nil {
+			return err
+		}
+
+		if *out.ContentLength != dLen {
+			return fmt.Errorf("expected the object content-length to be %v, instead got %v", dLen, *out.ContentLength)
+		}
+		if *out.VersionId != *r.res.VersionId {
+			return fmt.Errorf("expected the versionId to be %v, instead got %v", *r.res.VersionId, *out.VersionId)
+		}
+
+		bdy, err := io.ReadAll(out.Body)
+		if err != nil {
+			return err
+		}
+
+		outCsum := sha256.Sum256(bdy)
+		if outCsum != r.csum {
+			return fmt.Errorf("incorrect output content")
+		}
+
+		return nil
+	}, withVersioning())
+}
+
+func Versioning_GetObject_delete_marker(s *S3Conf) error {
+	testName := "Versioning_GetObject_delete_marker"
+	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
+		dLen := int64(2000)
+		obj := "my-obj"
+		_, err := putObjectWithData(dLen, &s3.PutObjectInput{
+			Bucket: &bucket,
+			Key:    &obj,
+		}, s3client)
+		if err != nil {
+			return err
+		}
+
+		ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
+		out, err := s3client.DeleteObject(ctx, &s3.DeleteObjectInput{
+			Bucket: &bucket,
+			Key:    &obj,
+		})
+		cancel()
+		if err != nil {
+			return err
+		}
+
+		if out.VersionId == nil || *out.VersionId == "" {
+			return fmt.Errorf("expected non empty versionId")
+		}
+
+		ctx, cancel = context.WithTimeout(context.Background(), shortTimeout)
+		_, err = s3client.GetObject(ctx, &s3.GetObjectInput{
+			Bucket:    &bucket,
+			Key:       &obj,
+			VersionId: out.VersionId,
+		})
+		cancel()
+		if err := checkApiErr(err, s3err.GetAPIError(s3err.ErrMethodNotAllowed)); err != nil {
+			return err
+		}
+
+		return nil
+	}, withVersioning())
+}
+
+func Versioning_DeleteObject_delete_object_version(s *S3Conf) error {
+	testName := "Versioning_DeleteObject_delete_object_version"
+	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
+		oLen := int64(1000)
+		obj := "my-obj"
+		r, err := putObjectWithData(oLen, &s3.PutObjectInput{
+			Bucket: &bucket,
+			Key:    &obj,
+		}, s3client)
+		if err != nil {
+			return err
+		}
+
+		versionId := r.res.VersionId
+		if versionId == nil || *versionId == "" {
+			return fmt.Errorf("expected non empty versionId")
+		}
+
+		_, err = putObjects(s3client, []string{obj}, bucket)
+		if err != nil {
+			return err
+		}
+
+		ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
+		out, err := s3client.DeleteObject(ctx, &s3.DeleteObjectInput{
+			Bucket:    &bucket,
+			Key:       &obj,
+			VersionId: versionId,
+		})
+		cancel()
+		if err != nil {
+			return err
+		}
+
+		if *out.VersionId != *versionId {
+			return fmt.Errorf("expected deleted object versionId to be %v, instead got %v", *versionId, *out.VersionId)
+		}
+
+		return nil
+	}, withVersioning())
+}
+
+func Versioning_DeleteObject_delete_a_delete_marker(s *S3Conf) error {
+	testName := "Versioning_DeleteObject_delete_a_delete_marker"
+	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
+		oLen := int64(1000)
+		obj := "my-obj"
+		_, err := putObjectWithData(oLen, &s3.PutObjectInput{
+			Bucket: &bucket,
+			Key:    &obj,
+		}, s3client)
+		if err != nil {
+			return err
+		}
+
+		ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
+		out, err := s3client.DeleteObject(ctx, &s3.DeleteObjectInput{
+			Bucket: &bucket,
+			Key:    &obj,
+		})
+		cancel()
+		if err != nil {
+			return err
+		}
+
+		if out.VersionId == nil || *out.VersionId == "" {
+			return fmt.Errorf("expected non empty versionId")
+		}
+
+		ctx, cancel = context.WithTimeout(context.Background(), shortTimeout)
+		res, err := s3client.DeleteObject(ctx, &s3.DeleteObjectInput{
+			Bucket:    &bucket,
+			Key:       &obj,
+			VersionId: out.VersionId,
+		})
+		cancel()
+		if err != nil {
+			return err
+		}
+
+		if res.DeleteMarker == nil || !*res.DeleteMarker {
+			return fmt.Errorf("expected the response DeleteMarker to be true")
+		}
+		if *res.VersionId != *out.VersionId {
+			return fmt.Errorf("expected the versionId to be %v, instead got %v", *out.VersionId, *res.VersionId)
+		}
+
+		return nil
+	}, withVersioning())
+}
+
+func Versioning_DeleteObjects_success(s *S3Conf) error {
+	testName := "Versioning_DeleteObjects_success"
+	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
+		obj1, obj2, obj3 := "foo", "bar", "baz"
+
+		obj1Version, err := createObjVersions(s3client, bucket, obj1, 1)
+		if err != nil {
+			return err
+		}
+		obj2Version, err := createObjVersions(s3client, bucket, obj2, 1)
+		if err != nil {
+			return err
+		}
+		obj3Version, err := createObjVersions(s3client, bucket, obj3, 1)
+		if err != nil {
+			return err
+		}
+
+		ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
+		out, err := s3client.DeleteObjects(ctx, &s3.DeleteObjectsInput{
+			Bucket: &bucket,
+			Delete: &types.Delete{
+				Objects: []types.ObjectIdentifier{
+					{
+						Key:       obj1Version[0].Key,
+						VersionId: obj1Version[0].VersionId,
+					},
+					{
+						Key: obj2Version[0].Key,
+					},
+					{
+						Key: obj3Version[0].Key,
+					},
+				},
+			},
+		})
+		cancel()
+		if err != nil {
+			return err
+		}
+
+		delResult := []types.DeletedObject{
+			{
+				Key:       obj1Version[0].Key,
+				VersionId: obj1Version[0].VersionId,
+			},
+			{
+				Key: obj2Version[0].Key,
+			},
+			{
+				Key: obj3Version[0].Key,
+			},
+		}
+
+		if len(out.Errors) != 0 {
+			return fmt.Errorf("errors occurred during the deletion: %v", out.Errors)
+		}
+		if !compareDelObjects(delResult, out.Deleted) {
+			return fmt.Errorf("expected the deleted objects to be %v, instead got %v", delResult, out.Deleted)
+		}
+
+		ctx, cancel = context.WithTimeout(context.Background(), shortTimeout)
+		res, err := s3client.ListObjectVersions(ctx, &s3.ListObjectVersionsInput{
+			Bucket: &bucket,
+		})
+		cancel()
+		if err != nil {
+			return err
+		}
+
+		obj2Version[0].IsLatest = getBoolPtr(false)
+		obj3Version[0].IsLatest = getBoolPtr(false)
+		versions := append(obj2Version, obj3Version...)
+
+		delMarkers := []types.DeleteMarkerEntry{
+			{
+				IsLatest:  getBoolPtr(true),
+				Key:       out.Deleted[1].Key,
+				VersionId: out.Deleted[1].VersionId,
+			},
+			{
+				IsLatest:  getBoolPtr(true),
+				Key:       out.Deleted[2].Key,
+				VersionId: out.Deleted[2].VersionId,
+			},
+		}
+
+		if !compareVersions(versions, res.Versions) {
+			return fmt.Errorf("expected the resulting versions to be %v, instead got %v", versions, res.Versions)
+		}
+		if !compareDelMarkers(delMarkers, res.DeleteMarkers) {
+			return fmt.Errorf("expected the resulting delete markers to be %v, instead got %v", delMarkers, res.DeleteMarkers)
+		}
+
+		return nil
+	}, withVersioning())
+}
+
+func Versioning_DeleteObjects_delete_deleteMarkers(s *S3Conf) error {
+	testName := "Versioning_DeleteObjects_delete_deleteMarkers"
+	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
+		obj1, obj2 := "foo", "bar"
+
+		obj1Version, err := createObjVersions(s3client, bucket, obj1, 1)
+		if err != nil {
+			return err
+		}
+		obj2Version, err := createObjVersions(s3client, bucket, obj2, 1)
+		if err != nil {
+			return err
+		}
+
+		ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
+		out, err := s3client.DeleteObjects(ctx, &s3.DeleteObjectsInput{
+			Bucket: &bucket,
+			Delete: &types.Delete{
+				Objects: []types.ObjectIdentifier{
+					{
+						Key: obj1Version[0].Key,
+					},
+					{
+						Key: obj2Version[0].Key,
+					},
+				},
+			},
+		})
+		cancel()
+		if err != nil {
+			return err
+		}
+
+		delResult := []types.DeletedObject{
+			{
+				Key: obj1Version[0].Key,
+			},
+			{
+				Key: obj2Version[0].Key,
+			},
+		}
+
+		if len(out.Errors) != 0 {
+			return fmt.Errorf("errors occurred during the deletion: %v", out.Errors)
+		}
+		if !compareDelObjects(delResult, out.Deleted) {
+			return fmt.Errorf("expected the deleted objects to be %v, instead got %v", delResult, out.Deleted)
+		}
+
+		ctx, cancel = context.WithTimeout(context.Background(), shortTimeout)
+		res, err := s3client.DeleteObjects(ctx, &s3.DeleteObjectsInput{
+			Bucket: &bucket,
+			Delete: &types.Delete{
+				Objects: []types.ObjectIdentifier{
+					{
+						Key:       out.Deleted[0].Key,
+						VersionId: out.Deleted[0].VersionId,
+					},
+					{
+						Key:       out.Deleted[1].Key,
+						VersionId: out.Deleted[1].VersionId,
+					},
+				},
+			},
+		})
+		cancel()
+		if err != nil {
+			return err
+		}
+		if len(out.Errors) != 0 {
+			return fmt.Errorf("errors occurred during the deletion: %v", out.Errors)
+		}
+
+		delResult = []types.DeletedObject{
+			{
+				Key:                   out.Deleted[0].Key,
+				DeleteMarker:          getBoolPtr(true),
+				DeleteMarkerVersionId: out.Deleted[0].VersionId,
+			},
+			{
+				Key:                   out.Deleted[1].Key,
+				DeleteMarker:          getBoolPtr(true),
+				DeleteMarkerVersionId: out.Deleted[1].VersionId,
+			},
+		}
+
+		if !compareDelObjects(delResult, res.Deleted) {
+			return fmt.Errorf("expected the deleted objects to be %v, instead got %v", delResult, res.Deleted)
+		}
+
+		return nil
+	}, withVersioning())
+}
+
+func ListObjectVersions_non_existing_bucket(s *S3Conf) error {
+	testName := "ListObjectVersions_non_existing_bucket"
+	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
+		ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
+		_, err := s3client.ListObjectVersions(ctx, &s3.ListObjectVersionsInput{
+			Bucket: getPtr(getBucketName()),
+		})
+		cancel()
+		if err := checkApiErr(err, s3err.GetAPIError(s3err.ErrNoSuchBucket)); err != nil {
+			return err
+		}
+
+		return nil
+	}, withVersioning())
+}
+
+func ListObjectVersions_list_single_object_versions(s *S3Conf) error {
+	testName := "ListObjectVersions_list_single_object_versions"
+	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
+		object := "my-obj"
+		versions, err := createObjVersions(s3client, bucket, object, 5)
+		if err != nil {
+			return err
+		}
+
+		ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
+		out, err := s3client.ListObjectVersions(ctx, &s3.ListObjectVersionsInput{
+			Bucket: &bucket,
+		})
+		cancel()
+		if err != nil {
+			return err
+		}
+
+		if !compareVersions(out.Versions, versions) {
+			return fmt.Errorf("expected the resulting versions to be %v, instead got %v", versions, out.Versions)
+		}
+
+		return nil
+	}, withVersioning())
+}
+
+func ListObjectVersions_list_multiple_object_versions(s *S3Conf) error {
+	testName := "ListObjectVersions_list_multiple_object_versions"
+	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
+		obj1, obj2, obj3 := "foo", "bar", "baz"
+
+		obj1Versions, err := createObjVersions(s3client, bucket, obj1, 4)
+		if err != nil {
+			return err
+		}
+		obj2Versions, err := createObjVersions(s3client, bucket, obj2, 3)
+		if err != nil {
+			return err
+		}
+		obj3Versions, err := createObjVersions(s3client, bucket, obj3, 5)
+		if err != nil {
+			return err
+		}
+
+		versions := append(append(obj2Versions, obj3Versions...), obj1Versions...)
+
+		ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
+		out, err := s3client.ListObjectVersions(ctx, &s3.ListObjectVersionsInput{
+			Bucket: &bucket,
+		})
+		cancel()
+		if err != nil {
+			return err
+		}
+
+		if !compareVersions(out.Versions, versions) {
+			return fmt.Errorf("expected the resulting versions to be %v, instead got %v", versions, out.Versions)
+		}
+
+		return nil
+	}, withVersioning())
+}
+
+func ListObjectVersions_multiple_object_versions_truncated(s *S3Conf) error {
+	testName := "ListObjectVersions_multiple_object_versions_truncated"
+	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
+		obj1, obj2, obj3 := "foo", "bar", "baz"
+
+		obj1Versions, err := createObjVersions(s3client, bucket, obj1, 4)
+		if err != nil {
+			return err
+		}
+		obj2Versions, err := createObjVersions(s3client, bucket, obj2, 3)
+		if err != nil {
+			return err
+		}
+		obj3Versions, err := createObjVersions(s3client, bucket, obj3, 5)
+		if err != nil {
+			return err
+		}
+
+		versions := append(append(obj2Versions, obj3Versions...), obj1Versions...)
+		maxKeys := int32(5)
+
+		ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
+		out, err := s3client.ListObjectVersions(ctx, &s3.ListObjectVersionsInput{
+			Bucket:  &bucket,
+			MaxKeys: &maxKeys,
+		})
+		cancel()
+		if err != nil {
+			return err
+		}
+
+		if *out.Name != bucket {
+			return fmt.Errorf("expected the bucket name to be %v, instead got %v", bucket, *out.Name)
+		}
+		if out.IsTruncated == nil || !*out.IsTruncated {
+			return fmt.Errorf("expected the output to be truncated")
+		}
+		if out.MaxKeys == nil || *out.MaxKeys != maxKeys {
+			return fmt.Errorf("expected the max-keys to be %v, instead got %v", maxKeys, *out.MaxKeys)
+		}
+		if *out.NextKeyMarker != *versions[maxKeys-1].Key {
+			return fmt.Errorf("expected the NextKeyMarker to be %v, instead got %v", *versions[maxKeys].Key, *out.NextKeyMarker)
+		}
+		if *out.NextVersionIdMarker != *versions[maxKeys-1].VersionId {
+			return fmt.Errorf("expected the NextVersionIdMarker to be %v, instead got %v", *versions[maxKeys].VersionId, *out.NextVersionIdMarker)
+		}
+
+		if !compareVersions(out.Versions, versions[:maxKeys]) {
+			return fmt.Errorf("expected the resulting object versions to be %v, instead got %v", versions[:maxKeys], out.Versions)
+		}
+
+		ctx, cancel = context.WithTimeout(context.Background(), shortTimeout)
+		out, err = s3client.ListObjectVersions(ctx, &s3.ListObjectVersionsInput{
+			Bucket:          &bucket,
+			KeyMarker:       out.NextKeyMarker,
+			VersionIdMarker: out.NextVersionIdMarker,
+		})
+		cancel()
+		if err != nil {
+			return err
+		}
+
+		if *out.Name != bucket {
+			return fmt.Errorf("expected the bucket name to be %v, instead got %v", bucket, *out.Name)
+		}
+		if out.IsTruncated != nil && *out.IsTruncated {
+			return fmt.Errorf("expected the output not to be truncated")
+		}
+		if *out.KeyMarker != *versions[maxKeys-1].Key {
+			return fmt.Errorf("expected the KeyMarker to be %v, instead got %v", *versions[maxKeys].Key, *out.KeyMarker)
+		}
+		if *out.VersionIdMarker != *versions[maxKeys-1].VersionId {
+			return fmt.Errorf("expected the VersionIdMarker to be %v, instead got %v", *versions[maxKeys].VersionId, *out.VersionIdMarker)
+		}
+
+		if !compareVersions(out.Versions, versions[maxKeys:]) {
+			return fmt.Errorf("expected the resulting object versions to be %v, instead got %v", versions[maxKeys:], out.Versions)
+		}
+
+		return nil
+	}, withVersioning())
+}
+
+func ListObjectVersions_with_delete_markers(s *S3Conf) error {
+	testName := "ListObjectVersions_with_delete_markers"
+	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
+		obj := "my-obj"
+		versions, err := createObjVersions(s3client, bucket, obj, 1)
+		if err != nil {
+			return err
+		}
+
+		versions[0].IsLatest = getBoolPtr(false)
+
+		ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
+		out, err := s3client.DeleteObject(ctx, &s3.DeleteObjectInput{
+			Bucket: &bucket,
+			Key:    &obj,
+		})
+		cancel()
+		if err != nil {
+			return err
+		}
+
+		delMarkers := []types.DeleteMarkerEntry{}
+		delMarkers = append(delMarkers, types.DeleteMarkerEntry{
+			Key:       &obj,
+			VersionId: out.VersionId,
+			IsLatest:  getBoolPtr(true),
+		})
+
+		ctx, cancel = context.WithTimeout(context.Background(), shortTimeout)
+		res, err := s3client.ListObjectVersions(ctx, &s3.ListObjectVersionsInput{
+			Bucket: &bucket,
+		})
+		cancel()
+		if err != nil {
+			return err
+		}
+
+		if !compareVersions(res.Versions, versions) {
+			return fmt.Errorf("expected the resulting versions to be %v, instead got %v", versions, res.Versions)
+		}
+		if !compareDelMarkers(res.DeleteMarkers, delMarkers) {
+			return fmt.Errorf("expected the resulting delete markers to be %v, instead got %v", delMarkers, res.DeleteMarkers)
+		}
+
+		return nil
+	}, withVersioning())
+}
diff --git a/tests/integration/utils.go b/tests/integration/utils.go
index 2f6c01b1..8f10caec 100644
--- a/tests/integration/utils.go
+++ b/tests/integration/utils.go
@@ -24,6 +24,7 @@ import (
 	"errors"
 	"fmt"
 	"io"
+	"math/big"
 	rnd "math/rand"
 	"net/http"
 	"net/url"
@@ -70,7 +71,25 @@ func setup(s *S3Conf, bucket string, opts ...setupOpt) error {
 		ObjectOwnership:            cfg.Ownership,
 	})
 	cancel()
-	return err
+	if err != nil {
+		return err
+	}
+
+	if cfg.VersioningEnabled {
+		ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
+		_, err := s3client.PutBucketVersioning(ctx, &s3.PutBucketVersioningInput{
+			Bucket: &bucket,
+			VersioningConfiguration: &types.VersioningConfiguration{
+				Status: types.BucketVersioningStatusEnabled,
+			},
+		})
+		cancel()
+		if err != nil {
+			return err
+		}
+	}
+
+	return nil
 }
 
 func teardown(s *S3Conf, bucket string) error {
@@ -90,24 +109,31 @@ func teardown(s *S3Conf, bucket string) error {
 		return nil
 	}
 
-	in := &s3.ListObjectsV2Input{Bucket: &bucket}
+	in := &s3.ListObjectVersionsInput{Bucket: &bucket}
 	for {
 		ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
-		out, err := s3client.ListObjectsV2(ctx, in)
+		out, err := s3client.ListObjectVersions(ctx, in)
 		cancel()
 		if err != nil {
 			return fmt.Errorf("failed to list objects: %w", err)
 		}
 
-		for _, item := range out.Contents {
-			err = deleteObject(&bucket, item.Key, nil)
+		for _, item := range out.Versions {
+			err = deleteObject(&bucket, item.Key, item.VersionId)
+			if err != nil {
+				return err
+			}
+		}
+		for _, item := range out.DeleteMarkers {
+			err = deleteObject(&bucket, item.Key, item.VersionId)
 			if err != nil {
 				return err
 			}
 		}
 
 		if out.IsTruncated != nil && *out.IsTruncated {
-			in.ContinuationToken = out.ContinuationToken
+			in.KeyMarker = out.KeyMarker
+			in.VersionIdMarker = out.NextVersionIdMarker
 		} else {
 			break
 		}
@@ -122,8 +148,9 @@ func teardown(s *S3Conf, bucket string) error {
 }
 
 type setupCfg struct {
-	LockEnabled bool
-	Ownership   types.ObjectOwnership
+	LockEnabled       bool
+	VersioningEnabled bool
+	Ownership         types.ObjectOwnership
 }
 
 type setupOpt func(*setupCfg)
@@ -134,6 +161,9 @@ func withLock() setupOpt {
 func withOwnership(o types.ObjectOwnership) setupOpt {
 	return func(s *setupCfg) { s.Ownership = o }
 }
+func withVersioning() setupOpt {
+	return func(s *setupCfg) { s.VersioningEnabled = true }
+}
 
 func actionHandler(s *S3Conf, testName string, handler func(s3client *s3.Client, bucket string) error, opts ...setupOpt) error {
 	runF(testName)
@@ -383,18 +413,31 @@ func contains(s []string, e string) bool {
 	return false
 }
 
-func putObjectWithData(lgth int64, input *s3.PutObjectInput, client *s3.Client) (csum [32]byte, data []byte, err error) {
-	data = make([]byte, lgth)
+type putObjectOutput struct {
+	csum [32]byte
+	data []byte
+	res  *s3.PutObjectOutput
+}
+
+func putObjectWithData(lgth int64, input *s3.PutObjectInput, client *s3.Client) (*putObjectOutput, error) {
+	data := make([]byte, lgth)
 	rand.Read(data)
-	csum = sha256.Sum256(data)
+	csum := sha256.Sum256(data)
 	r := bytes.NewReader(data)
 	input.Body = r
 
 	ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
-	_, err = client.PutObject(ctx, input)
+	res, err := client.PutObject(ctx, input)
 	cancel()
+	if err != nil {
+		return nil, err
+	}
 
-	return
+	return &putObjectOutput{
+		csum: csum,
+		data: data,
+		res:  res,
+	}, nil
 }
 
 func createMp(s3client *s3.Client, bucket, key string) (*s3.CreateMultipartUploadOutput, error) {
@@ -592,21 +635,40 @@ func comparePrefixes(list1 []string, list2 []types.CommonPrefix) bool {
 	return true
 }
 
-func compareDelObjects(list1 []string, list2 []types.DeletedObject) bool {
+func compareDelObjects(list1, list2 []types.DeletedObject) bool {
 	if len(list1) != len(list2) {
 		return false
 	}
 
-	elementMap := make(map[string]bool)
-
-	for _, elem := range list1 {
-		elementMap[elem] = true
-	}
-
-	for _, elem := range list2 {
-		if _, found := elementMap[*elem.Key]; !found {
+	for i, obj := range list1 {
+		if *obj.Key != *list2[i].Key {
 			return false
 		}
+
+		if obj.VersionId != nil {
+			if list2[i].VersionId == nil {
+				return false
+			}
+			if *obj.VersionId != *list2[i].VersionId {
+				return false
+			}
+		}
+		if obj.DeleteMarkerVersionId != nil {
+			if list2[i].DeleteMarkerVersionId == nil {
+				return false
+			}
+			if *obj.DeleteMarkerVersionId != *list2[i].DeleteMarkerVersionId {
+				return false
+			}
+		}
+		if obj.DeleteMarker != nil {
+			if list2[i].DeleteMarker == nil {
+				return false
+			}
+			if *obj.DeleteMarker != *list2[i].DeleteMarker {
+				return false
+			}
+		}
 	}
 
 	return true
@@ -860,3 +922,124 @@ func pfxStrings(pfxs []types.CommonPrefix) []string {
 	}
 	return pfxStrs
 }
+
+func createObjVersions(client *s3.Client, bucket, object string, count int) ([]types.ObjectVersion, error) {
+	versions := []types.ObjectVersion{}
+	for i := 0; i < count; i++ {
+		rNumber, err := rand.Int(rand.Reader, big.NewInt(100000))
+		dataLength := rNumber.Int64()
+		if err != nil {
+			return nil, err
+		}
+
+		r, err := putObjectWithData(dataLength, &s3.PutObjectInput{
+			Bucket: &bucket,
+			Key:    &object,
+		}, client)
+		if err != nil {
+			return nil, err
+		}
+
+		isLatest := i == count-1
+
+		versions = append(versions, types.ObjectVersion{
+			ETag:      r.res.ETag,
+			IsLatest:  &isLatest,
+			Key:       &object,
+			Size:      &dataLength,
+			VersionId: r.res.VersionId,
+		})
+	}
+
+	versions = reverseSlice(versions)
+
+	return versions, nil
+}
+
+// ReverseSlice reverses a slice of any type
+func reverseSlice[T any](s []T) []T {
+	for i, j := 0, len(s)-1; i < j; i, j = i+1, j-1 {
+		s[i], s[j] = s[j], s[i]
+	}
+	return s
+}
+
+func compareVersions(v1, v2 []types.ObjectVersion) bool {
+	if len(v1) != len(v2) {
+		return false
+	}
+
+	for i, version := range v1 {
+		if version.Key == nil || v2[i].Key == nil {
+			return false
+		}
+		if *version.Key != *v2[i].Key {
+			return false
+		}
+
+		if version.VersionId == nil || v2[i].VersionId == nil {
+			return false
+		}
+		if *version.VersionId != *v2[i].VersionId {
+			return false
+		}
+
+		if version.IsLatest == nil || v2[i].IsLatest == nil {
+			return false
+		}
+		if *version.IsLatest != *v2[i].IsLatest {
+			return false
+		}
+
+		if version.Size == nil || v2[i].Size == nil {
+			return false
+		}
+		if *version.Size != *v2[i].Size {
+			return false
+		}
+
+		if version.ETag == nil || v2[i].ETag == nil {
+			return false
+		}
+		if *version.ETag != *v2[i].ETag {
+			return false
+		}
+	}
+
+	return true
+}
+
+func compareDelMarkers(d1, d2 []types.DeleteMarkerEntry) bool {
+	if len(d1) != len(d2) {
+		return false
+	}
+
+	for i, dEntry := range d1 {
+		if dEntry.Key == nil || d2[i].Key == nil {
+			return false
+		}
+		if *dEntry.Key != *d2[i].Key {
+			return false
+		}
+
+		if dEntry.IsLatest == nil || d2[i].IsLatest == nil {
+			return false
+		}
+		if *dEntry.IsLatest != *d2[i].IsLatest {
+			return false
+		}
+
+		if dEntry.VersionId == nil || d2[i].VersionId == nil {
+			return false
+		}
+		if *dEntry.VersionId != *d2[i].VersionId {
+			return false
+		}
+	}
+
+	return true
+}
+
+func getBoolPtr(b bool) *bool {
+	return &b
+}