| Index: tools/telemetry/third_party/gsutilz/third_party/boto/tests/integration/s3/test_versioning.py
|
| diff --git a/tools/telemetry/third_party/gsutilz/third_party/boto/tests/integration/s3/test_versioning.py b/tools/telemetry/third_party/gsutilz/third_party/boto/tests/integration/s3/test_versioning.py
|
| deleted file mode 100644
|
| index 6758212ef63a88937639d53a85f5081310192518..0000000000000000000000000000000000000000
|
| --- a/tools/telemetry/third_party/gsutilz/third_party/boto/tests/integration/s3/test_versioning.py
|
| +++ /dev/null
|
| @@ -1,159 +0,0 @@
|
| -# 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 the S3 Versioning.
|
| -"""
|
| -
|
| -import unittest
|
| -import time
|
| -from boto.s3.connection import S3Connection
|
| -from boto.exception import S3ResponseError
|
| -from boto.s3.deletemarker import DeleteMarker
|
| -from boto.compat import six
|
| -
|
| -class S3VersionTest (unittest.TestCase):
|
| -
|
| - def setUp(self):
|
| - self.conn = S3Connection()
|
| - self.bucket_name = 'version-%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_1_versions(self):
|
| - # check versioning off
|
| - d = self.bucket.get_versioning_status()
|
| - self.assertFalse('Versioning' in d)
|
| -
|
| - # enable versioning
|
| - self.bucket.configure_versioning(versioning=True)
|
| - d = self.bucket.get_versioning_status()
|
| - self.assertEqual('Enabled', d['Versioning'])
|
| -
|
| - # create a new key in the versioned bucket
|
| - k = self.bucket.new_key("foobar")
|
| - s1 = 'This is v1'
|
| - k.set_contents_from_string(s1)
|
| -
|
| - # remember the version id of this object
|
| - v1 = k.version_id
|
| -
|
| - # now get the contents from s3
|
| - o1 = k.get_contents_as_string().decode('utf-8')
|
| -
|
| - # check to make sure content read from k is identical to original
|
| - self.assertEqual(s1, o1)
|
| -
|
| - # now overwrite that same key with new data
|
| - s2 = 'This is v2'
|
| - k.set_contents_from_string(s2)
|
| - v2 = k.version_id
|
| -
|
| - # now retrieve latest contents as a string and compare
|
| - k2 = self.bucket.new_key("foobar")
|
| - o2 = k2.get_contents_as_string().decode('utf-8')
|
| - self.assertEqual(s2, o2)
|
| -
|
| - # next retrieve explicit versions and compare
|
| - o1 = k.get_contents_as_string(version_id=v1).decode('utf-8')
|
| - o2 = k.get_contents_as_string(version_id=v2).decode('utf-8')
|
| - self.assertEqual(s1, o1)
|
| - self.assertEqual(s2, o2)
|
| -
|
| - # Now list all versions and compare to what we have
|
| - rs = self.bucket.get_all_versions()
|
| - self.assertEqual(v2, rs[0].version_id)
|
| - self.assertEqual(v1, rs[1].version_id)
|
| -
|
| - # Now do a regular list command and make sure only the new key shows up
|
| - rs = self.bucket.get_all_keys()
|
| - self.assertEqual(1, len(rs))
|
| -
|
| - # Now do regular delete
|
| - self.bucket.delete_key('foobar')
|
| -
|
| - # Now list versions and make sure old versions are there
|
| - # plus the DeleteMarker which is latest.
|
| - rs = self.bucket.get_all_versions()
|
| - self.assertEqual(3, len(rs))
|
| - self.assertTrue(isinstance(rs[0], DeleteMarker))
|
| -
|
| - # Now delete v1 of the key
|
| - self.bucket.delete_key('foobar', version_id=v1)
|
| -
|
| - # Now list versions again and make sure v1 is not there
|
| - rs = self.bucket.get_all_versions()
|
| - versions = [k.version_id for k in rs]
|
| - self.assertTrue(v1 not in versions)
|
| - self.assertTrue(v2 in versions)
|
| -
|
| - # Now suspend Versioning on the bucket
|
| - self.bucket.configure_versioning(False)
|
| - # Allow time for the change to fully propagate.
|
| - time.sleep(3)
|
| - d = self.bucket.get_versioning_status()
|
| - self.assertEqual('Suspended', d['Versioning'])
|
| -
|
| - def test_latest_version(self):
|
| - self.bucket.configure_versioning(versioning=True)
|
| -
|
| - # add v1 of an object
|
| - key_name = "key"
|
| - kv1 = self.bucket.new_key(key_name)
|
| - kv1.set_contents_from_string("v1")
|
| -
|
| - # read list which should contain latest v1
|
| - listed_kv1 = next(iter(self.bucket.get_all_versions()))
|
| - self.assertEqual(listed_kv1.name, key_name)
|
| - self.assertEqual(listed_kv1.version_id, kv1.version_id)
|
| - self.assertEqual(listed_kv1.is_latest, True)
|
| -
|
| - # add v2 of the object
|
| - kv2 = self.bucket.new_key(key_name)
|
| - kv2.set_contents_from_string("v2")
|
| -
|
| - # read 2 versions, confirm v2 is latest
|
| - i = iter(self.bucket.get_all_versions())
|
| - listed_kv2 = next(i)
|
| - listed_kv1 = next(i)
|
| - self.assertEqual(listed_kv2.version_id, kv2.version_id)
|
| - self.assertEqual(listed_kv1.version_id, kv1.version_id)
|
| - self.assertEqual(listed_kv2.is_latest, True)
|
| - self.assertEqual(listed_kv1.is_latest, False)
|
| -
|
| - # delete key, which creates a delete marker as latest
|
| - self.bucket.delete_key(key_name)
|
| - i = iter(self.bucket.get_all_versions())
|
| - listed_kv3 = next(i)
|
| - listed_kv2 = next(i)
|
| - listed_kv1 = next(i)
|
| - self.assertNotEqual(listed_kv3.version_id, None)
|
| - self.assertEqual(listed_kv2.version_id, kv2.version_id)
|
| - self.assertEqual(listed_kv1.version_id, kv1.version_id)
|
| - self.assertEqual(listed_kv3.is_latest, True)
|
| - self.assertEqual(listed_kv2.is_latest, False)
|
| - self.assertEqual(listed_kv1.is_latest, False)
|
|
|