OLD | NEW |
| (Empty) |
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, h
eaders=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 # 'c1e724fbfa0979a4448393c59a8c055011f739b6d102fb37a65f2641
4653cd67', | |
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 ---' | |
OLD | NEW |