千家信息网

s3cmd versioning测试方法是什么

发表于:2025-12-02 作者:千家信息网编辑
千家信息网最后更新 2025年12月02日,这篇文章主要讲解了"s3cmd versioning测试方法是什么",文中的讲解内容简单清晰,易于学习与理解,下面请大家跟着小编的思路慢慢深入,一起来研究和学习"s3cmd versioning测试方
千家信息网最后更新 2025年12月02日s3cmd versioning测试方法是什么

这篇文章主要讲解了"s3cmd versioning测试方法是什么",文中的讲解内容简单清晰,易于学习与理解,下面请大家跟着小编的思路慢慢深入,一起来研究和学习"s3cmd versioning测试方法是什么"吧!

"""Some unit tests for the S3 Versioning."""import unittestimport timefrom boto.s3.connection import S3Connectionfrom boto.exception import S3ResponseErrorfrom boto.s3.deletemarker import DeleteMarkerclass 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()                # 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()        self.assertEqual(s2, o2)        # next retrieve explicit versions and compare        o1 = k.get_contents_as_string(version_id=v1)        o2 = k.get_contents_as_string(version_id=v2)        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 = iter(self.bucket.get_all_versions()).next()        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 = i.next()        listed_kv1 = i.next()        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 = i.next()        listed_kv2 = i.next()        listed_kv1 = i.next()        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)

感谢各位的阅读,以上就是"s3cmd versioning测试方法是什么"的内容了,经过本文的学习后,相信大家对s3cmd versioning测试方法是什么这一问题有了更深刻的体会,具体使用情况还需要大家实践验证。这里是,小编将为大家推送更多相关知识点的文章,欢迎关注!

0