1# Copyright (c) 2010 Mitch Garnaat http://garnaat.org/ 2# Copyright (c) 2010, Eucalyptus Systems, Inc. 3# All rights reserved. 4# 5# Permission is hereby granted, free of charge, to any person obtaining a 6# copy of this software and associated documentation files (the 7# "Software"), to deal in the Software without restriction, including 8# without limitation the rights to use, copy, modify, merge, publish, dis- 9# tribute, sublicense, and/or sell copies of the Software, and to permit 10# persons to whom the Software is furnished to do so, subject to the fol- 11# lowing conditions: 12# 13# The above copyright notice and this permission notice shall be included 14# in all copies or substantial portions of the Software. 15# 16# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS 17# OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABIL- 18# ITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT 19# SHALL THE AUTHOR BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, 20# WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, 21# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS 22# IN THE SOFTWARE. 23 24""" 25Some unit tests for the S3 Versioning. 26""" 27 28import unittest 29import time 30from boto.s3.connection import S3Connection 31from boto.exception import S3ResponseError 32from boto.s3.deletemarker import DeleteMarker 33from boto.compat import six 34 35class S3VersionTest (unittest.TestCase): 36 37 def setUp(self): 38 self.conn = S3Connection() 39 self.bucket_name = 'version-%d' % int(time.time()) 40 self.bucket = self.conn.create_bucket(self.bucket_name) 41 42 def tearDown(self): 43 for k in self.bucket.list_versions(): 44 self.bucket.delete_key(k.name, version_id=k.version_id) 45 self.bucket.delete() 46 47 def test_1_versions(self): 48 # check versioning off 49 d = self.bucket.get_versioning_status() 50 self.assertFalse('Versioning' in d) 51 52 # enable versioning 53 self.bucket.configure_versioning(versioning=True) 54 d = self.bucket.get_versioning_status() 55 self.assertEqual('Enabled', d['Versioning']) 56 57 # create a new key in the versioned bucket 58 k = self.bucket.new_key("foobar") 59 s1 = 'This is v1' 60 k.set_contents_from_string(s1) 61 62 # remember the version id of this object 63 v1 = k.version_id 64 65 # now get the contents from s3 66 o1 = k.get_contents_as_string().decode('utf-8') 67 68 # check to make sure content read from k is identical to original 69 self.assertEqual(s1, o1) 70 71 # now overwrite that same key with new data 72 s2 = 'This is v2' 73 k.set_contents_from_string(s2) 74 v2 = k.version_id 75 76 # now retrieve latest contents as a string and compare 77 k2 = self.bucket.new_key("foobar") 78 o2 = k2.get_contents_as_string().decode('utf-8') 79 self.assertEqual(s2, o2) 80 81 # next retrieve explicit versions and compare 82 o1 = k.get_contents_as_string(version_id=v1).decode('utf-8') 83 o2 = k.get_contents_as_string(version_id=v2).decode('utf-8') 84 self.assertEqual(s1, o1) 85 self.assertEqual(s2, o2) 86 87 # Now list all versions and compare to what we have 88 rs = self.bucket.get_all_versions() 89 self.assertEqual(v2, rs[0].version_id) 90 self.assertEqual(v1, rs[1].version_id) 91 92 # Now do a regular list command and make sure only the new key shows up 93 rs = self.bucket.get_all_keys() 94 self.assertEqual(1, len(rs)) 95 96 # Now do regular delete 97 self.bucket.delete_key('foobar') 98 99 # Now list versions and make sure old versions are there 100 # plus the DeleteMarker which is latest. 101 rs = self.bucket.get_all_versions() 102 self.assertEqual(3, len(rs)) 103 self.assertTrue(isinstance(rs[0], DeleteMarker)) 104 105 # Now delete v1 of the key 106 self.bucket.delete_key('foobar', version_id=v1) 107 108 # Now list versions again and make sure v1 is not there 109 rs = self.bucket.get_all_versions() 110 versions = [k.version_id for k in rs] 111 self.assertTrue(v1 not in versions) 112 self.assertTrue(v2 in versions) 113 114 # Now suspend Versioning on the bucket 115 self.bucket.configure_versioning(False) 116 # Allow time for the change to fully propagate. 117 time.sleep(3) 118 d = self.bucket.get_versioning_status() 119 self.assertEqual('Suspended', d['Versioning']) 120 121 def test_latest_version(self): 122 self.bucket.configure_versioning(versioning=True) 123 124 # add v1 of an object 125 key_name = "key" 126 kv1 = self.bucket.new_key(key_name) 127 kv1.set_contents_from_string("v1") 128 129 # read list which should contain latest v1 130 listed_kv1 = next(iter(self.bucket.get_all_versions())) 131 self.assertEqual(listed_kv1.name, key_name) 132 self.assertEqual(listed_kv1.version_id, kv1.version_id) 133 self.assertEqual(listed_kv1.is_latest, True) 134 135 # add v2 of the object 136 kv2 = self.bucket.new_key(key_name) 137 kv2.set_contents_from_string("v2") 138 139 # read 2 versions, confirm v2 is latest 140 i = iter(self.bucket.get_all_versions()) 141 listed_kv2 = next(i) 142 listed_kv1 = next(i) 143 self.assertEqual(listed_kv2.version_id, kv2.version_id) 144 self.assertEqual(listed_kv1.version_id, kv1.version_id) 145 self.assertEqual(listed_kv2.is_latest, True) 146 self.assertEqual(listed_kv1.is_latest, False) 147 148 # delete key, which creates a delete marker as latest 149 self.bucket.delete_key(key_name) 150 i = iter(self.bucket.get_all_versions()) 151 listed_kv3 = next(i) 152 listed_kv2 = next(i) 153 listed_kv1 = next(i) 154 self.assertNotEqual(listed_kv3.version_id, None) 155 self.assertEqual(listed_kv2.version_id, kv2.version_id) 156 self.assertEqual(listed_kv1.version_id, kv1.version_id) 157 self.assertEqual(listed_kv3.is_latest, True) 158 self.assertEqual(listed_kv2.is_latest, False) 159 self.assertEqual(listed_kv1.is_latest, False) 160