Repository: fsouza/fake-gcs-server
Pull request #60 (2019-08-13 16:32:28)

Initial object versioning support - retention of old generations if version is enabled, and getObject by generation
Finally the first version of the versioning support is ready for review. First I will describe what has been covered, a functionality I want to add really soon (object listing that considers archived versions) and the ones I dont expect to cover in short term (I don't need them yet, and I think is not critical functionality to enjoy versioning), and I will add inline comments to highlight some decisions I've made that probably are worth to bear in mind during the review.

What we have covered:
* Related to versioning implementation:
   * We have targeted the in memory backend, extending the backend interface: fs backend will not store the new fields, and will refuse/fail on any operation that targets versioning or specific generations
   * Object storing updated/created/deleted dates, we can create or just inherit from inputs (but delete time, that will be always consistent)
   * Return consistent ObjectAttrs.Generation
   * Bucket versioning flag (BucketAttrs.VersioningEnabled) available for the in memory engine, and modifying the CreateBucket function (with a mandatory parameter, versioning enabled or not), and also implying a bucket is no longer just the bucketName, but an object with several fields.
      * Bucket listing / props in bucket level responses considering the versioning status
   * Keeping multiple object versions when bucket versioning is enabled, returning always the last
   * Implement deletion with bucket versioning is enabled, returning consistent results (archived vs live versions)
   * Implement lookups of specific generations, that can target archived version
   * Returning a consistent object ID (that should contain name, bucket and generation)
* Non related to the functionality, but accelerates testing:
   * Execute linting+tests as part of the Dockerbuild 
      * This may be rolled back depending on the owner point of view: we are replicating, and sadly duplicating, logic we have in CI to reproduce tests locally

In the next PR we expect to cover:
* Implement object listing with Query.Versions = true

Things we expect not cover under this PR neither short term (if the previous implementation does not require to proceed in the end)
* Deletion of specific generations / archived objects
* Adding support to Conditions (in order to perform conditional operations as described [here](https://godoc.org/cloud.google.com/go/storage#hdr-Conditions) ), specifically GenerationMatch
* Bucket metageneration management
* Object metageneration management
* lifecycles rules
* Support of versions in the filesystem backend (that can be fun...)

Thanks for your patience, thanks so much for your project and hope this does not generate a big headache. 

dcaba: Hi @fsouza ! 
 
  still WIP (but maybe there are some parts I should split and just raise as different PRs?), but if you have any opinions around this work, feel free to let me know.

   Best!
fsouza: Hey @dcaba, this sounds amazing! This week is kinda busy, but I'll try to have a look on this later today or over the weekend!

Thank you very much for opening it!
dcaba: > Hey @dcaba, this sounds amazing! This week is kinda busy, but I'll try to have a look on this later today or over the weekend!
> 
> Thank you very much for opening it!

No rush @fsouza , and apologies: it's still WIP. Hope I can continue (and hopefully finish) next week, but yeah, if you want to start taking a look, great.

   Best!
dcaba: Hi @fsouza : these days are being more challenging than expected in my side, but i hope i can continue next week. And yeah, happy to start getting comments for any part you dont like! I added a note in the PR description describing a bit the plans: in regards to conditions, `Conditions.GenerationMatch` is probably something I will need to cover, but probably not everything... so, yeah, i guess there will be parts to extend after my contribution
dcaba: @fsouza , ready for review. PR description updated, and added comments inline highlighting some implementation topics. Open to your feedback and extra changes. Just seen there has been recent changes conflicting with my changes: I will take a look asap, and work in a different branch from now to implement old-generations listing
dcaba: > @fsouza , ready for review. PR description updated, and added comments inline highlighting some implementation topics. Open to your feedback and extra changes. Just seen there has been recent changes conflicting with my changes: I will take a look asap, and work in a different branch from now to implement old-generations listing

rebase done; no conflicts now
fsouza: @dcaba this is awesome, I haven't had a chance to take another look at the PR yet, but I'd like to thank you for putting the work and getting this done! 🎉 🎉 🎉 

I'll likely be able to go through it tomorrow or Saturday.
dcaba: Hi @fsouza . Feedback has been considered. I have marked all comments as resolved (as ive been using them to track remaining work), but feel free to reopen or add extra comments if there is something you dont like. Thanks for your time!

Patch:
diff --git a/.dockerignore b/.dockerignore
index 15c8dfb..db37c3b 100644
--- a/.dockerignore
+++ b/.dockerignore
@@ -3,3 +3,4 @@ README.md
 examples
 LICENSE
 .git
+ci/
diff --git a/Dockerfile b/Dockerfile
index c33986a..5617fb0 100644
--- a/Dockerfile
+++ b/Dockerfile
@@ -2,15 +2,31 @@
 # Use of this source code is governed by a BSD-style
 # license that can be found in the LICENSE file.
 
-FROM golang:1.13.5-alpine AS builder
-
+FROM golang:1.13.5 AS tester
 WORKDIR /code
-ENV CGO_ENABLED=0
 ADD go.mod go.sum ./
 RUN go mod download
 ADD . ./
+RUN go test -race -vet all -mod readonly ./...
+
+FROM golangci/golangci-lint AS linter
+WORKDIR /code
+COPY --from=tester /go/pkg /go/pkg
+COPY --from=tester /code .
+RUN golangci-lint run --enable-all \
+    -D errcheck -D lll -D dupl -D gochecknoglobals -D unparam \
+    --deadline 5m \
+    ./... \
+    && rm -rf /root/.cache
+
+FROM golang:1.13.5-alpine AS builder
+WORKDIR /code
+ENV CGO_ENABLED=0
+COPY --from=tester /go/pkg /go/pkg
+COPY --from=tester /code .
 RUN go build -o fake-gcs-server
 
 FROM alpine:3.10.3
 COPY --from=builder /code/fake-gcs-server /bin/fake-gcs-server
+RUN /bin/fake-gcs-server -h
 ENTRYPOINT ["/bin/fake-gcs-server"]
diff --git a/fakestorage/bucket.go b/fakestorage/bucket.go
index f16c172..36cbd26 100644
--- a/fakestorage/bucket.go
+++ b/fakestorage/bucket.go
@@ -15,8 +15,29 @@ import (
 // require the bucket name will recognize this bucket.
 //
 // If the bucket already exists, this method does nothing.
+//
+// Deprecated: use CreateBucketWithOpts()
 func (s *Server) CreateBucket(name string) {
-	err := s.backend.CreateBucket(name)
+	err := s.backend.CreateBucket(name, false)
+	if err != nil {
+		panic(err)
+	}
+}
+
+// CreateBucketOpts defines the properties of a bucket you can create
+// with CreateBucketWithOpts()
+type CreateBucketOpts struct {
+	Name              string
+	VersioningEnabled bool
+}
+
+// CreateBucketWithOpts creates a bucket inside the server, so any API calls that
+// require the bucket name will recognize this bucket. Use CreateBucketOpts to
+// customize the options for this bucket
+//
+// If the bucket already exists, this method does nothing but panics if props differs
+func (s *Server) CreateBucketWithOpts(opts CreateBucketOpts) {
+	err := s.backend.CreateBucket(opts.Name, opts.VersioningEnabled)
 	if err != nil {
 		panic(err)
 	}
@@ -25,49 +46,55 @@ func (s *Server) CreateBucket(name string) {
 // createBucketByPost handles a POST request to create a bucket
 func (s *Server) createBucketByPost(w http.ResponseWriter, r *http.Request) {
 	// Minimal version of Bucket from google.golang.org/api/storage/v1
+
 	var data struct {
-		Name string
+		Name       string            `json:"name,omitempty"`
+		Versioning *bucketVersioning `json:"versioning,omitempty"`
 	}
 
-	// Read the bucket name from the request body JSON
+	// Read the bucket props from the request body JSON
 	decoder := json.NewDecoder(r.Body)
 	if err := decoder.Decode(&data); err != nil {
 		http.Error(w, err.Error(), http.StatusBadRequest)
 		return
 	}
 	name := data.Name
-
+	versioning := false
+	if data.Versioning != nil {
+		versioning = data.Versioning.Enabled
+	}
 	// Create the named bucket
-	if err := s.backend.CreateBucket(name); err != nil {
+	if err := s.backend.CreateBucket(name, versioning); err != nil {
 		http.Error(w, err.Error(), http.StatusInternalServerError)
 		return
 	}
 
 	// Return the created bucket:
-	resp := newBucketResponse(name)
+	resp := newBucketResponse(name, versioning)
 	json.NewEncoder(w).Encode(resp)
 }
 
 func (s *Server) listBuckets(w http.ResponseWriter, r *http.Request) {
-	bucketNames, err := s.backend.ListBuckets()
+	buckets, err := s.backend.ListBuckets()
 	if err != nil {
 		http.Error(w, err.Error(), http.StatusInternalServerError)
 		return
 	}
-	resp := newListBucketsResponse(bucketNames)
+	resp := newListBucketsResponse(buckets)
 	json.NewEncoder(w).Encode(resp)
 }
 
 func (s *Server) getBucket(w http.ResponseWriter, r *http.Request) {
 	bucketName := mux.Vars(r)["bucketName"]
 	encoder := json.NewEncoder(w)
-	if err := s.backend.GetBucket(bucketName); err != nil {
+	bucket, err := s.backend.GetBucket(bucketName)
+	if err != nil {
 		w.WriteHeader(http.StatusNotFound)
 		err := newErrorResponse(http.StatusNotFound, "Not found", nil)
 		encoder.Encode(err)
 		return
 	}
-	resp := newBucketResponse(bucketName)
+	resp := newBucketResponse(bucket.Name, bucket.VersioningEnabled)
 	w.WriteHeader(http.StatusOK)
 	encoder.Encode(resp)
 }
diff --git a/fakestorage/object.go b/fakestorage/object.go
index 6886601..99d1e14 100644
--- a/fakestorage/object.go
+++ b/fakestorage/object.go
@@ -11,6 +11,7 @@ import (
 	"sort"
 	"strconv"
 	"strings"
+	"time"
 
 	"cloud.google.com/go/storage"
 	"github.com/fsouza/fake-gcs-server/internal/backend"
@@ -28,6 +29,12 @@ type Object struct {
 	Crc32c  string            `json:"crc32c,omitempty"`
 	Md5Hash string            `json:"md5hash,omitempty"`
 	ACL     []storage.ACLRule `json:"acl,omitempty"`
+	// Dates and generation can be manually injected, so you can do assertions on them,
+	// or let us fill these fields for you
+	Created    time.Time `json:"created,omitempty"`
+	Updated    time.Time `json:"updated,omitempty"`
+	Deleted    time.Time `json:"deleted,omitempty"`
+	Generation int64     `json:"generation,omitempty"`
 }
 
 func (o *Object) id() string {
@@ -95,6 +102,13 @@ func (s *Server) ListObjects(bucketName, prefix, delimiter string) ([]Object, []
 	return respObjects, respPrefixes, nil
 }
 
+func getCurrentIfZero(date time.Time) time.Time {
+	if date.IsZero() {
+		return time.Now()
+	}
+	return date
+}
+
 func toBackendObjects(objects []Object) []backend.Object {
 	backendObjects := []backend.Object{}
 	for _, o := range objects {
@@ -107,6 +121,10 @@ func toBackendObjects(objects []Object) []backend.Object {
 			Crc32c:          o.Crc32c,
 			Md5Hash:         o.Md5Hash,
 			ACL:             o.ACL,
+			Created:         getCurrentIfZero(o.Created).Format(time.RFC3339),
+			Deleted:         o.Deleted.Format(time.RFC3339),
+			Updated:         getCurrentIfZero(o.Updated).Format(time.RFC3339),
+			Generation:      o.Generation,
 		})
 	}
 	return backendObjects
@@ -124,11 +142,21 @@ func fromBackendObjects(objects []backend.Object) []Object {
 			Crc32c:          o.Crc32c,
 			Md5Hash:         o.Md5Hash,
 			ACL:             o.ACL,
+			Created:         convertTimeWithoutError(o.Created),
+			Deleted:         convertTimeWithoutError(o.Deleted),
+			Updated:         convertTimeWithoutError(o.Updated),
+			Generation:      o.Generation,
 		})
 	}
 	return backendObjects
 }
 
+// https://github.com/googleapis/google-cloud-go/blob/2f857649c55302802e95b96119dd05032a61c87a/storage/storage.go#L1023
+func convertTimeWithoutError(t string) time.Time {
+	r, _ := time.Parse(time.RFC3339, t)
+	return r
+}
+
 // GetObject returns the object with the given name in the given bucket, or an
 // error if the object doesn't exist.
 func (s *Server) GetObject(bucketName, objectName string) (Object, error) {
@@ -140,6 +168,17 @@ func (s *Server) GetObject(bucketName, objectName string) (Object, error) {
 	return obj, nil
 }
 
+// GetObjectWithGeneration returns the object with the given name and given generation ID in the given bucket,
+// or an error if the object doesn't exist. If versioning is enabled, archived versions are considered
+func (s *Server) GetObjectWithGeneration(bucketName, objectName string, generation int64) (Object, error) {
+	backendObj, err := s.backend.GetObjectWithGeneration(bucketName, objectName, generation)
+	if err != nil {
+		return Object{}, err
+	}
+	obj := fromBackendObjects([]backend.Object{backendObj})[0]
+	return obj, nil
+}
+
 func (s *Server) listObjects(w http.ResponseWriter, r *http.Request) {
 	bucketName := mux.Vars(r)["bucketName"]
 	prefix := r.URL.Query().Get("prefix")
@@ -164,7 +203,25 @@ func (s *Server) getObject(w http.ResponseWriter, r *http.Request) {
 	}
 
 	encoder := json.NewEncoder(w)
-	obj, err := s.GetObject(vars["bucketName"], vars["objectName"])
+	generationStr := r.FormValue("generation")
+	var (
+		obj        Object
+		err        error
+		generation int64
+	)
+	if generationStr != "" {
+		generation, err = strconv.ParseInt(generationStr, 10, 64)
+		if err != nil {
+			errResp := newErrorResponse(http.StatusBadRequest, "Wrong generation ID", nil)
+			w.WriteHeader(http.StatusBadRequest)
+			encoder.Encode(errResp)
+			return
+		}
+		obj, err = s.GetObjectWithGeneration(vars["bucketName"], vars["objectName"], generation)
+	} else {
+		obj, err = s.GetObject(vars["bucketName"], vars["objectName"])
+	}
+
 	if err != nil {
 		errResp := newErrorResponse(http.StatusNotFound, "Not Found", nil)
 		w.WriteHeader(http.StatusNotFound)
@@ -258,7 +315,22 @@ func (s *Server) rewriteObject(w http.ResponseWriter, r *http.Request) {
 
 func (s *Server) downloadObject(w http.ResponseWriter, r *http.Request) {
 	vars := mux.Vars(r)
-	obj, err := s.GetObject(vars["bucketName"], vars["objectName"])
+	var (
+		obj        Object
+		err        error
+		generation int64
+	)
+	generationStr := r.FormValue("generation")
+	if generationStr != "" {
+		generation, err = strconv.ParseInt(generationStr, 10, 64)
+		if err != nil {
+			http.Error(w, "Wrong generation ID", http.StatusBadRequest)
+			return
+		}
+		obj, err = s.GetObjectWithGeneration(vars["bucketName"], vars["objectName"], generation)
+	} else {
+		obj, err = s.GetObject(vars["bucketName"], vars["objectName"])
+	}
 	if err != nil {
 		http.Error(w, "not found", http.StatusNotFound)
 		return
diff --git a/fakestorage/response.go b/fakestorage/response.go
index ce0cad0..62f26be 100644
--- a/fakestorage/response.go
+++ b/fakestorage/response.go
@@ -6,8 +6,9 @@ package fakestorage
 
 import (
 	"net/http"
-	"sort"
+	"time"
 
+	"github.com/fsouza/fake-gcs-server/internal/backend"
 	"google.golang.org/api/googleapi"
 	"google.golang.org/api/storage/v1"
 )
@@ -18,29 +19,34 @@ type listResponse struct {
 	Prefixes []string      `json:"prefixes"`
 }
 
-func newListBucketsResponse(bucketNames []string) listResponse {
+func newListBucketsResponse(buckets []backend.Bucket) listResponse {
 	resp := listResponse{
 		Kind:  "storage#buckets",
-		Items: make([]interface{}, len(bucketNames)),
+		Items: make([]interface{}, len(buckets)),
 	}
-	sort.Strings(bucketNames)
-	for i, name := range bucketNames {
-		resp.Items[i] = newBucketResponse(name)
+	for i, bucket := range buckets {
+		resp.Items[i] = newBucketResponse(bucket.Name, bucket.VersioningEnabled)
 	}
 	return resp
 }
 
 type bucketResponse struct {
-	Kind string `json:"kind"`
-	ID   string `json:"id"`
-	Name string `json:"name"`
+	Kind       string            `json:"kind"`
+	ID         string            `json:"id"`
+	Name       string            `json:"name"`
+	Versioning *bucketVersioning `json:"versioning,omitempty"`
 }
 
-func newBucketResponse(bucketName string) bucketResponse {
+type bucketVersioning struct {
+	Enabled bool `json:"enabled,omitempty"`
+}
+
+func newBucketResponse(bucketName string, versioningEnabled bool) bucketResponse {
 	return bucketResponse{
-		Kind: "storage#bucket",
-		ID:   bucketName,
-		Name: bucketName,
+		Kind:       "storage#bucket",
+		ID:         bucketName,
+		Name:       bucketName,
+		Versioning: &bucketVersioning{versioningEnabled},
 	}
 }
 
@@ -67,6 +73,10 @@ type objectResponse struct {
 	Crc32c          string                         `json:"crc32c,omitempty"`
 	ACL             []*storage.ObjectAccessControl `json:"acl,omitempty"`
 	Md5Hash         string                         `json:"md5hash,omitempty"`
+	TimeCreated     string                         `json:"timeCreated,omitempty"`
+	TimeDeleted     string                         `json:"timeDeleted,omitempty"`
+	Updated         string                         `json:"updated,omitempty"`
+	Generation      int64                          `json:"generation,string"`
 }
 
 func newObjectResponse(obj Object) objectResponse {
@@ -83,6 +93,10 @@ func newObjectResponse(obj Object) objectResponse {
 		Crc32c:          obj.Crc32c,
 		Md5Hash:         obj.Md5Hash,
 		ACL:             acl,
+		TimeCreated:     obj.Created.Format(time.RFC3339),
+		TimeDeleted:     obj.Deleted.Format(time.RFC3339),
+		Updated:         obj.Updated.Format(time.RFC3339),
+		Generation:      obj.Generation,
 	}
 }
 
diff --git a/fakestorage/upload.go b/fakestorage/upload.go
index e1e4b25..72d7d63 100644
--- a/fakestorage/upload.go
+++ b/fakestorage/upload.go
@@ -41,7 +41,7 @@ type contentRange struct {
 
 func (s *Server) insertObject(w http.ResponseWriter, r *http.Request) {
 	bucketName := mux.Vars(r)["bucketName"]
-	if err := s.backend.GetBucket(bucketName); err != nil {
+	if _, err := s.backend.GetBucket(bucketName); err != nil {
 		w.WriteHeader(http.StatusNotFound)
 		err := newErrorResponse(http.StatusNotFound, "Not found", nil)
 		json.NewEncoder(w).Encode(err)
diff --git a/internal/backend/bucket.go b/internal/backend/bucket.go
new file mode 100644
index 0000000..40b4421
--- /dev/null
+++ b/internal/backend/bucket.go
@@ -0,0 +1,7 @@
+package backend
+
+// Bucket represents the bucket that is stored within the fake server.
+type Bucket struct {
+	Name              string
+	VersioningEnabled bool
+}
diff --git a/internal/backend/fs.go b/internal/backend/fs.go
index e1a0f15..1c3d5f3 100644
--- a/internal/backend/fs.go
+++ b/internal/backend/fs.go
@@ -6,6 +6,7 @@ package backend
 
 import (
 	"encoding/json"
+	"errors"
 	"fmt"
 	"io/ioutil"
 	"net/url"
@@ -47,7 +48,10 @@ func NewStorageFS(objects []Object, rootDir string) (Storage, error) {
 }
 
 // CreateBucket creates a bucket
-func (s *StorageFS) CreateBucket(name string) error {
+func (s *StorageFS) CreateBucket(name string, versioningEnabled bool) error {
+	if versioningEnabled {
+		return fmt.Errorf("not implemented: fs storage type does not support versioning yet")
+	}
 	s.mtx.Lock()
 	defer s.mtx.Unlock()
 	return s.createBucket(name)
@@ -58,36 +62,39 @@ func (s *StorageFS) createBucket(name string) error {
 }
 
 // ListBuckets lists buckets
-func (s *StorageFS) ListBuckets() ([]string, error) {
+func (s *StorageFS) ListBuckets() ([]Bucket, error) {
 	s.mtx.RLock()
 	defer s.mtx.RUnlock()
 	infos, err := ioutil.ReadDir(s.rootDir)
 	if err != nil {
 		return nil, err
 	}
-	buckets := []string{}
+	buckets := []Bucket{}
 	for _, info := range infos {
 		if info.IsDir() {
 			unescaped, err := url.PathUnescape(info.Name())
 			if err != nil {
 				return nil, fmt.Errorf("failed to unescape object name %s: %s", info.Name(), err)
 			}
-			buckets = append(buckets, unescaped)
+			buckets = append(buckets, Bucket{Name: unescaped})
 		}
 	}
 	return buckets, nil
 }
 
 // GetBucket checks if a bucket exists
-func (s *StorageFS) GetBucket(name string) error {
+func (s *StorageFS) GetBucket(name string) (Bucket, error) {
 	s.mtx.RLock()
 	defer s.mtx.RUnlock()
 	_, err := os.Stat(filepath.Join(s.rootDir, url.PathEscape(name)))
-	return err
+	return Bucket{Name: name}, err
 }
 
 // CreateObject stores an object
 func (s *StorageFS) CreateObject(obj Object) error {
+	if obj.Generation > 0 {
+		return errors.New("not implemented: fs storage type does not support objects generation yet")
+	}
 	s.mtx.Lock()
 	defer s.mtx.Unlock()
 	err := s.createBucket(obj.BucketName)
@@ -131,6 +138,11 @@ func (s *StorageFS) GetObject(bucketName, objectName string) (Object, error) {
 	return s.getObject(bucketName, objectName)
 }
 
+// GetObjectWithGeneration retrieves an specific version of the object. Not implemented
+func (s *StorageFS) GetObjectWithGeneration(bucketName, objectName string, generation int64) (Object, error) {
+	return Object{}, errors.New("not implemented: fs storage type does not support versioning yet")
+}
+
 func (s *StorageFS) getObject(bucketName, objectName string) (Object, error) {
 	encoded, err := ioutil.ReadFile(filepath.Join(s.rootDir, url.PathEscape(bucketName), url.PathEscape(objectName)))
 	if err != nil {
diff --git a/internal/backend/memory.go b/internal/backend/memory.go
index 257843a..61f4acf 100644
--- a/internal/backend/memory.go
+++ b/internal/backend/memory.go
@@ -8,117 +8,216 @@ import (
 	"errors"
 	"fmt"
 	"sync"
+	"time"
 )
 
 // StorageMemory is an implementation of the backend storage that stores data in memory
 type StorageMemory struct {
-	buckets map[string][]Object
+	buckets map[string]bucketInMemory
 	mtx     sync.RWMutex
 }
 
+type bucketInMemory struct {
+	Bucket
+	activeObjects   []Object
+	archivedObjects []Object
+}
+
+func newBucketInMemory(name string, versioningEnabled bool) bucketInMemory {
+	return bucketInMemory{Bucket{name, versioningEnabled}, []Object{}, []Object{}}
+}
+
+func (bm *bucketInMemory) addObject(obj Object) {
+	obj.Generation = getNewGenerationIfZero(obj.Generation)
+	index := findObject(obj, bm.activeObjects, false)
+	if index >= 0 {
+		if bm.VersioningEnabled {
+			bm.activeObjects[index].Deleted = time.Now().Format(time.RFC3339)
+			bm.cpToArchive(bm.activeObjects[index])
+		}
+		bm.activeObjects[index] = obj
+	} else {
+		bm.activeObjects = append(bm.activeObjects, obj)
+	}
+}
+
+func getNewGenerationIfZero(generation int64) int64 {
+	if generation == 0 {
+		return time.Now().UnixNano() / 1000
+	}
+	return generation
+}
+
+func (bm *bucketInMemory) deleteObject(obj Object, matchGeneration bool) {
+	index := findObject(obj, bm.activeObjects, matchGeneration)
+	if index < 0 {
+		return
+	}
+	if bm.VersioningEnabled {
+		obj.Deleted = time.Now().Format(time.RFC3339)
+		bm.mvToArchive(obj)
+	} else {
+		bm.deleteFromObjectList(obj, true)
+	}
+}
+
+func (bm *bucketInMemory) cpToArchive(obj Object) {
+	bm.archivedObjects = append(bm.archivedObjects, obj)
+}
+
+func (bm *bucketInMemory) mvToArchive(obj Object) {
+	bm.cpToArchive(obj)
+	bm.deleteFromObjectList(obj, true)
+}
+
+func (bm *bucketInMemory) deleteFromObjectList(obj Object, active bool) {
+	objects := bm.activeObjects
+	if !active {
+		objects = bm.archivedObjects
+	}
+	index := findObject(obj, objects, !active)
+	objects[index] = objects[len(objects)-1]
+	if active {
+		bm.activeObjects = objects[:len(objects)-1]
+	} else {
+		bm.archivedObjects = objects[:len(objects)-1]
+	}
+}
+
+// findObject looks for an object in the given list and return the index where it
+// was found, or -1 if the object doesn't exist.
+func findObject(obj Object, objectList []Object, matchGeneration bool) int {
+	for i, o := range objectList {
+		if matchGeneration && obj.ID() == o.ID() {
+			return i
+		}
+		if !matchGeneration && obj.IDNoGen() == o.IDNoGen() {
+			return i
+		}
+	}
+	return -1
+}
+
 // NewStorageMemory creates an instance of StorageMemory
 func NewStorageMemory(objects []Object) Storage {
 	s := &StorageMemory{
-		buckets: make(map[string][]Object),
+		buckets: make(map[string]bucketInMemory),
 	}
 	for _, o := range objects {
-		s.buckets[o.BucketName] = append(s.buckets[o.BucketName], o)
+		s.CreateBucket(o.BucketName, false)
+		bucket := s.buckets[o.BucketName]
+		bucket.addObject(o)
+		s.buckets[o.BucketName] = bucket
 	}
 	return s
 }
 
 // CreateBucket creates a bucket
-func (s *StorageMemory) CreateBucket(name string) error {
+func (s *StorageMemory) CreateBucket(name string, versioningEnabled bool) error {
 	s.mtx.Lock()
 	defer s.mtx.Unlock()
-	if _, ok := s.buckets[name]; !ok {
-		s.buckets[name] = nil
+	bucket, err := s.getBucketInMemory(name)
+	if err == nil {
+		if bucket.VersioningEnabled != versioningEnabled {
+			return fmt.Errorf("a bucket named %s already exists, but with different properties", name)
+		}
+		return nil
 	}
+	s.buckets[name] = newBucketInMemory(name, versioningEnabled)
 	return nil
 }
 
 // ListBuckets lists buckets
-func (s *StorageMemory) ListBuckets() ([]string, error) {
-	s.mtx.Lock()
-	defer s.mtx.Unlock()
-	buckets := []string{}
+func (s *StorageMemory) ListBuckets() ([]Bucket, error) {
+	s.mtx.RLock()
+	defer s.mtx.RUnlock()
+	buckets := []Bucket{}
 	for bucket := range s.buckets {
-		buckets = append(buckets, bucket)
+		buckets = append(buckets, Bucket{s.buckets[bucket].Name, s.buckets[bucket].VersioningEnabled})
 	}
 	return buckets, nil
 }
 
 // GetBucket checks if a bucket exists
-func (s *StorageMemory) GetBucket(name string) error {
-	s.mtx.Lock()
-	defer s.mtx.Unlock()
+func (s *StorageMemory) GetBucket(name string) (Bucket, error) {
+	s.mtx.RLock()
+	defer s.mtx.RUnlock()
+	bucketInMemory, err := s.getBucketInMemory(name)
+	return Bucket{bucketInMemory.Name, bucketInMemory.VersioningEnabled}, err
+}
 
-	if _, ok := s.buckets[name]; !ok {
-		return fmt.Errorf("no bucket named %s", name)
+func (s *StorageMemory) getBucketInMemory(name string) (bucketInMemory, error) {
+	if bucketInMemory, found := s.buckets[name]; found {
+		return bucketInMemory, nil
 	}
-	return nil
+	return bucketInMemory{}, fmt.Errorf("no bucket named %s", name)
 }
 
 // CreateObject stores an object
 func (s *StorageMemory) CreateObject(obj Object) error {
 	s.mtx.Lock()
 	defer s.mtx.Unlock()
-
-	index := s.findObject(obj)
-	if index < 0 {
-		s.buckets[obj.BucketName] = append(s.buckets[obj.BucketName], obj)
-	} else {
-		s.buckets[obj.BucketName][index] = obj
+	bucketInMemory, err := s.getBucketInMemory(obj.BucketName)
+	if err != nil {
+		bucketInMemory = newBucketInMemory(obj.BucketName, false)
 	}
+	bucketInMemory.addObject(obj)
+	s.buckets[obj.BucketName] = bucketInMemory
 	return nil
 }
 
-// findObject looks for an object in its bucket and return the index where it
-// was found, or -1 if the object doesn't exist.
-//
-// It doesn't lock the mutex, callers must lock the mutex before calling this
-// method.
-func (s *StorageMemory) findObject(obj Object) int {
-	for i, o := range s.buckets[obj.BucketName] {
-		if obj.ID() == o.ID() {
-			return i
-		}
-	}
-	return -1
-}
-
 // ListObjects lists the objects in a given bucket with a given prefix and delimeter
 func (s *StorageMemory) ListObjects(bucketName string) ([]Object, error) {
 	s.mtx.RLock()
 	defer s.mtx.RUnlock()
-	objects, ok := s.buckets[bucketName]
-	if !ok {
-		return nil, errors.New("bucket not found")
+	bucketInMemory, err := s.getBucketInMemory(bucketName)
+	if err != nil {
+		return []Object{}, err
 	}
-	return objects, nil
+	return bucketInMemory.activeObjects, nil
 }
 
 // GetObject get an object by bucket and name
 func (s *StorageMemory) GetObject(bucketName, objectName string) (Object, error) {
-	obj := Object{BucketName: bucketName, Name: objectName}
+	return s.GetObjectWithGeneration(bucketName, objectName, 0)
+}
+
+// GetObjectWithGeneration retrieves an specific version of the object
+func (s *StorageMemory) GetObjectWithGeneration(bucketName, objectName string, generation int64) (Object, error) {
 	s.mtx.RLock()
 	defer s.mtx.RUnlock()
-	index := s.findObject(obj)
+	bucketInMemory, err := s.getBucketInMemory(bucketName)
+	if err != nil {
+		return Object{}, err
+	}
+	matchGeneration := false
+	obj := Object{BucketName: bucketName, Name: objectName}
+	listToConsider := bucketInMemory.activeObjects
+	if generation != 0 {
+		matchGeneration = true
+		obj.Generation = generation
+		listToConsider = append(listToConsider, bucketInMemory.archivedObjects...)
+	}
+	index := findObject(obj, listToConsider, matchGeneration)
 	if index < 0 {
 		return obj, errors.New("object not found")
 	}
-	return s.buckets[bucketName][index], nil
+	return listToConsider[index], nil
 }
 
 // DeleteObject deletes an object by bucket and name
 func (s *StorageMemory) DeleteObject(bucketName, objectName string) error {
-	obj := Object{BucketName: bucketName, Name: objectName}
-	index := s.findObject(obj)
-	if index < 0 {
-		return fmt.Errorf("no such object in bucket %s: %s", bucketName, objectName)
+	obj, err := s.GetObject(bucketName, objectName)
+	if err != nil {
+		return err
+	}
+	s.mtx.Lock()
+	defer s.mtx.Unlock()
+	bucketInMemory, err := s.getBucketInMemory(bucketName)
+	if err != nil {
+		return err
 	}
-	bucket := s.buckets[obj.BucketName]
-	bucket[index] = bucket[len(bucket)-1]
-	s.buckets[obj.BucketName] = bucket[:len(bucket)-1]
+	bucketInMemory.deleteObject(obj, true)
+	s.buckets[bucketName] = bucketInMemory
 	return nil
 }
diff --git a/internal/backend/object.go b/internal/backend/object.go
index 7d9bafb..16d71ba 100644
--- a/internal/backend/object.go
+++ b/internal/backend/object.go
@@ -4,7 +4,12 @@
 
 package backend
 
-import "cloud.google.com/go/storage"
+import (
+	"bytes"
+	"fmt"
+
+	"cloud.google.com/go/storage"
+)
 
 // Object represents the object that is stored within the fake server.
 type Object struct {
@@ -16,9 +21,44 @@ type Object struct {
 	Crc32c          string
 	Md5Hash         string
 	ACL             []storage.ACLRule
+	Created         string
+	Deleted         string
+	Updated         string
+	Generation      int64
 }
 
 // ID is useful for comparing objects
 func (o *Object) ID() string {
-	return o.BucketName + "/" + o.Name
+	return fmt.Sprintf("%s#%d", o.IDNoGen(), o.Generation)
+}
+
+// IDNoGen does not consider the generation field
+func (o *Object) IDNoGen() string {
+	return fmt.Sprintf("%s/%s", o.BucketName, o.Name)
+}
+
+// compare also bears in mind most of the attributes, but generation if one is 0 and timestamp fields
+func (o *Object) compare(o2 Object) error {
+	if o.BucketName != o2.BucketName {
+		return fmt.Errorf("bucket name differs:\nmain %q\narg  %q", o.BucketName, o2.BucketName)
+	}
+	if o.Name != o2.Name {
+		return fmt.Errorf("wrong object name:\nmain %q\narg  %q", o.Name, o2.Name)
+	}
+	if o.ContentType != o2.ContentType {
+		return fmt.Errorf("wrong object contenttype:\nmain %q\narg  %q", o.ContentType, o2.ContentType)
+	}
+	if o.Crc32c != o2.Crc32c {
+		return fmt.Errorf("wrong crc:\nmain %q\narg  %q", o.Crc32c, o2.Crc32c)
+	}
+	if o.Md5Hash != o2.Md5Hash {
+		return fmt.Errorf("wrong md5:\nmain %q\narg  %q", o.Md5Hash, o2.Md5Hash)
+	}
+	if o.Generation != 0 && o2.Generation != 0 && o.Generation != o2.Generation {
+		return fmt.Errorf("generations different from 0, but not equal:\nmain %q\narg  %q", o.Generation, o2.Generation)
+	}
+	if !bytes.Equal(o.Content, o2.Content) {
+		return fmt.Errorf("wrong object content:\nmain %q\narg  %q", o.Content, o2.Content)
+	}
+	return nil
 }
diff --git a/internal/backend/storage.go b/internal/backend/storage.go
index 1b6ecd2..6199198 100644
--- a/internal/backend/storage.go
+++ b/internal/backend/storage.go
@@ -7,11 +7,12 @@ package backend
 
 // Storage is the generic interface for implementing the backend storage of the server
 type Storage interface {
-	CreateBucket(name string) error
-	ListBuckets() ([]string, error)
-	GetBucket(name string) error
+	CreateBucket(name string, versioningEnabled bool) error
+	ListBuckets() ([]Bucket, error)
+	GetBucket(name string) (Bucket, error)
 	CreateObject(obj Object) error
 	ListObjects(bucketName string) ([]Object, error)
 	GetObject(bucketName, objectName string) (Object, error)
+	GetObjectWithGeneration(bucketName, objectName string, generation int64) (Object, error)
 	DeleteObject(bucketName, objectName string) error
 }
diff --git a/main.go b/main.go
index 845e68c..4b50c2a 100644
--- a/main.go
+++ b/main.go
@@ -40,7 +40,7 @@ func main() {
 	}
 	logger.Infof("server started at %s", server.URL())
 	for _, bucketName := range emptyBuckets {
-		server.CreateBucket(bucketName)
+		server.CreateBucketWithOpts(fakestorage.CreateBucketOpts{Name: bucketName})
 	}
 
 	ch := make(chan os.Signal, 1)


Test patch:
diff --git a/fakestorage/bucket_test.go b/fakestorage/bucket_test.go
index e8f6439..4dbed7b 100644
--- a/fakestorage/bucket_test.go
+++ b/fakestorage/bucket_test.go
@@ -8,9 +8,9 @@ import (
 	"context"
 	"io/ioutil"
 	"os"
-	"reflect"
 	"testing"
 
+	"cloud.google.com/go/storage"
 	"google.golang.org/api/iterator"
 )
 
@@ -32,40 +32,60 @@ func TestServerClientBucketAttrs(t *testing.T) {
 		if attrs.Name != expectedName {
 			t.Errorf("wrong bucket name returned\nwant %q\ngot  %q", expectedName, attrs.Name)
 		}
+		if attrs.VersioningEnabled != false {
+			t.Errorf("wrong bucket props for %q\nexpecting no versioning by default, got it enabled", expectedName)
+		}
 	})
 }
 
 func TestServerClientBucketAttrsAfterCreateBucket(t *testing.T) {
-	runServersTest(t, nil, func(t *testing.T, server *Server) {
-		const bucketName = "best-bucket-ever"
-		server.CreateBucket(bucketName)
-		client := server.Client()
-		attrs, err := client.Bucket(bucketName).Attrs(context.Background())
-		if err != nil {
-			t.Fatal(err)
-		}
-		if attrs.Name != bucketName {
-			t.Errorf("wrong bucket name returned\nwant %q\ngot  %q", bucketName, attrs.Name)
-		}
-	})
+	for _, versioningEnabled := range []bool{true, false} {
+		versioningEnabled := versioningEnabled
+		runServersTest(t, nil, func(t *testing.T, server *Server) {
+			const bucketName = "best-bucket-ever"
+			server.CreateBucketWithOpts(CreateBucketOpts{Name: bucketName, VersioningEnabled: versioningEnabled})
+			client := server.Client()
+			attrs, err := client.Bucket(bucketName).Attrs(context.Background())
+			if err != nil {
+				t.Fatal(err)
+			}
+			if attrs.Name != bucketName {
+				t.Errorf("wrong bucket name returned\nwant %q\ngot  %q", bucketName, attrs.Name)
+			}
+			if attrs.VersioningEnabled != versioningEnabled {
+				t.Errorf("wrong bucket props for %q:\nwant versioningEnabled: %t\ngot versioningEnabled: %t", bucketName, versioningEnabled, attrs.VersioningEnabled)
+			}
+		})
+	}
 }
 
 func TestServerClientBucketAttrsAfterCreateBucketByPost(t *testing.T) {
-	runServersTest(t, nil, func(t *testing.T, server *Server) {
-		const bucketName = "post-bucket"
-		client := server.Client()
-		bucket := client.Bucket(bucketName)
-		if err := bucket.Create(context.Background(), "whatever", nil); err != nil {
-			t.Fatal(err)
-		}
-		attrs, err := client.Bucket(bucketName).Attrs(context.Background())
-		if err != nil {
-			t.Fatal(err)
-		}
-		if attrs.Name != bucketName {
-			t.Errorf("wrong bucket name returned\nwant %q\ngot  %q", bucketName, attrs.Name)
-		}
-	})
+	for _, versioningEnabled := range []bool{true, false} {
+		versioningEnabled := versioningEnabled
+		runServersTest(t, nil, func(t *testing.T, server *Server) {
+			const bucketName = "post-bucket"
+			client := server.Client()
+			bucket := client.Bucket(bucketName)
+
+			bucketAttrs := storage.BucketAttrs{
+				VersioningEnabled: versioningEnabled,
+			}
+			if err := bucket.Create(context.Background(), "whatever", &bucketAttrs); err != nil {
+				t.Fatal(err)
+			}
+			attrs, err := client.Bucket(bucketName).Attrs(context.Background())
+			if err != nil {
+				t.Fatal(err)
+			}
+			if attrs.Name != bucketName {
+				t.Errorf("wrong bucket name returned\nwant %q\ngot  %q", bucketName, attrs.Name)
+			}
+
+			if attrs.VersioningEnabled != bucketAttrs.VersioningEnabled {
+				t.Errorf("wrong bucket props for %q:\nwant versioningEnabled: %t\ngot versioningEnabled: %t", bucketName, bucketAttrs.VersioningEnabled, attrs.VersioningEnabled)
+			}
+		})
+	}
 }
 
 func TestServerClientBucketAttrsNotFound(t *testing.T) {
@@ -91,18 +111,35 @@ func TestServerClientListBuckets(t *testing.T) {
 
 	runServersTest(t, objs, func(t *testing.T, server *Server) {
 		client := server.Client()
+		const versionedBucketName = "post-bucket-with-versioning"
+		versionedBucketAttrs := storage.BucketAttrs{
+			VersioningEnabled: true,
+		}
+		if err := client.Bucket(versionedBucketName).Create(context.Background(), "whatever", &versionedBucketAttrs); err != nil {
+			t.Fatal(err)
+		}
 		it := client.Buckets(context.Background(), "whatever")
-		var returnedNames []string
+		expectedBuckets := map[string]bool{
+			"other-bucket": false, "some-bucket": false, versionedBucketName: true}
 		b, err := it.Next()
+		numberOfBuckets := 0
 		for ; err == nil; b, err = it.Next() {
-			returnedNames = append(returnedNames, b.Name)
+			numberOfBuckets++
+			versioning, found := expectedBuckets[b.Name]
+			if !found {
+				t.Errorf("unexpected bucket found\nname %s", b.Name)
+				continue
+			}
+			if versioning != b.VersioningEnabled {
+				t.Errorf("unexpected versioning value for %s\nwant %t\ngot  %t", b.Name, versioning, b.VersioningEnabled)
+			}
 		}
 		if err != iterator.Done {
 			t.Fatal(err)
 		}
-		expectedNames := []string{"other-bucket", "some-bucket"}
-		if !reflect.DeepEqual(returnedNames, expectedNames) {
-			t.Errorf("wrong names returned\nwant %#v\ngot  %#v", expectedNames, returnedNames)
+
+		if len(expectedBuckets) != numberOfBuckets {
+			t.Errorf("wrong number of buckets returned\nwant %d\ngot  %d", len(expectedBuckets), numberOfBuckets)
 		}
 	})
 }
@@ -139,7 +176,7 @@ func TestServerClientListObjects(t *testing.T) {
 			objAttrs, err := it.Next()
 			for ; err == nil; objAttrs, err = it.Next() {
 				seenFiles[objAttrs.Name] = struct{}{}
-				t.Logf("Seen file %s", objAttrs.Name)
+				t.Logf("seen file %s", objAttrs.Name)
 			}
 			if len(objects) != len(seenFiles) {
 				t.Errorf("wrong number of files\nwant %d\ngot %d", len(objects), len(seenFiles))
diff --git a/fakestorage/object_test.go b/fakestorage/object_test.go
index bda49b5..dacf976 100644
--- a/fakestorage/object_test.go
+++ b/fakestorage/object_test.go
@@ -8,10 +8,12 @@ import (
 	"bytes"
 	"context"
 	"encoding/binary"
+	"fmt"
 	"hash/crc32"
 	"io/ioutil"
 	"reflect"
 	"testing"
+	"time"
 
 	"cloud.google.com/go/storage"
 	"google.golang.org/api/iterator"
@@ -29,85 +31,163 @@ func uint32Checksum(b []byte) uint32 {
 	return checksummer.Sum32()
 }
 
-func TestServerClientObjectAttrs(t *testing.T) {
+type objectTestCases []struct {
+	testCase string
+	obj      Object
+}
+
+func getObjectTestCases() objectTestCases {
 	const (
 		bucketName      = "some-bucket"
-		objectName      = "img/hi-res/party-01.jpg"
 		content         = "some nice content"
 		contentType     = "text/plain; charset=utf-8"
 		contentEncoding = "gzip"
 	)
+	testInitExecTime := time.Now()
 	checksum := uint32Checksum([]byte(content))
 	hash := md5Hash([]byte(content))
-	objs := []Object{
+
+	tests := objectTestCases{
+		{
+			"object but no creation nor modification date",
+			Object{BucketName: bucketName, Name: "img/low-res/party-01.jpg", Content: []byte(content), ContentType: contentType, ContentEncoding: contentEncoding, Crc32c: encodedChecksum(uint32ToBytes(checksum)), Md5Hash: encodedHash(hash)},
+		},
+		{
+			"object with creation and modification dates",
+			Object{BucketName: bucketName, Name: "img/low-res/party-02.jpg", Content: []byte(content), ContentType: contentType, ContentEncoding: contentEncoding, Crc32c: encodedChecksum(uint32ToBytes(checksum)), Md5Hash: encodedHash(hash), Created: testInitExecTime, Updated: testInitExecTime},
+		},
+		{
+			"object with creation, modification dates, and generation",
+			Object{BucketName: bucketName, Name: "img/low-res/party-02.jpg", Content: []byte(content), ContentType: contentType, Crc32c: encodedChecksum(uint32ToBytes(checksum)), Md5Hash: encodedHash(hash), Created: testInitExecTime, Updated: testInitExecTime, Generation: testInitExecTime.UnixNano()},
+		},
 		{
-			BucketName:      bucketName,
-			Name:            objectName,
-			Content:         []byte(content),
-			ContentType:     contentType,
-			ContentEncoding: contentEncoding,
-			Crc32c:          encodedChecksum(uint32ToBytes(checksum)),
-			Md5Hash:         encodedHash(hash),
+			"object with no contents neither dates",
+			Object{BucketName: bucketName, Name: "video/hi-res/best_video_1080p.mp4", ContentType: "text/html; charset=utf-8"},
 		},
 	}
+	return tests
+}
 
-	runServersTest(t, objs, func(t *testing.T, server *Server) {
-		client := server.Client()
-		objHandle := client.Bucket(bucketName).Object(objectName)
-		attrs, err := objHandle.Attrs(context.TODO())
-		if err != nil {
-			t.Fatal(err)
-		}
-		if attrs.Bucket != bucketName {
-			t.Errorf("wrong bucket name\nwant %q\ngot  %q", bucketName, attrs.Bucket)
-		}
-		if attrs.Name != objectName {
-			t.Errorf("wrong object name\nwant %q\ngot  %q", objectName, attrs.Name)
-		}
-		if attrs.Size != int64(len(content)) {
-			t.Errorf("wrong size returned\nwant %d\ngot  %d", len(content), attrs.Size)
-		}
-		if attrs.ContentType != contentType {
-			t.Errorf("wrong content type\nwant %q\ngot  %q", contentType, attrs.ContentType)
-		}
-		if attrs.ContentEncoding != contentEncoding {
-			t.Errorf("wrong content encoding\nwant %q\ngot  %q", contentEncoding, attrs.ContentEncoding)
-		}
-		if attrs.CRC32C != checksum {
-			t.Errorf("wrong checksum returned\nwant %d\ngot   %d", checksum, attrs.CRC32C)
-		}
-		if !bytes.Equal(attrs.MD5, hash) {
-			t.Errorf("wrong hash returned\nwant %d\ngot   %d", hash, attrs.MD5)
-		}
-	})
+func checkObjectAttrs(testObj Object, attrs *storage.ObjectAttrs, t *testing.T) {
+	if attrs == nil {
+		t.Fatalf("unexpected nil attrs")
+	}
+	if attrs.Bucket != testObj.BucketName {
+		t.Errorf("wrong bucket name\nwant %q\ngot  %q", testObj.BucketName, attrs.Bucket)
+	}
+	if attrs.Name != testObj.Name {
+		t.Errorf("wrong object name\nwant %q\ngot  %q", testObj.Name, attrs.Name)
+	}
+	if !(testObj.Created.IsZero()) && testObj.Created.Equal(attrs.Created) {
+		t.Errorf("wrong created date\nwant %v\ngot   %v\nname %v", testObj.Created, attrs.Created, attrs.Name)
+	}
+	if !(testObj.Updated.IsZero()) && testObj.Updated.Equal(attrs.Updated) {
+		t.Errorf("wrong updated date\nwant %v\ngot   %v\nname %v", testObj.Updated, attrs.Updated, attrs.Name)
+	}
+	if testObj.Created.IsZero() && attrs.Created.IsZero() {
+		t.Errorf("wrong created date\nwant non zero, got   %v\nname %v", attrs.Created, attrs.Name)
+	}
+	if testObj.Updated.IsZero() && attrs.Updated.IsZero() {
+		t.Errorf("wrong updated date\nwant non zero, got   %v\nname %v", attrs.Updated, attrs.Name)
+	}
+	if testObj.Generation != 0 && attrs.Generation != testObj.Generation {
+		t.Errorf("wrong generation\nwant %d\ngot   %d\nname %v", testObj.Generation, attrs.Generation, attrs.Name)
+	}
+	if testObj.Generation == 0 && attrs.Generation == 0 {
+		t.Errorf("generation value is zero")
+	}
+	if attrs.ContentType != testObj.ContentType {
+		t.Errorf("wrong content type\nwant %q\ngot  %q", testObj.ContentType, attrs.ContentType)
+	}
+	if attrs.ContentEncoding != testObj.ContentEncoding {
+		t.Errorf("wrong content encoding\nwant %q\ngot  %q", testObj.ContentEncoding, attrs.ContentEncoding)
+	}
+	if testObj.Content != nil && attrs.Size != int64(len(testObj.Content)) {
+		t.Errorf("wrong size returned\nwant %d\ngot  %d", len(testObj.Content), attrs.Size)
+	}
+	if testObj.Content != nil && attrs.CRC32C != uint32Checksum(testObj.Content) {
+		t.Errorf("wrong checksum returned\nwant %d\ngot   %d", uint32Checksum(testObj.Content), attrs.CRC32C)
+	}
+	if testObj.Content != nil && !bytes.Equal(attrs.MD5, md5Hash(testObj.Content)) {
+		t.Errorf("wrong hash returned\nwant %d\ngot   %d", md5Hash(testObj.Content), attrs.MD5)
+	}
+}
+
+func TestServerClientObjectAttrs(t *testing.T) {
+	tests := getObjectTestCases()
+	for _, test := range tests {
+		test := test
+		runServersTest(t, []Object{test.obj}, func(t *testing.T, server *Server) {
+			t.Run(test.testCase, func(t *testing.T) {
+				client := server.Client()
+				objHandle := client.Bucket(test.obj.BucketName).Object(test.obj.Name)
+				attrs, err := objHandle.Attrs(context.TODO())
+				if err != nil {
+					t.Fatal(err)
+				}
+				checkObjectAttrs(test.obj, attrs, t)
+			})
+		})
+	}
 }
 
 func TestServerClientObjectAttrsAfterCreateObject(t *testing.T) {
+	tests := getObjectTestCases()
+	for _, test := range tests {
+		test := test
+		runServersTest(t, nil, func(t *testing.T, server *Server) {
+			server.CreateObject(test.obj)
+			client := server.Client()
+			objHandle := client.Bucket(test.obj.BucketName).Object(test.obj.Name)
+			attrs, err := objHandle.Attrs(context.TODO())
+			if err != nil {
+				t.Fatal(err)
+			}
+			checkObjectAttrs(test.obj, attrs, t)
+		})
+	}
+}
+
+func TestServerClientObjectAttrsAfterOverwriteWithVersioning(t *testing.T) {
 	runServersTest(t, nil, func(t *testing.T, server *Server) {
 		const (
-			bucketName  = "prod-bucket"
-			objectName  = "video/hi-res/best_video_1080p.mp4"
-			contentType = "text/html; charset=utf-8"
+			bucketName  = "some-bucket-with-ver"
+			content     = "some nice content"
+			content2    = "some nice content x2"
+			contentType = "text/plain; charset=utf-8"
 		)
-		server.CreateObject(Object{
-			BucketName:  bucketName,
-			Name:        objectName,
-			ContentType: contentType,
-		})
+		server.CreateBucketWithOpts(CreateBucketOpts{Name: bucketName, VersioningEnabled: true})
+		initialObj := Object{BucketName: bucketName, Name: "img/low-res/party-01.jpg", Content: []byte(content), ContentType: contentType, Crc32c: encodedChecksum(uint32ToBytes(uint32Checksum([]byte(content)))), Md5Hash: encodedHash(md5Hash([]byte(content)))}
+		server.CreateObject(initialObj)
 		client := server.Client()
-		objHandle := client.Bucket(bucketName).Object(objectName)
-		attrs, err := objHandle.Attrs(context.TODO())
+		objHandle := client.Bucket(bucketName).Object(initialObj.Name)
+		originalObjAttrs, err := objHandle.Attrs(context.TODO())
 		if err != nil {
 			t.Fatal(err)
 		}
-		if attrs.Bucket != bucketName {
-			t.Errorf("wrong bucket name\nwant %q\ngot  %q", bucketName, attrs.Bucket)
+		t.Logf("checking initial object attributes")
+		checkObjectAttrs(initialObj, originalObjAttrs, t)
+
+		latestObjVersion := Object{BucketName: bucketName, Name: "img/low-res/party-01.jpg", Content: []byte(content2), ContentType: contentType, Crc32c: encodedChecksum(uint32ToBytes(uint32Checksum([]byte(content2)))), Md5Hash: encodedHash(md5Hash([]byte(content2)))}
+		server.CreateObject(latestObjVersion)
+		objHandle = client.Bucket(bucketName).Object(latestObjVersion.Name)
+		latestAttrs, err := objHandle.Attrs(context.TODO())
+		if err != nil {
+			t.Fatal(err)
 		}
-		if attrs.Name != objectName {
-			t.Errorf("wrong object name\n want %q\ngot  %q", objectName, attrs.Name)
+		t.Logf("checking object attributes after overwrite")
+		checkObjectAttrs(latestObjVersion, latestAttrs, t)
+
+		objHandle = client.Bucket(bucketName).Object(initialObj.Name).Generation(originalObjAttrs.Generation)
+		originalObjAttrsAfterOverwrite, err := objHandle.Attrs(context.TODO())
+		if err != nil {
+			t.Fatal(err)
 		}
-		if attrs.ContentType != contentType {
-			t.Errorf("wrong content type\n want %q\ngot  %q", contentType, attrs.ContentType)
+		t.Logf("checking initial object attributes after overwrite")
+		initialObj.Generation = originalObjAttrs.Generation
+		checkObjectAttrs(initialObj, originalObjAttrsAfterOverwrite, t)
+		if originalObjAttrsAfterOverwrite.Deleted.IsZero() || originalObjAttrsAfterOverwrite.Deleted.Before(originalObjAttrsAfterOverwrite.Created) {
+			t.Errorf("unexpected delete time, %v", originalObjAttrsAfterOverwrite.Deleted)
 		}
 	})
 }
@@ -290,6 +370,60 @@ func TestServerClientObjectReaderAfterCreateObject(t *testing.T) {
 	})
 }
 
+func TestServerClientObjectReaderAgainstSpecificGenerations(t *testing.T) {
+	const (
+		bucketName  = "staging-bucket"
+		objectName  = "items/data-overwritten.txt"
+		content     = "data inside the object"
+		contentType = "text/plain; charset=iso-8859"
+	)
+
+	runServersTest(t, nil, func(t *testing.T, server *Server) {
+		server.CreateBucketWithOpts(CreateBucketOpts{Name: bucketName, VersioningEnabled: true})
+		object1 := Object{
+			BucketName:  bucketName,
+			Name:        objectName,
+			Content:     []byte(content),
+			ContentType: contentType,
+			Generation:  1111,
+		}
+		server.CreateObject(object1)
+		object2 := Object{
+			BucketName:  bucketName,
+			Name:        objectName,
+			Content:     []byte(content + "2"),
+			ContentType: contentType,
+		}
+		server.CreateObject(object2)
+		client := server.Client()
+		latestHandle := client.Bucket(bucketName).Object(objectName)
+		latestAttrs, err := latestHandle.Attrs(context.TODO())
+		if err != nil {
+			t.Fatal(err)
+		}
+		object2.Generation = latestAttrs.Generation
+		for _, object := range []Object{object1, object2} {
+			t.Log("about to get contents by generation for object", object)
+			objHandle := client.Bucket(bucketName).Object(objectName).Generation(object.Generation)
+			reader, err := objHandle.NewReader(context.TODO())
+			if err != nil {
+				t.Fatal(err)
+			}
+			data, err := ioutil.ReadAll(reader)
+			if err != nil {
+				t.Fatal(err)
+			}
+			if string(data) != string(object.Content) {
+				t.Errorf("wrong data returned\nwant %q\ngot  %q", string(object.Content), string(data))
+			}
+			if ct := reader.Attrs.ContentType; ct != object.ContentType {
+				t.Errorf("wrong content type\nwant %q\ngot  %q", object.ContentType, ct)
+			}
+			reader.Close()
+		}
+	})
+}
+
 func TestServerClientObjectReaderError(t *testing.T) {
 	objs := []Object{
 		{BucketName: "some-bucket", Name: "img/hi-res/party-01.jpg"},
@@ -328,8 +462,8 @@ func TestServerClientObjectReaderError(t *testing.T) {
 	})
 }
 
-func TestServiceClientListObjects(t *testing.T) {
-	objs := []Object{
+func getObjectsForListTests() []Object {
+	return []Object{
 		{BucketName: "some-bucket", Name: "img/low-res/party-01.jpg"},
 		{BucketName: "some-bucket", Name: "img/hi-res/party-01.jpg"},
 		{BucketName: "some-bucket", Name: "img/low-res/party-02.jpg"},
@@ -340,83 +474,91 @@ func TestServiceClientListObjects(t *testing.T) {
 		{BucketName: "some-bucket", Name: "video/hi-res/some_video_1080p.mp4"},
 		{BucketName: "other-bucket", Name: "static/css/style.css"},
 	}
+}
 
-	runServersTest(t, objs, func(t *testing.T, server *Server) {
-		server.CreateBucket("empty-bucket")
-		tests := []struct {
-			testCase         string
-			bucketName       string
-			query            *storage.Query
-			expectedNames    []string
-			expectedPrefixes []string
-		}{
-			{
-				"no prefix, no delimiter, multiple objects",
-				"some-bucket",
-				nil,
-				[]string{
-					"img/brand.jpg",
-					"img/hi-res/party-01.jpg",
-					"img/hi-res/party-02.jpg",
-					"img/hi-res/party-03.jpg",
-					"img/low-res/party-01.jpg",
-					"img/low-res/party-02.jpg",
-					"img/low-res/party-03.jpg",
-					"video/hi-res/some_video_1080p.mp4",
-				},
-				nil,
-			},
-			{
-				"no prefix, no delimiter, single object",
-				"other-bucket",
-				nil,
-				[]string{"static/css/style.css"},
-				nil,
-			},
-			{
-				"no prefix, no delimiter, no objects",
-				"empty-bucket",
-				nil,
-				[]string{},
-				nil,
-			},
-			{
-				"filtering prefix only",
-				"some-bucket",
-				&storage.Query{Prefix: "img/"},
-				[]string{
-					"img/brand.jpg",
-					"img/hi-res/party-01.jpg",
-					"img/hi-res/party-02.jpg",
-					"img/hi-res/party-03.jpg",
-					"img/low-res/party-01.jpg",
-					"img/low-res/party-02.jpg",
-					"img/low-res/party-03.jpg",
-				},
-				nil,
-			},
-			{
-				"full prefix",
-				"some-bucket",
-				&storage.Query{Prefix: "img/brand.jpg"},
-				[]string{"img/brand.jpg"},
-				nil,
-			},
-			{
-				"filtering prefix and delimiter",
-				"some-bucket",
-				&storage.Query{Prefix: "img/", Delimiter: "/"},
-				[]string{"img/brand.jpg"},
-				[]string{"img/hi-res/", "img/low-res/"},
+type listTest struct {
+	testCase         string
+	bucketName       string
+	query            *storage.Query
+	expectedNames    []string
+	expectedPrefixes []string
+}
+
+func getTestCasesForListTests(versioningEnabled, withOverwrites bool) []listTest {
+	return []listTest{
+		{
+			fmt.Sprintf("no prefix, no delimiter, multiple objects, versioning %t and overwrites %t", versioningEnabled, withOverwrites),
+			"some-bucket",
+			nil,
+			[]string{
+				"img/brand.jpg",
+				"img/hi-res/party-01.jpg",
+				"img/hi-res/party-02.jpg",
+				"img/hi-res/party-03.jpg",
+				"img/low-res/party-01.jpg",
+				"img/low-res/party-02.jpg",
+				"img/low-res/party-03.jpg",
+				"video/hi-res/some_video_1080p.mp4",
 			},
-			{
-				"filtering prefix, no objects",
-				"some-bucket",
-				&storage.Query{Prefix: "static/"},
-				[]string{},
-				nil,
+			nil,
+		},
+		{
+			fmt.Sprintf("no prefix, no delimiter, single object, versioning %t and overwrites %t", versioningEnabled, withOverwrites),
+			"other-bucket",
+			nil,
+			[]string{"static/css/style.css"},
+			nil,
+		},
+		{
+			fmt.Sprintf("no prefix, no delimiter, no objects, versioning %t and overwrites %t", versioningEnabled, withOverwrites),
+			"empty-bucket",
+			nil,
+			[]string{},
+			nil,
+		},
+		{
+			fmt.Sprintf("filtering prefix only, versioning %t and overwrites %t", versioningEnabled, withOverwrites),
+			"some-bucket",
+			&storage.Query{Prefix: "img/"},
+			[]string{
+				"img/brand.jpg",
+				"img/hi-res/party-01.jpg",
+				"img/hi-res/party-02.jpg",
+				"img/hi-res/party-03.jpg",
+				"img/low-res/party-01.jpg",
+				"img/low-res/party-02.jpg",
+				"img/low-res/party-03.jpg",
 			},
-		}
+			nil,
+		},
+		{
+			fmt.Sprintf("full prefix, versioning %t and overwrites %t", versioningEnabled, withOverwrites),
+			"some-bucket",
+			&storage.Query{Prefix: "img/brand.jpg"},
+			[]string{"img/brand.jpg"},
+			nil,
+		},
+		{
+			fmt.Sprintf("filtering prefix and delimiter, versioning %t and overwrites %t", versioningEnabled, withOverwrites),
+			"some-bucket",
+			&storage.Query{Prefix: "img/", Delimiter: "/"},
+			[]string{"img/brand.jpg"},
+			[]string{"img/hi-res/", "img/low-res/"},
+		},
+		{
+			fmt.Sprintf("filtering prefix, no objects, versioning %t and overwrites %t", versioningEnabled, withOverwrites),
+			"some-bucket",
+			&storage.Query{Prefix: "static/"},
+			[]string{},
+			nil,
+		},
+	}
+}
+
+func TestServiceClientListObjects(t *testing.T) {
+	runServersTest(t, getObjectsForListTests(), func(t *testing.T, server *Server) {
+		server.CreateBucketWithOpts(CreateBucketOpts{Name: "empty-bucket"})
+		tests := getTestCasesForListTests(false, false)
 		client := server.Client()
 		for _, test := range tests {
 			test := test
@@ -447,6 +589,58 @@ func TestServiceClientListObjects(t *testing.T) {
 	})
 }
 
+func TestServerClientListAfterCreate(t *testing.T) {
+	for _, versioningEnabled := range []bool{true, false} {
+		for _, withOverwrites := range []bool{true, false} {
+			versioningEnabled := versioningEnabled
+			withOverwrites := withOverwrites
+			runServersTest(t, nil, func(t *testing.T, server *Server) {
+				for _, bucketName := range []string{"some-bucket", "other-bucket", "empty-bucket"} {
+					server.CreateBucketWithOpts(CreateBucketOpts{
+						Name:              bucketName,
+						VersioningEnabled: versioningEnabled,
+					})
+				}
+				for _, obj := range getObjectsForListTests() {
+					server.CreateObject(obj)
+					if withOverwrites {
+						obj.Content = []byte("final content")
+						server.CreateObject(obj)
+					}
+				}
+				tests := getTestCasesForListTests(versioningEnabled, withOverwrites)
+				client := server.Client()
+				for _, test := range tests {
+					test := test
+					t.Run(test.testCase, func(t *testing.T) {
+						iter := client.Bucket(test.bucketName).Objects(context.TODO(), test.query)
+						var prefixes []string
+						names := []string{}
+						obj, err := iter.Next()
+						for ; err == nil; obj, err = iter.Next() {
+							if obj.Name != "" {
+								names = append(names, obj.Name)
+							}
+							if obj.Prefix != "" {
+								prefixes = append(prefixes, obj.Prefix)
+							}
+						}
+						if err != iterator.Done {
+							t.Fatal(err)
+						}
+						if !reflect.DeepEqual(names, test.expectedNames) {
+							t.Errorf("wrong names returned\nwant %#v\ngot  %#v", test.expectedNames, names)
+						}
+						if !reflect.DeepEqual(prefixes, test.expectedPrefixes) {
+							t.Errorf("wrong prefixes returned\nwant %#v\ngot  %#v", test.expectedPrefixes, prefixes)
+						}
+					})
+				}
+			})
+		}
+	}
+}
+
 func TestServiceClientListObjectsBucketNotFound(t *testing.T) {
 	runServersTest(t, nil, func(t *testing.T, server *Server) {
 		iter := server.Client().Bucket("some-bucket").Objects(context.TODO(), nil)
@@ -479,7 +673,7 @@ func TestServiceClientRewriteObject(t *testing.T) {
 	}
 
 	runServersTest(t, objs, func(t *testing.T, server *Server) {
-		server.CreateBucket("empty-bucket")
+		server.CreateBucketWithOpts(CreateBucketOpts{Name: "empty-bucket"})
 		tests := []struct {
 			testCase   string
 			bucketName string
@@ -570,7 +764,34 @@ func TestServerClientObjectDelete(t *testing.T) {
 		}
 		obj, err := server.GetObject(bucketName, objectName)
 		if err == nil {
-			t.Fatalf("unexpected nil error. obj: %#v", obj)
+			t.Fatalf("unexpected success. obj: %#v", obj)
+		}
+	})
+}
+
+func TestServerClientObjectDeleteWithVersioning(t *testing.T) {
+	obj := Object{BucketName: "some-bucket", Name: "img/hi-res/party-01.jpg", Content: []byte("some nice content"), Generation: 123}
+
+	runServersTest(t, nil, func(t *testing.T, server *Server) {
+		server.CreateBucketWithOpts(CreateBucketOpts{Name: obj.BucketName, VersioningEnabled: true})
+		server.CreateObject(obj)
+
+		client := server.Client()
+		objHandle := client.Bucket(obj.BucketName).Object(obj.Name)
+		err := objHandle.Delete(context.TODO())
+		if err != nil {
+			t.Fatal(err)
+		}
+		objAfterDelete, err := server.GetObject(obj.BucketName, obj.Name)
+		if err == nil {
+			t.Fatalf("unexpected success. obj: %#v", objAfterDelete)
+		}
+		objWithGen, err := server.GetObjectWithGeneration(obj.BucketName, obj.Name, obj.Generation)
+		if err != nil {
+			t.Fatalf("unable to retrieve archived object. err: %v", err)
+		}
+		if objWithGen.Deleted.IsZero() || objWithGen.Deleted.Before(objWithGen.Created) {
+			t.Errorf("unexpected delete time, %v", objWithGen.Deleted)
 		}
 	})
 }
diff --git a/fakestorage/upload_test.go b/fakestorage/upload_test.go
index 4d893d1..06a7b84 100644
--- a/fakestorage/upload_test.go
+++ b/fakestorage/upload_test.go
@@ -47,7 +47,7 @@ func TestServerClientObjectWriter(t *testing.T) {
 			test := test
 			t.Run(test.testCase, func(t *testing.T) {
 				const contentType = "text/plain; charset=utf-8"
-				server.CreateBucket(test.bucketName)
+				server.CreateBucketWithOpts(CreateBucketOpts{Name: test.bucketName})
 				client := server.Client()
 
 				objHandle := client.Bucket(test.bucketName).Object(test.objectName)
@@ -146,7 +146,7 @@ func TestServerClientObjectWriterBucketNotFound(t *testing.T) {
 func TestServerClientSimpleUpload(t *testing.T) {
 	server := NewServer(nil)
 	defer server.Stop()
-	server.CreateBucket("other-bucket")
+	server.CreateBucketWithOpts(CreateBucketOpts{Name: "other-bucket"})
 
 	const data = "some nice content"
 	const contentType = "text/plain"
@@ -187,7 +187,7 @@ func TestServerClientSimpleUpload(t *testing.T) {
 func TestServerClientUploadWithPredefinedAclPublicRead(t *testing.T) {
 	server := NewServer(nil)
 	defer server.Stop()
-	server.CreateBucket("other-bucket")
+	server.CreateBucketWithOpts(CreateBucketOpts{Name: "other-bucket"})
 
 	const data = "some nice content"
 	const contentType = "text/plain"
@@ -247,7 +247,7 @@ func TestServerClientUploadWithPredefinedAclPublicRead(t *testing.T) {
 func TestServerClientSimpleUploadNoName(t *testing.T) {
 	server := NewServer(nil)
 	defer server.Stop()
-	server.CreateBucket("other-bucket")
+	server.CreateBucketWithOpts(CreateBucketOpts{Name: "other-bucket"})
 
 	const data = "some nice content"
 	req, err := http.NewRequest("POST", server.URL()+"/storage/v1/b/other-bucket/o?uploadType=media", strings.NewReader(data))
@@ -274,7 +274,7 @@ func TestServerClientSimpleUploadNoName(t *testing.T) {
 func TestServerInvalidUploadType(t *testing.T) {
 	server := NewServer(nil)
 	defer server.Stop()
-	server.CreateBucket("other-bucket")
+	server.CreateBucketWithOpts(CreateBucketOpts{Name: "other-bucket"})
 	const data = "some nice content"
 	req, err := http.NewRequest("POST", server.URL()+"/storage/v1/b/other-bucket/o?uploadType=bananas&name=some-object.txt", strings.NewReader(data))
 	if err != nil {
diff --git a/internal/backend/backend_test.go b/internal/backend/backend_test.go
index 6abc296..54b9748 100644
--- a/internal/backend/backend_test.go
+++ b/internal/backend/backend_test.go
@@ -5,10 +5,10 @@
 package backend
 
 import (
-	"bytes"
 	"fmt"
 	"io/ioutil"
 	"os"
+	"reflect"
 	"testing"
 )
 
@@ -44,15 +44,51 @@ func testForStorageBackends(t *testing.T, test func(t *testing.T, storage Storag
 }
 
 func noError(t *testing.T, err error) {
+	t.Helper()
 	if err != nil {
-		t.Fatal(err)
+		t.Fatalf("should not error, but got: %v", err)
 	}
 }
 
-func shouldError(t *testing.T, err error, message string) {
+func shouldError(t *testing.T, err error) {
+	t.Helper()
 	if err == nil {
-		t.Fatal(message)
+		t.Fatalf("should error, but error is nil")
+	}
+}
+
+func uploadAndCompare(t *testing.T, storage Storage, obj Object) int64 {
+	isFSStorage := reflect.TypeOf(storage) == reflect.TypeOf(&StorageFS{})
+	err := storage.CreateObject(obj)
+	if isFSStorage && obj.Generation != 0 {
+		t.Log("FS should not support objects generation")
+		shouldError(t, err)
+		obj.Generation = 0
+		err = storage.CreateObject(obj)
+	}
+	noError(t, err)
+	activeObj, err := storage.GetObject(obj.BucketName, obj.Name)
+	noError(t, err)
+	if isFSStorage && activeObj.Generation != 0 {
+		t.Errorf("FS should leave generation empty, as it does not persist it. Value: %d", activeObj.Generation)
+	}
+	if !isFSStorage && activeObj.Generation == 0 {
+		t.Errorf("generation is empty, but we expect a unique int")
+	}
+	if err := activeObj.compare(obj); err != nil {
+		t.Errorf("object retrieved differs from the created one. Descr: %v", err)
+	}
+	objFromGeneration, err := storage.GetObjectWithGeneration(obj.BucketName, obj.Name, activeObj.Generation)
+	if isFSStorage {
+		t.Log("FS should not implement fetch with generation")
+		shouldError(t, err)
+	} else {
+		noError(t, err)
+		if err := objFromGeneration.compare(obj); err != nil {
+			t.Errorf("object retrieved differs from the created one. Descr: %v", err)
+		}
 	}
+	return activeObj.Generation
 }
 
 func TestObjectCRUD(t *testing.T) {
@@ -62,95 +98,152 @@ func TestObjectCRUD(t *testing.T) {
 	const crc1 = "crc1"
 	const md51 = "md51"
 	content2 := []byte("content2")
-	testForStorageBackends(t, func(t *testing.T, storage Storage) {
-		// Get in non-existent case
-		_, err := storage.GetObject(bucketName, objectName)
-		shouldError(t, err, "object found before being created")
-		// Delete in non-existent case
-		err = storage.DeleteObject(bucketName, objectName)
-		shouldError(t, err, "object successfully delete before being created")
-		// Create in non-existent case
-		noError(t, storage.CreateObject(Object{BucketName: bucketName, Name: objectName, Content: content1, Crc32c: crc1, Md5Hash: md51}))
-		// Get in existent case
-		obj, err := storage.GetObject(bucketName, objectName)
-		noError(t, err)
-		if obj.BucketName != bucketName {
-			t.Errorf("wrong bucket name\nwant %q\ngot  %q", bucketName, obj.BucketName)
-		}
-		if obj.Name != objectName {
-			t.Errorf("wrong object name\n want %q\ngot  %q", objectName, obj.Name)
-		}
-		if obj.Crc32c != crc1 {
-			t.Errorf("wrong crc\n want %q\ngot  %q", crc1, obj.Crc32c)
-		}
-		if obj.Md5Hash != md51 {
-			t.Errorf("wrong md5\n want %q\ngot  %q", md51, obj.Md5Hash)
-		}
-		if !bytes.Equal(obj.Content, content1) {
-			t.Errorf("wrong object content\n want %q\ngot  %q", content1, obj.Content)
-		}
-		// Create (update) in existent case
-		err = storage.CreateObject(Object{BucketName: bucketName, Name: objectName, Content: content2})
-		noError(t, err)
-		obj, err = storage.GetObject(bucketName, objectName)
-		noError(t, err)
-		if obj.BucketName != bucketName {
-			t.Errorf("wrong bucket name\nwant %q\ngot  %q", bucketName, obj.BucketName)
-		}
-		if obj.Name != objectName {
-			t.Errorf("wrong object name\n want %q\ngot  %q", objectName, obj.Name)
-		}
-		if !bytes.Equal(obj.Content, content2) {
-			t.Errorf("wrong object content\n want %q\ngot  %q", content2, obj.Content)
-		}
+	for _, versioningEnabled := range []bool{true, false} {
+		versioningEnabled := versioningEnabled
+		testForStorageBackends(t, func(t *testing.T, storage Storage) {
+			// Get in non-existent case
+			_, err := storage.GetObject(bucketName, objectName)
+			shouldError(t, err)
+			// Delete in non-existent case
+			err = storage.DeleteObject(bucketName, objectName)
+			shouldError(t, err)
+			err = storage.CreateBucket(bucketName, versioningEnabled)
+			if reflect.TypeOf(storage) == reflect.TypeOf(&StorageFS{}) && versioningEnabled {
+				t.Log("FS storage type should not implement versioning")
+				shouldError(t, err)
+				return
+			}
 
-		// List objects
-		objs, err := storage.ListObjects(bucketName)
-		noError(t, err)
-		if len(objs) != 1 {
-			t.Errorf("wrong number of objects returned\nwant 1\ngot  %d", len(objs))
-		}
-		if objs[0].Name != objectName {
-			t.Errorf("wrong object name\nwant %q\ngot  %q", objectName, objs[0].Name)
-		}
+			initialObject := Object{BucketName: bucketName, Name: objectName, Content: content1, Crc32c: crc1, Md5Hash: md51}
+			t.Logf("create an initial object on an empty bucket with versioning %t", versioningEnabled)
+			initialGeneration := uploadAndCompare(t, storage, initialObject)
 
-		// Delete in existent case
-		err = storage.DeleteObject(bucketName, objectName)
-		noError(t, err)
-	})
+			t.Logf("create (update) in existent case with explicit generation and versioning %t", versioningEnabled)
+			secondVersionWithGeneration := Object{BucketName: bucketName, Name: objectName, Content: content2, Generation: 1234}
+			uploadAndCompare(t, storage, secondVersionWithGeneration)
+
+			initialObjectFromGeneration, err := storage.GetObjectWithGeneration(initialObject.BucketName, initialObject.Name, initialGeneration)
+			if !versioningEnabled {
+				shouldError(t, err)
+			} else {
+				noError(t, err)
+				if err := initialObjectFromGeneration.compare(initialObject); err != nil {
+					t.Errorf("get initial generation - object retrieved differs from the created one. Descr: %v", err)
+				}
+			}
+
+			objs, err := storage.ListObjects(bucketName)
+			noError(t, err)
+			if len(objs) != 1 {
+				t.Errorf("wrong number of objects returned\nwant 1\ngot  %d", len(objs))
+			}
+			if objs[0].Name != objectName {
+				t.Errorf("wrong object name\nwant %q\ngot  %q", objectName, objs[0].Name)
+			}
+
+			err = storage.DeleteObject(bucketName, objectName)
+			noError(t, err)
+
+			_, err = storage.GetObject(bucketName, objectName)
+			shouldError(t, err)
+
+			retrievedObject, err := storage.GetObjectWithGeneration(secondVersionWithGeneration.BucketName, secondVersionWithGeneration.Name, secondVersionWithGeneration.Generation)
+			if !versioningEnabled {
+				shouldError(t, err)
+				return
+			}
+			noError(t, err)
+			if err := retrievedObject.compare(secondVersionWithGeneration); err != nil {
+				t.Errorf("get object by generation after removal - object retrieved differs from the created one. Descr: %v", err)
+			}
+		})
+	}
 }
 
+func TestObjectQueryErrors(t *testing.T) {
+	for _, versioningEnabled := range []bool{true, false} {
+		versioningEnabled := versioningEnabled
+		testForStorageBackends(t, func(t *testing.T, storage Storage) {
+			const bucketName = "random-bucket"
+			err := storage.CreateBucket(bucketName, versioningEnabled)
+			if reflect.TypeOf(storage) == reflect.TypeOf(&StorageFS{}) && versioningEnabled {
+				t.Log("FS storage type should not implement versioning")
+				shouldError(t, err)
+				return
+			}
+			validObject := Object{BucketName: bucketName, Name: "random-object", Content: []byte("random-content")}
+			err = storage.CreateObject(validObject)
+			noError(t, err)
+			_, err = storage.GetObjectWithGeneration(validObject.BucketName, validObject.Name, 33333)
+			shouldError(t, err)
+		})
+	}
+}
 func TestBucketCreateGetList(t *testing.T) {
-	const bucketName = "prod-bucket"
 	testForStorageBackends(t, func(t *testing.T, storage Storage) {
-		err := storage.GetBucket(bucketName)
-		if err == nil {
-			t.Fatal("bucket exists before being created")
-		}
 		buckets, err := storage.ListBuckets()
 		if err != nil {
 			t.Fatal(err)
 		}
 		if len(buckets) != 0 {
-			t.Fatalf("more than zero buckets found: %d", len(buckets))
+			t.Fatalf("more than zero buckets found: %d, and expecting zero when starting the test", len(buckets))
 		}
-		err = storage.CreateBucket(bucketName)
-		if err != nil {
-			t.Fatal(err)
+		bucketsToTest := []Bucket{
+			{"prod-bucket", false},
+			{"prod-bucket-with-versioning", true},
 		}
-		err = storage.GetBucket(bucketName)
-		if err != nil {
-			t.Fatal(err)
+		for i, bucket := range bucketsToTest {
+			_, err := storage.GetBucket(bucket.Name)
+			if err == nil {
+				t.Fatalf("bucket %s, exists before being created", bucket.Name)
+			}
+			err = storage.CreateBucket(bucket.Name, bucket.VersioningEnabled)
+			if reflect.TypeOf(storage) == reflect.TypeOf(&StorageFS{}) && bucket.VersioningEnabled {
+				if err == nil {
+					t.Fatal("fs storage should not accept creating buckets with versioning, but it's not failing")
+				}
+				continue
+			}
+			if err != nil {
+				t.Fatal(err)
+			}
+			bucketFromStorage, err := storage.GetBucket(bucket.Name)
+			if err != nil {
+				t.Fatal(err)
+			}
+			if bucketFromStorage != bucket {
+				t.Errorf("bucket %v does not have the expected props after retrieving. Expected %v", bucketFromStorage, bucket)
+			}
+			buckets, err = storage.ListBuckets()
+			if err != nil {
+				t.Fatal(err)
+			}
+			if len(buckets) != i+1 {
+				t.Errorf("number of buckets does not match the times we have lopped. Expected %d, found %d", i, len(buckets))
+			}
+			found := false
+			for _, listedBucket := range buckets {
+				if listedBucket.Name == bucket.Name {
+					found = true
+				}
+			}
+			if !found {
+				t.Errorf("Bucket we have just created is not part of the bucket listing. Expected %s, results: %v", bucket.Name, buckets)
+			}
 		}
-		buckets, err = storage.ListBuckets()
+	})
+}
+func TestBucketDuplication(t *testing.T) {
+	const bucketName = "prod-bucket"
+	testForStorageBackends(t, func(t *testing.T, storage Storage) {
+		err := storage.CreateBucket(bucketName, false)
 		if err != nil {
 			t.Fatal(err)
 		}
-		if len(buckets) != 1 {
-			t.Fatalf("one bucket not found after creating it, found: %d", len(buckets))
-		}
-		if buckets[0] != bucketName {
-			t.Fatalf("wrong bucket name; expected %s, got %s", bucketName, buckets[0])
+
+		err = storage.CreateBucket(bucketName, true)
+		if err == nil {
+			t.Fatal("we were expecting a bucket duplication error")
 		}
 	})
 }