| OLD | NEW |
| 1 # -*- coding: utf-8 -*- | 1 # -*- coding: utf-8 -*- |
| 2 # Copyright (c) 2006-2011 Mitch Garnaat http://garnaat.org/ | 2 # Copyright (c) 2006-2011 Mitch Garnaat http://garnaat.org/ |
| 3 # | 3 # |
| 4 # Permission is hereby granted, free of charge, to any person obtaining a | 4 # Permission is hereby granted, free of charge, to any person obtaining a |
| 5 # copy of this software and associated documentation files (the | 5 # copy of this software and associated documentation files (the |
| 6 # "Software"), to deal in the Software without restriction, including | 6 # "Software"), to deal in the Software without restriction, including |
| 7 # without limitation the rights to use, copy, modify, merge, publish, dis- | 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 | 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- | 9 # persons to whom the Software is furnished to do so, subject to the fol- |
| 10 # lowing conditions: | 10 # lowing conditions: |
| (...skipping 11 matching lines...) Expand all Loading... |
| 22 | 22 |
| 23 """ | 23 """ |
| 24 Some unit tests for the S3Connection | 24 Some unit tests for the S3Connection |
| 25 """ | 25 """ |
| 26 | 26 |
| 27 import unittest | 27 import unittest |
| 28 import time | 28 import time |
| 29 import os | 29 import os |
| 30 import urllib | 30 import urllib |
| 31 from boto.s3.connection import S3Connection | 31 from boto.s3.connection import S3Connection |
| 32 from boto.exception import S3PermissionsError | 32 from boto.s3.bucket import Bucket |
| 33 from boto.exception import S3PermissionsError, S3ResponseError |
| 33 | 34 |
| 34 class S3ConnectionTest (unittest.TestCase): | 35 class S3ConnectionTest (unittest.TestCase): |
| 35 | 36 |
| 36 def test_1_basic(self): | 37 def test_1_basic(self): |
| 37 print '--- running S3Connection tests ---' | 38 print '--- running S3Connection tests ---' |
| 38 c = S3Connection() | 39 c = S3Connection() |
| 39 # create a new, empty bucket | 40 # create a new, empty bucket |
| 40 bucket_name = 'test-%d' % int(time.time()) | 41 bucket_name = 'test-%d' % int(time.time()) |
| 41 bucket = c.create_bucket(bucket_name) | 42 bucket = c.create_bucket(bucket_name) |
| 42 # now try a get_bucket call and see if it's really there | 43 # now try a get_bucket call and see if it's really there |
| (...skipping 19 matching lines...) Expand all Loading... |
| 62 fp.close() | 63 fp.close() |
| 63 # test generated URLs | 64 # test generated URLs |
| 64 url = k.generate_url(3600) | 65 url = k.generate_url(3600) |
| 65 file = urllib.urlopen(url) | 66 file = urllib.urlopen(url) |
| 66 assert s1 == file.read(), 'invalid URL %s' % url | 67 assert s1 == file.read(), 'invalid URL %s' % url |
| 67 url = k.generate_url(3600, force_http=True) | 68 url = k.generate_url(3600, force_http=True) |
| 68 file = urllib.urlopen(url) | 69 file = urllib.urlopen(url) |
| 69 assert s1 == file.read(), 'invalid URL %s' % url | 70 assert s1 == file.read(), 'invalid URL %s' % url |
| 70 url = k.generate_url(3600, force_http=True, headers={'x-amz-x-token' : '
XYZ'}) | 71 url = k.generate_url(3600, force_http=True, headers={'x-amz-x-token' : '
XYZ'}) |
| 71 file = urllib.urlopen(url) | 72 file = urllib.urlopen(url) |
| 73 rh = {'response-content-disposition': 'attachment; filename="foo.txt"'} |
| 74 url = k.generate_url(60, response_headers=rh) |
| 72 assert s1 == file.read(), 'invalid URL %s' % url | 75 assert s1 == file.read(), 'invalid URL %s' % url |
| 73 bucket.delete_key(k) | 76 bucket.delete_key(k) |
| 74 # test a few variations on get_all_keys - first load some data | 77 # test a few variations on get_all_keys - first load some data |
| 75 # for the first one, let's override the content type | 78 # for the first one, let's override the content type |
| 76 phony_mimetype = 'application/x-boto-test' | 79 phony_mimetype = 'application/x-boto-test' |
| 77 headers = {'Content-Type': phony_mimetype} | 80 headers = {'Content-Type': phony_mimetype} |
| 78 k.name = 'foo/bar' | 81 k.name = 'foo/bar' |
| 79 k.set_contents_from_string(s1, headers) | 82 k.set_contents_from_string(s1, headers) |
| 80 k.name = 'foo/bas' | 83 k.name = 'foo/bas' |
| 81 k.set_contents_from_filename('foobar') | 84 k.set_contents_from_filename('foobar') |
| (...skipping 98 matching lines...) Expand 10 before | Expand all | Expand 10 after Loading... |
| 180 | 183 |
| 181 # now delete all keys in bucket | 184 # now delete all keys in bucket |
| 182 for k in bucket: | 185 for k in bucket: |
| 183 if k.name == 'reduced_redundancy': | 186 if k.name == 'reduced_redundancy': |
| 184 assert k.storage_class == 'REDUCED_REDUNDANCY' | 187 assert k.storage_class == 'REDUCED_REDUNDANCY' |
| 185 bucket.delete_key(k) | 188 bucket.delete_key(k) |
| 186 # now delete bucket | 189 # now delete bucket |
| 187 time.sleep(5) | 190 time.sleep(5) |
| 188 c.delete_bucket(bucket) | 191 c.delete_bucket(bucket) |
| 189 print '--- tests completed ---' | 192 print '--- tests completed ---' |
| 193 |
| 194 def test_basic_anon(self): |
| 195 auth_con = S3Connection() |
| 196 # create a new, empty bucket |
| 197 bucket_name = 'test-%d' % int(time.time()) |
| 198 auth_bucket = auth_con.create_bucket(bucket_name) |
| 199 |
| 200 # try read the bucket anonymously |
| 201 anon_con = S3Connection(anon=True) |
| 202 anon_bucket = Bucket(anon_con, bucket_name) |
| 203 try: |
| 204 iter(anon_bucket.list()).next() |
| 205 self.fail("anon bucket list should fail") |
| 206 except S3ResponseError: |
| 207 pass |
| 208 |
| 209 # give bucket anon user access and anon read again |
| 210 auth_bucket.set_acl('public-read') |
| 211 try: |
| 212 iter(anon_bucket.list()).next() |
| 213 self.fail("not expecting contents") |
| 214 except S3ResponseError: |
| 215 self.fail("we should have public-read access.") |
| 216 except StopIteration: |
| 217 pass |
| 218 |
| 219 # cleanup |
| 220 auth_con.delete_bucket(auth_bucket) |
| OLD | NEW |