| OLD | NEW |
|---|
| (Empty) |
| 1 #!/usr/bin/env python | |
| 2 # -*- coding: utf-8 -*- | |
| 3 # Copyright (c) 2010 Mitch Garnaat http://garnaat.org/ | |
| 4 # Copyright (c) 2010, Eucalyptus Systems, Inc. | |
| 5 # All rights reserved. | |
| 6 # | |
| 7 # Permission is hereby granted, free of charge, to any person obtaining a | |
| 8 # copy of this software and associated documentation files (the | |
| 9 # "Software"), to deal in the Software without restriction, including | |
| 10 # without limitation the rights to use, copy, modify, merge, publish, dis- | |
| 11 # tribute, sublicense, and/or sell copies of the Software, and to permit | |
| 12 # persons to whom the Software is furnished to do so, subject to the fol- | |
| 13 # lowing conditions: | |
| 14 # | |
| 15 # The above copyright notice and this permission notice shall be included | |
| 16 # in all copies or substantial portions of the Software. | |
| 17 # | |
| 18 # THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS | |
| 19 # OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABIL- | |
| 20 # ITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT | |
| 21 # SHALL THE AUTHOR BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, | |
| 22 # WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, | |
| 23 # OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS | |
| 24 # IN THE SOFTWARE. | |
| 25 | |
| 26 """ | |
| 27 Some unit tests for the S3 Versioning and MfaDelete | |
| 28 """ | |
| 29 | |
| 30 import unittest | |
| 31 import time | |
| 32 from boto.s3.connection import S3Connection | |
| 33 from boto.exception import S3ResponseError | |
| 34 from boto.s3.deletemarker import DeleteMarker | |
| 35 | |
| 36 class S3VersionTest (unittest.TestCase): | |
| 37 | |
| 38 def test_1_versions(self): | |
| 39 print '--- running S3Version tests ---' | |
| 40 c = S3Connection() | |
| 41 # create a new, empty bucket | |
| 42 bucket_name = 'version-%d' % int(time.time()) | |
| 43 bucket = c.create_bucket(bucket_name) | |
| 44 | |
| 45 # now try a get_bucket call and see if it's really there | |
| 46 bucket = c.get_bucket(bucket_name) | |
| 47 | |
| 48 # enable versions | |
| 49 d = bucket.get_versioning_status() | |
| 50 assert not d.has_key('Versioning') | |
| 51 bucket.configure_versioning(versioning=True) | |
| 52 time.sleep(5) | |
| 53 d = bucket.get_versioning_status() | |
| 54 assert d['Versioning'] == 'Enabled' | |
| 55 | |
| 56 # create a new key in the versioned bucket | |
| 57 k = bucket.new_key() | |
| 58 k.name = 'foobar' | |
| 59 s1 = 'This is a test of s3 versioning' | |
| 60 s2 = 'This is the second test of s3 versioning' | |
| 61 k.set_contents_from_string(s1) | |
| 62 time.sleep(5) | |
| 63 | |
| 64 # remember the version id of this object | |
| 65 v1 = k.version_id | |
| 66 | |
| 67 # now get the contents from s3 | |
| 68 o1 = k.get_contents_as_string() | |
| 69 | |
| 70 # check to make sure content read from s3 is identical to original | |
| 71 assert o1 == s1 | |
| 72 | |
| 73 # now overwrite that same key with new data | |
| 74 k.set_contents_from_string(s2) | |
| 75 v2 = k.version_id | |
| 76 time.sleep(5) | |
| 77 | |
| 78 # now retrieve the contents as a string and compare | |
| 79 s3 = k.get_contents_as_string(version_id=v2) | |
| 80 assert s3 == s2 | |
| 81 | |
| 82 # Now list all versions and compare to what we have | |
| 83 rs = bucket.get_all_versions() | |
| 84 assert rs[0].version_id == v2 | |
| 85 assert rs[1].version_id == v1 | |
| 86 | |
| 87 # Now do a regular list command and make sure only the new key shows up | |
| 88 rs = bucket.get_all_keys() | |
| 89 assert len(rs) == 1 | |
| 90 | |
| 91 # Now do regular delete | |
| 92 bucket.delete_key('foobar') | |
| 93 time.sleep(5) | |
| 94 | |
| 95 # Now list versions and make sure old versions are there | |
| 96 # plus the DeleteMarker | |
| 97 rs = bucket.get_all_versions() | |
| 98 assert len(rs) == 3 | |
| 99 assert isinstance(rs[0], DeleteMarker) | |
| 100 | |
| 101 # Now delete v1 of the key | |
| 102 bucket.delete_key('foobar', version_id=v1) | |
| 103 time.sleep(5) | |
| 104 | |
| 105 # Now list versions again and make sure v1 is not there | |
| 106 rs = bucket.get_all_versions() | |
| 107 versions = [k.version_id for k in rs] | |
| 108 assert v1 not in versions | |
| 109 assert v2 in versions | |
| 110 | |
| 111 # Now try to enable MfaDelete | |
| 112 mfa_sn = raw_input('MFA S/N: ') | |
| 113 mfa_code = raw_input('MFA Code: ') | |
| 114 bucket.configure_versioning(True, mfa_delete=True, mfa_token=(mfa_sn, mf
a_code)) | |
| 115 i = 0 | |
| 116 for i in range(1,8): | |
| 117 time.sleep(2**i) | |
| 118 d = bucket.get_versioning_status() | |
| 119 if d['Versioning'] == 'Enabled' and d['MfaDelete'] == 'Enabled': | |
| 120 break | |
| 121 assert d['Versioning'] == 'Enabled' | |
| 122 assert d['MfaDelete'] == 'Enabled' | |
| 123 | |
| 124 # Now try to delete v2 without the MFA token | |
| 125 try: | |
| 126 bucket.delete_key('foobar', version_id=v2) | |
| 127 except S3ResponseError: | |
| 128 pass | |
| 129 | |
| 130 # Now try to delete v2 with the MFA token | |
| 131 mfa_code = raw_input('MFA Code: ') | |
| 132 bucket.delete_key('foobar', version_id=v2, mfa_token=(mfa_sn, mfa_code)) | |
| 133 | |
| 134 # Now disable MfaDelete on the bucket | |
| 135 mfa_code = raw_input('MFA Code: ') | |
| 136 bucket.configure_versioning(True, mfa_delete=False, mfa_token=(mfa_sn, m
fa_code)) | |
| 137 | |
| 138 # Now suspend Versioning on the bucket | |
| 139 bucket.configure_versioning(False) | |
| 140 | |
| 141 # now delete all keys and deletemarkers in bucket | |
| 142 for k in bucket.list_versions(): | |
| 143 bucket.delete_key(k.name, version_id=k.version_id) | |
| 144 | |
| 145 # now delete bucket | |
| 146 c.delete_bucket(bucket) | |
| 147 print '--- tests completed ---' | |
| OLD | NEW |
|---|
Chromium Code Reviews
