| OLD | NEW |
|---|
| (Empty) |
| 1 # -*- coding: utf-8 -*- | |
| 2 # Copyright (c) 2006-2011 Mitch Garnaat http://garnaat.org/ | |
| 3 # | |
| 4 # Permission is hereby granted, free of charge, to any person obtaining a | |
| 5 # copy of this software and associated documentation files (the | |
| 6 # "Software"), to deal in the Software without restriction, including | |
| 7 # without limitation the rights to use, copy, modify, merge, publish, dis- | |
| 8 # tribute, sublicense, and/or sell copies of the Software, and to permit | |
| 9 # persons to whom the Software is furnished to do so, subject to the fol- | |
| 10 # lowing conditions: | |
| 11 # | |
| 12 # The above copyright notice and this permission notice shall be included | |
| 13 # in all copies or substantial portions of the Software. | |
| 14 # | |
| 15 # THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS | |
| 16 # OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABIL- | |
| 17 # ITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT | |
| 18 # SHALL THE AUTHOR BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, | |
| 19 # WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, | |
| 20 # OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS | |
| 21 # IN THE SOFTWARE. | |
| 22 | |
| 23 """ | |
| 24 Some unit tests for the S3Connection | |
| 25 """ | |
| 26 | |
| 27 import unittest | |
| 28 import time | |
| 29 import os | |
| 30 import urllib | |
| 31 from boto.s3.connection import S3Connection | |
| 32 from boto.exception import S3PermissionsError | |
| 33 | |
| 34 class S3ConnectionTest (unittest.TestCase): | |
| 35 | |
| 36 def test_1_basic(self): | |
| 37 print '--- running S3Connection tests ---' | |
| 38 c = S3Connection() | |
| 39 # create a new, empty bucket | |
| 40 bucket_name = 'test-%d' % int(time.time()) | |
| 41 bucket = c.create_bucket(bucket_name) | |
| 42 # now try a get_bucket call and see if it's really there | |
| 43 bucket = c.get_bucket(bucket_name) | |
| 44 # test logging | |
| 45 logging_bucket = c.create_bucket(bucket_name + '-log') | |
| 46 logging_bucket.set_as_logging_target() | |
| 47 bucket.enable_logging(target_bucket=logging_bucket, target_prefix=bucket
.name) | |
| 48 bucket.disable_logging() | |
| 49 c.delete_bucket(logging_bucket) | |
| 50 k = bucket.new_key() | |
| 51 k.name = 'foobar' | |
| 52 s1 = 'This is a test of file upload and download' | |
| 53 s2 = 'This is a second string to test file upload and download' | |
| 54 k.set_contents_from_string(s1) | |
| 55 fp = open('foobar', 'wb') | |
| 56 # now get the contents from s3 to a local file | |
| 57 k.get_contents_to_file(fp) | |
| 58 fp.close() | |
| 59 fp = open('foobar') | |
| 60 # check to make sure content read from s3 is identical to original | |
| 61 assert s1 == fp.read(), 'corrupted file' | |
| 62 fp.close() | |
| 63 # test generated URLs | |
| 64 url = k.generate_url(3600) | |
| 65 file = urllib.urlopen(url) | |
| 66 assert s1 == file.read(), 'invalid URL %s' % url | |
| 67 url = k.generate_url(3600, force_http=True) | |
| 68 file = urllib.urlopen(url) | |
| 69 assert s1 == file.read(), 'invalid URL %s' % url | |
| 70 url = k.generate_url(3600, force_http=True, headers={'x-amz-x-token' : '
XYZ'}) | |
| 71 file = urllib.urlopen(url) | |
| 72 assert s1 == file.read(), 'invalid URL %s' % url | |
| 73 bucket.delete_key(k) | |
| 74 # test a few variations on get_all_keys - first load some data | |
| 75 # for the first one, let's override the content type | |
| 76 phony_mimetype = 'application/x-boto-test' | |
| 77 headers = {'Content-Type': phony_mimetype} | |
| 78 k.name = 'foo/bar' | |
| 79 k.set_contents_from_string(s1, headers) | |
| 80 k.name = 'foo/bas' | |
| 81 k.set_contents_from_filename('foobar') | |
| 82 k.name = 'foo/bat' | |
| 83 k.set_contents_from_string(s1) | |
| 84 k.name = 'fie/bar' | |
| 85 k.set_contents_from_string(s1) | |
| 86 k.name = 'fie/bas' | |
| 87 k.set_contents_from_string(s1) | |
| 88 k.name = 'fie/bat' | |
| 89 k.set_contents_from_string(s1) | |
| 90 # try resetting the contents to another value | |
| 91 md5 = k.md5 | |
| 92 k.set_contents_from_string(s2) | |
| 93 assert k.md5 != md5 | |
| 94 os.unlink('foobar') | |
| 95 all = bucket.get_all_keys() | |
| 96 assert len(all) == 6 | |
| 97 rs = bucket.get_all_keys(prefix='foo') | |
| 98 assert len(rs) == 3 | |
| 99 rs = bucket.get_all_keys(prefix='', delimiter='/') | |
| 100 assert len(rs) == 2 | |
| 101 rs = bucket.get_all_keys(maxkeys=5) | |
| 102 assert len(rs) == 5 | |
| 103 # test the lookup method | |
| 104 k = bucket.lookup('foo/bar') | |
| 105 assert isinstance(k, bucket.key_class) | |
| 106 assert k.content_type == phony_mimetype | |
| 107 k = bucket.lookup('notthere') | |
| 108 assert k == None | |
| 109 # try some metadata stuff | |
| 110 k = bucket.new_key() | |
| 111 k.name = 'has_metadata' | |
| 112 mdkey1 = 'meta1' | |
| 113 mdval1 = 'This is the first metadata value' | |
| 114 k.set_metadata(mdkey1, mdval1) | |
| 115 mdkey2 = 'meta2' | |
| 116 mdval2 = 'This is the second metadata value' | |
| 117 k.set_metadata(mdkey2, mdval2) | |
| 118 # try a unicode metadata value | |
| 119 mdval3 = u'föö' | |
| 120 mdkey3 = 'meta3' | |
| 121 k.set_metadata(mdkey3, mdval3) | |
| 122 k.set_contents_from_string(s1) | |
| 123 k = bucket.lookup('has_metadata') | |
| 124 assert k.get_metadata(mdkey1) == mdval1 | |
| 125 assert k.get_metadata(mdkey2) == mdval2 | |
| 126 assert k.get_metadata(mdkey3) == mdval3 | |
| 127 k = bucket.new_key() | |
| 128 k.name = 'has_metadata' | |
| 129 k.get_contents_as_string() | |
| 130 assert k.get_metadata(mdkey1) == mdval1 | |
| 131 assert k.get_metadata(mdkey2) == mdval2 | |
| 132 assert k.get_metadata(mdkey3) == mdval3 | |
| 133 bucket.delete_key(k) | |
| 134 # test list and iterator | |
| 135 rs1 = bucket.list() | |
| 136 num_iter = 0 | |
| 137 for r in rs1: | |
| 138 num_iter = num_iter + 1 | |
| 139 rs = bucket.get_all_keys() | |
| 140 num_keys = len(rs) | |
| 141 assert num_iter == num_keys | |
| 142 # try a key with a funny character | |
| 143 k = bucket.new_key() | |
| 144 k.name = 'testnewline\n' | |
| 145 k.set_contents_from_string('This is a test') | |
| 146 rs = bucket.get_all_keys() | |
| 147 assert len(rs) == num_keys + 1 | |
| 148 bucket.delete_key(k) | |
| 149 rs = bucket.get_all_keys() | |
| 150 assert len(rs) == num_keys | |
| 151 # try some acl stuff | |
| 152 bucket.set_acl('public-read') | |
| 153 policy = bucket.get_acl() | |
| 154 assert len(policy.acl.grants) == 2 | |
| 155 bucket.set_acl('private') | |
| 156 policy = bucket.get_acl() | |
| 157 assert len(policy.acl.grants) == 1 | |
| 158 k = bucket.lookup('foo/bar') | |
| 159 k.set_acl('public-read') | |
| 160 policy = k.get_acl() | |
| 161 assert len(policy.acl.grants) == 2 | |
| 162 k.set_acl('private') | |
| 163 policy = k.get_acl() | |
| 164 assert len(policy.acl.grants) == 1 | |
| 165 # try the convenience methods for grants | |
| 166 bucket.add_user_grant('FULL_CONTROL', | |
| 167 'c1e724fbfa0979a4448393c59a8c055011f739b6d102fb37a
65f26414653cd67') | |
| 168 try: | |
| 169 bucket.add_email_grant('foobar', 'foo@bar.com') | |
| 170 except S3PermissionsError: | |
| 171 pass | |
| 172 # now try to create an RRS key | |
| 173 k = bucket.new_key('reduced_redundancy') | |
| 174 k.set_contents_from_string('This key has reduced redundancy', | |
| 175 reduced_redundancy=True) | |
| 176 | |
| 177 # now try to inject a response header | |
| 178 data = k.get_contents_as_string(response_headers={'response-content-type
' : 'foo/bar'}) | |
| 179 assert k.content_type == 'foo/bar' | |
| 180 | |
| 181 # now delete all keys in bucket | |
| 182 for k in bucket: | |
| 183 if k.name == 'reduced_redundancy': | |
| 184 assert k.storage_class == 'REDUCED_REDUNDANCY' | |
| 185 bucket.delete_key(k) | |
| 186 # now delete bucket | |
| 187 time.sleep(5) | |
| 188 c.delete_bucket(bucket) | |
| 189 print '--- tests completed ---' | |
| OLD | NEW |
|---|
Chromium Code Reviews
Issue 10199002:
Upgrade gsutil to 3.4 (Closed)
Base URL: https://dart.googlecode.com/svn/branches/bleeding_edge/dart