3
# Copyright (c) 2006,2007 Mitch Garnaat http://garnaat.org/
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-
13
# The above copyright notice and this permission notice shall be included
14
# in all copies or substantial portions of the Software.
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
25
Some unit tests for the S3Connection
32
from boto.s3.connection import S3Connection
33
from boto.exception import S3PermissionsError
35
# this test requires a devpay product and user token to run:
37
AMAZON_USER_TOKEN = '{UserToken}...your token here...'
38
DEVPAY_HEADERS = { 'x-amz-security-token': AMAZON_USER_TOKEN }
40
print '--- running S3Connection tests (DevPay) ---'
42
# create a new, empty bucket
43
bucket_name = 'test-%d' % int(time.time())
44
bucket = c.create_bucket(bucket_name, headers=DEVPAY_HEADERS)
45
# now try a get_bucket call and see if it's really there
46
bucket = c.get_bucket(bucket_name, headers=DEVPAY_HEADERS)
48
logging_bucket = c.create_bucket(bucket_name + '-log', headers=DEVPAY_HEADERS)
49
logging_bucket.set_as_logging_target(headers=DEVPAY_HEADERS)
50
bucket.enable_logging(target_bucket=logging_bucket, target_prefix=bucket.name, headers=DEVPAY_HEADERS)
51
bucket.disable_logging(headers=DEVPAY_HEADERS)
52
c.delete_bucket(logging_bucket, headers=DEVPAY_HEADERS)
53
# create a new key and store it's content from a string
56
s1 = 'This is a test of file upload and download'
57
s2 = 'This is a second string to test file upload and download'
58
k.set_contents_from_string(s1, headers=DEVPAY_HEADERS)
59
fp = open('foobar', 'wb')
60
# now get the contents from s3 to a local file
61
k.get_contents_to_file(fp, headers=DEVPAY_HEADERS)
64
# check to make sure content read from s3 is identical to original
65
assert s1 == fp.read(), 'corrupted file'
68
url = k.generate_url(3600, headers=DEVPAY_HEADERS)
69
file = urllib.urlopen(url)
70
assert s1 == file.read(), 'invalid URL %s' % url
71
url = k.generate_url(3600, force_http=True, headers=DEVPAY_HEADERS)
72
file = urllib.urlopen(url)
73
assert s1 == file.read(), 'invalid URL %s' % url
74
bucket.delete_key(k, headers=DEVPAY_HEADERS)
75
# test a few variations on get_all_keys - first load some data
76
# for the first one, let's override the content type
77
phony_mimetype = 'application/x-boto-test'
78
headers = {'Content-Type': phony_mimetype}
79
headers.update(DEVPAY_HEADERS)
81
k.set_contents_from_string(s1, headers)
83
k.set_contents_from_filename('foobar', headers=DEVPAY_HEADERS)
85
k.set_contents_from_string(s1, headers=DEVPAY_HEADERS)
87
k.set_contents_from_string(s1, headers=DEVPAY_HEADERS)
89
k.set_contents_from_string(s1, headers=DEVPAY_HEADERS)
91
k.set_contents_from_string(s1, headers=DEVPAY_HEADERS)
92
# try resetting the contents to another value
94
k.set_contents_from_string(s2, headers=DEVPAY_HEADERS)
97
all = bucket.get_all_keys(headers=DEVPAY_HEADERS)
99
rs = bucket.get_all_keys(prefix='foo', headers=DEVPAY_HEADERS)
101
rs = bucket.get_all_keys(prefix='', delimiter='/', headers=DEVPAY_HEADERS)
103
rs = bucket.get_all_keys(maxkeys=5, headers=DEVPAY_HEADERS)
105
# test the lookup method
106
k = bucket.lookup('foo/bar', headers=DEVPAY_HEADERS)
107
assert isinstance(k, bucket.key_class)
108
assert k.content_type == phony_mimetype
109
k = bucket.lookup('notthere', headers=DEVPAY_HEADERS)
111
# try some metadata stuff
113
k.name = 'has_metadata'
115
mdval1 = 'This is the first metadata value'
116
k.set_metadata(mdkey1, mdval1)
118
mdval2 = 'This is the second metadata value'
119
k.set_metadata(mdkey2, mdval2)
120
k.set_contents_from_string(s1, headers=DEVPAY_HEADERS)
121
k = bucket.lookup('has_metadata', headers=DEVPAY_HEADERS)
122
assert k.get_metadata(mdkey1) == mdval1
123
assert k.get_metadata(mdkey2) == mdval2
125
k.name = 'has_metadata'
126
k.get_contents_as_string(headers=DEVPAY_HEADERS)
127
assert k.get_metadata(mdkey1) == mdval1
128
assert k.get_metadata(mdkey2) == mdval2
129
bucket.delete_key(k, headers=DEVPAY_HEADERS)
130
# test list and iterator
131
rs1 = bucket.list(headers=DEVPAY_HEADERS)
134
num_iter = num_iter + 1
135
rs = bucket.get_all_keys(headers=DEVPAY_HEADERS)
137
assert num_iter == num_keys
138
# try a key with a funny character
140
k.name = 'testnewline\n'
141
k.set_contents_from_string('This is a test', headers=DEVPAY_HEADERS)
142
rs = bucket.get_all_keys(headers=DEVPAY_HEADERS)
143
assert len(rs) == num_keys + 1
144
bucket.delete_key(k, headers=DEVPAY_HEADERS)
145
rs = bucket.get_all_keys(headers=DEVPAY_HEADERS)
146
assert len(rs) == num_keys
148
bucket.set_acl('public-read', headers=DEVPAY_HEADERS)
149
policy = bucket.get_acl(headers=DEVPAY_HEADERS)
150
assert len(policy.acl.grants) == 2
151
bucket.set_acl('private', headers=DEVPAY_HEADERS)
152
policy = bucket.get_acl(headers=DEVPAY_HEADERS)
153
assert len(policy.acl.grants) == 1
154
k = bucket.lookup('foo/bar', headers=DEVPAY_HEADERS)
155
k.set_acl('public-read', headers=DEVPAY_HEADERS)
156
policy = k.get_acl(headers=DEVPAY_HEADERS)
157
assert len(policy.acl.grants) == 2
158
k.set_acl('private', headers=DEVPAY_HEADERS)
159
policy = k.get_acl(headers=DEVPAY_HEADERS)
160
assert len(policy.acl.grants) == 1
161
# try the convenience methods for grants
162
# this doesn't work with devpay
163
#bucket.add_user_grant('FULL_CONTROL',
164
# 'c1e724fbfa0979a4448393c59a8c055011f739b6d102fb37a65f26414653cd67',
165
# headers=DEVPAY_HEADERS)
167
bucket.add_email_grant('foobar', 'foo@bar.com', headers=DEVPAY_HEADERS)
168
except S3PermissionsError:
170
# now delete all keys in bucket
172
bucket.delete_key(k, headers=DEVPAY_HEADERS)
175
c.delete_bucket(bucket, headers=DEVPAY_HEADERS)
177
print '--- tests completed ---'