| OLD | NEW |
|---|
| (Empty) |
| 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 """ | |
| 25 Some unit tests for the S3 Versioning and MfaDelete | |
| 26 """ | |
| 27 | |
| 28 import unittest | |
| 29 import time | |
| 30 from boto.s3.connection import S3Connection | |
| 31 from boto.exception import S3ResponseError | |
| 32 from boto.s3.deletemarker import DeleteMarker | |
| 33 | |
| 34 class S3VersionTest (unittest.TestCase): | |
| 35 | |
| 36 def test_1_versions(self): | |
| 37 print '--- running S3Version tests ---' | |
| 38 c = S3Connection() | |
| 39 # create a new, empty bucket | |
| 40 bucket_name = 'version-%d' % int(time.time()) | |
| 41 bucket = c.create_bucket(bucket_name) | |
| 42 | |
| 43 # now try a get_bucket call and see if it's really there | |
| 44 bucket = c.get_bucket(bucket_name) | |
| 45 | |
| 46 # enable versions | |
| 47 d = bucket.get_versioning_status() | |
| 48 assert not d.has_key('Versioning') | |
| 49 bucket.configure_versioning(versioning=True) | |
| 50 time.sleep(5) | |
| 51 d = bucket.get_versioning_status() | |
| 52 assert d['Versioning'] == 'Enabled' | |
| 53 | |
| 54 # create a new key in the versioned bucket | |
| 55 k = bucket.new_key() | |
| 56 k.name = 'foobar' | |
| 57 s1 = 'This is a test of s3 versioning' | |
| 58 s2 = 'This is the second test of s3 versioning' | |
| 59 k.set_contents_from_string(s1) | |
| 60 time.sleep(5) | |
| 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() | |
| 67 | |
| 68 # check to make sure content read from s3 is identical to original | |
| 69 assert o1 == s1 | |
| 70 | |
| 71 # now overwrite that same key with new data | |
| 72 k.set_contents_from_string(s2) | |
| 73 v2 = k.version_id | |
| 74 time.sleep(5) | |
| 75 | |
| 76 # now retrieve the contents as a string and compare | |
| 77 s3 = k.get_contents_as_string(version_id=v2) | |
| 78 assert s3 == s2 | |
| 79 | |
| 80 # Now list all versions and compare to what we have | |
| 81 rs = bucket.get_all_versions() | |
| 82 assert rs[0].version_id == v2 | |
| 83 assert rs[1].version_id == v1 | |
| 84 | |
| 85 # Now do a regular list command and make sure only the new key shows up | |
| 86 rs = bucket.get_all_keys() | |
| 87 assert len(rs) == 1 | |
| 88 | |
| 89 # Now do regular delete | |
| 90 bucket.delete_key('foobar') | |
| 91 time.sleep(5) | |
| 92 | |
| 93 # Now list versions and make sure old versions are there | |
| 94 # plus the DeleteMarker | |
| 95 rs = bucket.get_all_versions() | |
| 96 assert len(rs) == 3 | |
| 97 assert isinstance(rs[0], DeleteMarker) | |
| 98 | |
| 99 # Now delete v1 of the key | |
| 100 bucket.delete_key('foobar', version_id=v1) | |
| 101 time.sleep(5) | |
| 102 | |
| 103 # Now list versions again and make sure v1 is not there | |
| 104 rs = bucket.get_all_versions() | |
| 105 versions = [k.version_id for k in rs] | |
| 106 assert v1 not in versions | |
| 107 assert v2 in versions | |
| 108 | |
| 109 # Now try to enable MfaDelete | |
| 110 mfa_sn = raw_input('MFA S/N: ') | |
| 111 mfa_code = raw_input('MFA Code: ') | |
| 112 bucket.configure_versioning(True, mfa_delete=True, mfa_token=(mfa_sn, mf
a_code)) | |
| 113 i = 0 | |
| 114 for i in range(1,8): | |
| 115 time.sleep(2**i) | |
| 116 d = bucket.get_versioning_status() | |
| 117 if d['Versioning'] == 'Enabled' and d['MfaDelete'] == 'Enabled': | |
| 118 break | |
| 119 assert d['Versioning'] == 'Enabled' | |
| 120 assert d['MfaDelete'] == 'Enabled' | |
| 121 | |
| 122 # Now try to delete v2 without the MFA token | |
| 123 try: | |
| 124 bucket.delete_key('foobar', version_id=v2) | |
| 125 except S3ResponseError: | |
| 126 pass | |
| 127 | |
| 128 # Now try to delete v2 with the MFA token | |
| 129 mfa_code = raw_input('MFA Code: ') | |
| 130 bucket.delete_key('foobar', version_id=v2, mfa_token=(mfa_sn, mfa_code)) | |
| 131 | |
| 132 # Now disable MfaDelete on the bucket | |
| 133 mfa_code = raw_input('MFA Code: ') | |
| 134 bucket.configure_versioning(True, mfa_delete=False, mfa_token=(mfa_sn, m
fa_code)) | |
| 135 | |
| 136 # Now suspend Versioning on the bucket | |
| 137 bucket.configure_versioning(False) | |
| 138 | |
| 139 # now delete all keys and deletemarkers in bucket | |
| 140 for k in bucket.list_versions(): | |
| 141 bucket.delete_key(k.name, version_id=k.version_id) | |
| 142 | |
| 143 # now delete bucket | |
| 144 c.delete_bucket(bucket) | |
| 145 print '--- tests completed ---' | |
| OLD | NEW |
|---|
Chromium Code Reviews
Issue 10199002:
Upgrade gsutil to 3.4 (Closed)
Base URL: https://dart.googlecode.com/svn/branches/bleeding_edge/dart