← Back to branch summary

~justin-fathomdb/nova/justinsb-openstack-api-volumes

« back to all changes in this revision

Viewing changes to vendor/boto/boto/tests/devpay_s3.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/devpay_s3.py
 
1
#!/usr/bin/env python
 
2
 
 
3
# Copyright (c) 2006,2007 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 S3Connection
 
26
"""
 
27
 
 
28
import time
 
29
import os
 
30
import urllib
 
31
 
 
32
from boto.s3.connection import S3Connection
 
33
from boto.exception import S3PermissionsError
 
34
 
 
35
# this test requires a devpay product and user token to run:
 
36
 
 
37
AMAZON_USER_TOKEN = '{UserToken}...your token here...'
 
38
DEVPAY_HEADERS = { 'x-amz-security-token': AMAZON_USER_TOKEN }
 
39
 
 
40
print '--- running S3Connection tests (DevPay) ---'
 
41
c = S3Connection()
 
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)
 
47
# test logging
 
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
 
54
k = bucket.new_key()
 
55
k.name = 'foobar'
 
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)
 
62
fp.close()
 
63
fp = open('foobar')
 
64
# check to make sure content read from s3 is identical to original
 
65
assert s1 == fp.read(), 'corrupted file'
 
66
fp.close()
 
67
# test generated URLs
 
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)
 
80
k.name = 'foo/bar'
 
81
k.set_contents_from_string(s1, headers)
 
82
k.name = 'foo/bas'
 
83
k.set_contents_from_filename('foobar', headers=DEVPAY_HEADERS)
 
84
k.name = 'foo/bat'
 
85
k.set_contents_from_string(s1, headers=DEVPAY_HEADERS)
 
86
k.name = 'fie/bar'
 
87
k.set_contents_from_string(s1, headers=DEVPAY_HEADERS)
 
88
k.name = 'fie/bas'
 
89
k.set_contents_from_string(s1, headers=DEVPAY_HEADERS)
 
90
k.name = 'fie/bat'
 
91
k.set_contents_from_string(s1, headers=DEVPAY_HEADERS)
 
92
# try resetting the contents to another value
 
93
md5 = k.md5
 
94
k.set_contents_from_string(s2, headers=DEVPAY_HEADERS)
 
95
assert k.md5 != md5
 
96
os.unlink('foobar')
 
97
all = bucket.get_all_keys(headers=DEVPAY_HEADERS)
 
98
assert len(all) == 6
 
99
rs = bucket.get_all_keys(prefix='foo', headers=DEVPAY_HEADERS)
 
100
assert len(rs) == 3
 
101
rs = bucket.get_all_keys(prefix='', delimiter='/', headers=DEVPAY_HEADERS)
 
102
assert len(rs) == 2
 
103
rs = bucket.get_all_keys(maxkeys=5, headers=DEVPAY_HEADERS)
 
104
assert len(rs) == 5
 
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)
 
110
assert k == None
 
111
# try some metadata stuff
 
112
k = bucket.new_key()
 
113
k.name = 'has_metadata'
 
114
mdkey1 = 'meta1'
 
115
mdval1 = 'This is the first metadata value'
 
116
k.set_metadata(mdkey1, mdval1)
 
117
mdkey2 = 'meta2'
 
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
 
124
k = bucket.new_key()
 
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)
 
132
num_iter = 0
 
133
for r in rs1:
 
134
    num_iter = num_iter + 1
 
135
rs = bucket.get_all_keys(headers=DEVPAY_HEADERS)
 
136
num_keys = len(rs)
 
137
assert num_iter == num_keys
 
138
# try a key with a funny character
 
139
k = bucket.new_key()
 
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
 
147
# try some acl stuff
 
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)
 
166
try:
 
167
    bucket.add_email_grant('foobar', 'foo@bar.com', headers=DEVPAY_HEADERS)
 
168
except S3PermissionsError:
 
169
    pass
 
170
# now delete all keys in bucket
 
171
for k in all:
 
172
    bucket.delete_key(k, headers=DEVPAY_HEADERS)
 
173
# now delete bucket
 
174
 
 
175
c.delete_bucket(bucket, headers=DEVPAY_HEADERS)
 
176
 
 
177
print '--- tests completed ---'