Keep track of previous versions of keys

moto/s3/models.py

@@ -7,7 +7,7 @@ import copy
 from moto.core import BaseBackend
 from moto.core.utils import iso_8601_datetime, rfc_1123_datetime
 from .exceptions import BucketAlreadyExists
-from .utils import clean_key_name
+from .utils import clean_key_name, _VersionedKeyStore
 
 UPLOAD_ID_BYTES = 43
 UPLOAD_PART_MIN_SIZE = 5242880
@@ -151,7 +151,7 @@ class FakeBucket(object):
 
     def __init__(self, name):
         self.name = name
-        self.keys = {}
+        self.keys = _VersionedKeyStore()
         self.multiparts = {}
         self.versioning_status = None
 

moto/s3/utils.py

@@ -1,4 +1,5 @@
 import re
+import sys
 import urllib2
 import urlparse
 
@@ -25,3 +26,73 @@ def bucket_name_from_url(url):
 
 def clean_key_name(key_name):
     return urllib2.unquote(key_name)
+
+
+class _VersionedKeyStore(dict):
+
+    """ A simplified/modified version of Django's `MultiValueDict` taken from:
+    https://github.com/django/django/blob/70576740b0bb5289873f5a9a9a4e1a26b2c330e5/django/utils/datastructures.py#L282
+    """
+
+    def __sgetitem__(self, key):
+        return super(_VersionedKeyStore, self).__getitem__(key)
+
+    def __getitem__(self, key):
+        return self.__sgetitem__(key)[-1]
+
+    def __setitem__(self, key, value):
+        try:
+            current = self.__sgetitem__(key)
+            current.append(value)
+        except (KeyError, IndexError):
+            current = [value]
+
+        super(_VersionedKeyStore, self).__setitem__(key, current)
+
+    def get(self, key, default=None):
+        try:
+            return self[key]
+        except (KeyError, IndexError):
+            pass
+        return default
+
+    def getlist(self, key, default=None):
+        try:
+            return self.__sgetitem__(key)
+        except (KeyError, IndexError):
+            pass
+        return default
+
+    def setlist(self, key, list_):
+        if isinstance(list_, tuple):
+            list_ = list(list_)
+        elif not isinstance(list_, list):
+            list_ = [list_]
+
+        super(_VersionedKeyStore, self).__setitem__(key, list_)
+
+    def _iteritems(self):
+        for key in self:
+            yield key, self[key]
+
+    def _itervalues(self):
+        for key in self:
+            yield self[key]
+
+    def _iterlists(self):
+        for key in self:
+            yield key, self.getlist(key)
+
+    items = iteritems = _iteritems
+    lists = iterlists = _iterlists
+    values = itervalues = _itervalues
+
+    if sys.version_info[0] < 3:
+        def items(self):
+            return list(self.iteritems())
+
+        def values(self):
+            return list(self.itervalues())
+
+        def lists(self):
+            return list(self.iterlists())

tests/test_s3/test_s3_utils.py

@@ -1,5 +1,5 @@
 from sure import expect
-from moto.s3.utils import bucket_name_from_url
+from moto.s3.utils import bucket_name_from_url, _VersionedKeyStore
 
 
 def test_base_url():
@@ -12,3 +12,40 @@ def test_localhost_bucket():
 
 def test_localhost_without_bucket():
     expect(bucket_name_from_url('https://www.localhost:5000/def')).should.equal(None)
+
+def test_versioned_key_store():
+    d = _VersionedKeyStore()
+
+    d.should.have.length_of(0)
+
+    d['key'] = [1]
+
+    d.should.have.length_of(1)
+
+    d['key'] = 2
+    d.should.have.length_of(1)
+
+    d.should.have.key('key').being.equal(2)
+
+    d.get.when.called_with('key').should.return_value(2)
+    d.get.when.called_with('badkey').should.return_value(None)
+    d.get.when.called_with('badkey', 'HELLO').should.return_value('HELLO')
+
+    # Tests key[
+    d.shouldnt.have.key('badkey')
+    d.__getitem__.when.called_with('badkey').should.throw(KeyError)
+
+    d.getlist('key').should.have.length_of(2)
+    d.getlist('key').should.be.equal([[1], 2])
+    d.getlist('badkey').should.be.none
+
+    d.setlist('key', 1)
+    d.getlist('key').should.be.equal([1])
+
+    d.setlist('key', (1, 2))
+    d.getlist('key').shouldnt.be.equal((1, 2))
+    d.getlist('key').should.be.equal([1, 2])
+
+    d.setlist('key', [[1], [2]])
+    d['key'].should.have.length_of(1)
+    d.getlist('key').should.be.equal([[1], [2]])