← Back to branch summary

~ubuntu-branches/ubuntu/trusty/python-boto/trusty

« back to all changes in this revision

Viewing changes to boto/tests/devpay_s3.py

  • Committer: Package Import Robot
  • Author(s): Eric Evans
  • Date: 2011-11-13 11:58:40 UTC
  • mfrom: (14.1.1 experimental)
  • Revision ID: package-import@ubuntu.com-20111113115840-ckzyt3h17uh8s41y
Tags: 2.0-2
http://bugs.debian.org/638931
Promote new upstream to unstable (Closes: #638931).

Show diffs side-by-side

added added

removed removed

Lines of Context:
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 ---'