← Back to branch summary

~vishvananda/nova/network-refactor

« back to all changes in this revision

Viewing changes to vendor/boto/boto/tests/test_s3versioning.py

  • Committer: Jesse Andrews
  • Date: 2010-05-28 06:05:26 UTC
  • Revision ID: git-v1:bf6e6e718cdc7488e2da87b21e258ccc065fe499
initial commit

Show diffs side-by-side

added added

removed removed

Lines of Context:
vendor/boto/boto/tests/test_s3versioning.py
 
1
#!/usr/bin/env python
 
2
# -*- coding: utf-8 -*-
 
3
# Copyright (c) 2010 Mitch Garnaat http://garnaat.org/
 
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
import os
 
31
import urllib
 
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()
 
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, mfa_code))
 
115
        time.sleep(5)
 
116
        d = bucket.get_versioning_status()
 
117
        assert d['Versioning'] == 'Enabled'
 
118
        assert d['MfaDelete'] == 'Enabled'
 
119
        
 
120
        # Now try to delete v2 without the MFA token
 
121
        try:
 
122
            bucket.delete_key('foobar', version_id=v2)
 
123
        except S3ResponseError:
 
124
            pass
 
125
        
 
126
        # Now try to delete v2 with the MFA token
 
127
        mfa_code = raw_input('MFA Code: ')
 
128
        bucket.delete_key('foobar', version_id=v2, mfa_token=(mfa_sn, mfa_code))
 
129
 
 
130
        # Now disable MfaDelete on the bucket
 
131
        mfa_code = raw_input('MFA Code: ')
 
132
        bucket.configure_versioning(True, mfa_delete=False, mfa_token=(mfa_sn, mfa_code))
 
133
 
 
134
        # Now suspend Versioning on the bucket
 
135
        bucket.configure_versioning(False)
 
136
        
 
137
        # now delete all keys and deletemarkers in bucket
 
138
        for k in bucket.list_versions():
 
139
            bucket.delete_key(k.name, version_id=k.version_id)
 
140
 
 
141
        # now delete bucket
 
142
        c.delete_bucket(bucket)
 
143
        print '--- tests completed ---'