Index: third_party/gsutil/boto/tests/integration/s3/test_mfa.py |
diff --git a/third_party/gsutil/boto/tests/integration/s3/test_mfa.py b/third_party/gsutil/boto/tests/integration/s3/test_mfa.py |
new file mode 100644 |
index 0000000000000000000000000000000000000000..1d6d62fc9e028e94ebf3a3e87a6fe7bfd90467dc |
--- /dev/null |
+++ b/third_party/gsutil/boto/tests/integration/s3/test_mfa.py |
@@ -0,0 +1,95 @@ |
+# Copyright (c) 2010 Mitch Garnaat http://garnaat.org/ |
+# Copyright (c) 2010, Eucalyptus Systems, Inc. |
+# All rights reserved. |
+# |
+# Permission is hereby granted, free of charge, to any person obtaining a |
+# copy of this software and associated documentation files (the |
+# "Software"), to deal in the Software without restriction, including |
+# without limitation the rights to use, copy, modify, merge, publish, dis- |
+# tribute, sublicense, and/or sell copies of the Software, and to permit |
+# persons to whom the Software is furnished to do so, subject to the fol- |
+# lowing conditions: |
+# |
+# The above copyright notice and this permission notice shall be included |
+# in all copies or substantial portions of the Software. |
+# |
+# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS |
+# OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABIL- |
+# ITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT |
+# SHALL THE AUTHOR BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, |
+# WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, |
+# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS |
+# IN THE SOFTWARE. |
+ |
+""" |
+Some unit tests for S3 MfaDelete with versioning |
+""" |
+ |
+import unittest |
+import time |
+from nose.plugins.attrib import attr |
+ |
+from boto.s3.connection import S3Connection |
+from boto.exception import S3ResponseError |
+from boto.s3.deletemarker import DeleteMarker |
+ |
+ |
+@attr('notdefault', 's3mfa') |
+class S3MFATest (unittest.TestCase): |
+ |
+ def setUp(self): |
+ self.conn = S3Connection() |
+ self.bucket_name = 'mfa-%d' % int(time.time()) |
+ self.bucket = self.conn.create_bucket(self.bucket_name) |
+ |
+ def tearDown(self): |
+ for k in self.bucket.list_versions(): |
+ self.bucket.delete_key(k.name, version_id=k.version_id) |
+ self.bucket.delete() |
+ |
+ def test_mfadel(self): |
+ # Enable Versioning with MfaDelete |
+ mfa_sn = raw_input('MFA S/N: ') |
+ mfa_code = raw_input('MFA Code: ') |
+ self.bucket.configure_versioning(True, mfa_delete=True, mfa_token=(mfa_sn, mfa_code)) |
+ |
+ # Check enabling mfa worked. |
+ i = 0 |
+ for i in range(1, 8): |
+ time.sleep(2**i) |
+ d = self.bucket.get_versioning_status() |
+ if d['Versioning'] == 'Enabled' and d['MfaDelete'] == 'Enabled': |
+ break |
+ self.assertEqual('Enabled', d['Versioning']) |
+ self.assertEqual('Enabled', d['MfaDelete']) |
+ |
+ # Add a key to the bucket |
+ k = self.bucket.new_key('foobar') |
+ s1 = 'This is v1' |
+ k.set_contents_from_string(s1) |
+ v1 = k.version_id |
+ |
+ # Now try to delete v1 without the MFA token |
+ try: |
+ self.bucket.delete_key('foobar', version_id=v1) |
+ self.fail("Must fail if not using MFA token") |
+ except S3ResponseError: |
+ pass |
+ |
+ # Now try delete again with the MFA token |
+ mfa_code = raw_input('MFA Code: ') |
+ self.bucket.delete_key('foobar', version_id=v1, mfa_token=(mfa_sn, mfa_code)) |
+ |
+ # Next suspend versioning and disable MfaDelete on the bucket |
+ mfa_code = raw_input('MFA Code: ') |
+ self.bucket.configure_versioning(False, mfa_delete=False, mfa_token=(mfa_sn, mfa_code)) |
+ |
+ # Lastly, check disabling mfa worked. |
+ i = 0 |
+ for i in range(1, 8): |
+ time.sleep(2**i) |
+ d = self.bucket.get_versioning_status() |
+ if d['Versioning'] == 'Suspended' and d['MfaDelete'] != 'Enabled': |
+ break |
+ self.assertEqual('Suspended', d['Versioning']) |
+ self.assertNotEqual('Enabled', d['MfaDelete']) |