| OLD | NEW |
|---|
| (Empty) |
| 1 #!/usr/bin/env python | |
| 2 # -*- coding: utf-8 -*- | |
| 3 # Copyright (c) 2006-2010 Mitch Garnaat http://garnaat.org/ | |
| 4 # Copyright (c) 2010, Eucalyptus Systems, Inc. | |
| 5 # All rights reserved. | |
| 6 # | |
| 7 # Permission is hereby granted, free of charge, to any person obtaining a | |
| 8 # copy of this software and associated documentation files (the | |
| 9 # "Software"), to deal in the Software without restriction, including | |
| 10 # without limitation the rights to use, copy, modify, merge, publish, dis- | |
| 11 # tribute, sublicense, and/or sell copies of the Software, and to permit | |
| 12 # persons to whom the Software is furnished to do so, subject to the fol- | |
| 13 # lowing conditions: | |
| 14 # | |
| 15 # The above copyright notice and this permission notice shall be included | |
| 16 # in all copies or substantial portions of the Software. | |
| 17 # | |
| 18 # THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS | |
| 19 # OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABIL- | |
| 20 # ITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT | |
| 21 # SHALL THE AUTHOR BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, | |
| 22 # WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, | |
| 23 # OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS | |
| 24 # IN THE SOFTWARE. | |
| 25 | |
| 26 """ | |
| 27 Some unit tests for the GSConnection | |
| 28 """ | |
| 29 | |
| 30 import unittest | |
| 31 import time | |
| 32 import os | |
| 33 from boto.gs.connection import GSConnection | |
| 34 | |
| 35 class GSConnectionTest (unittest.TestCase): | |
| 36 | |
| 37 def test_1_basic(self): | |
| 38 print '--- running GSConnection tests ---' | |
| 39 c = GSConnection() | |
| 40 # create a new, empty bucket | |
| 41 bucket_name = 'test-%d' % int(time.time()) | |
| 42 bucket = c.create_bucket(bucket_name) | |
| 43 # now try a get_bucket call and see if it's really there | |
| 44 bucket = c.get_bucket(bucket_name) | |
| 45 k = bucket.new_key() | |
| 46 k.name = 'foobar' | |
| 47 s1 = 'This is a test of file upload and download' | |
| 48 s2 = 'This is a second string to test file upload and download' | |
| 49 k.set_contents_from_string(s1) | |
| 50 fp = open('foobar', 'wb') | |
| 51 # now get the contents from s3 to a local file | |
| 52 k.get_contents_to_file(fp) | |
| 53 fp.close() | |
| 54 fp = open('foobar') | |
| 55 # check to make sure content read from s3 is identical to original | |
| 56 assert s1 == fp.read(), 'corrupted file' | |
| 57 fp.close() | |
| 58 bucket.delete_key(k) | |
| 59 # test a few variations on get_all_keys - first load some data | |
| 60 # for the first one, let's override the content type | |
| 61 phony_mimetype = 'application/x-boto-test' | |
| 62 headers = {'Content-Type': phony_mimetype} | |
| 63 k.name = 'foo/bar' | |
| 64 k.set_contents_from_string(s1, headers) | |
| 65 k.name = 'foo/bas' | |
| 66 k.set_contents_from_filename('foobar') | |
| 67 k.name = 'foo/bat' | |
| 68 k.set_contents_from_string(s1) | |
| 69 k.name = 'fie/bar' | |
| 70 k.set_contents_from_string(s1) | |
| 71 k.name = 'fie/bas' | |
| 72 k.set_contents_from_string(s1) | |
| 73 k.name = 'fie/bat' | |
| 74 k.set_contents_from_string(s1) | |
| 75 # try resetting the contents to another value | |
| 76 md5 = k.md5 | |
| 77 k.set_contents_from_string(s2) | |
| 78 assert k.md5 != md5 | |
| 79 os.unlink('foobar') | |
| 80 all = bucket.get_all_keys() | |
| 81 assert len(all) == 6 | |
| 82 rs = bucket.get_all_keys(prefix='foo') | |
| 83 assert len(rs) == 3 | |
| 84 rs = bucket.get_all_keys(prefix='', delimiter='/') | |
| 85 assert len(rs) == 2 | |
| 86 rs = bucket.get_all_keys(maxkeys=5) | |
| 87 assert len(rs) == 5 | |
| 88 # test the lookup method | |
| 89 k = bucket.lookup('foo/bar') | |
| 90 assert isinstance(k, bucket.key_class) | |
| 91 assert k.content_type == phony_mimetype | |
| 92 k = bucket.lookup('notthere') | |
| 93 assert k == None | |
| 94 # try some metadata stuff | |
| 95 k = bucket.new_key() | |
| 96 k.name = 'has_metadata' | |
| 97 mdkey1 = 'meta1' | |
| 98 mdval1 = 'This is the first metadata value' | |
| 99 k.set_metadata(mdkey1, mdval1) | |
| 100 mdkey2 = 'meta2' | |
| 101 mdval2 = 'This is the second metadata value' | |
| 102 k.set_metadata(mdkey2, mdval2) | |
| 103 # try a unicode metadata value | |
| 104 | |
| 105 mdval3 = u'föö' | |
| 106 mdkey3 = 'meta3' | |
| 107 k.set_metadata(mdkey3, mdval3) | |
| 108 k.set_contents_from_string(s1) | |
| 109 | |
| 110 k = bucket.lookup('has_metadata') | |
| 111 assert k.get_metadata(mdkey1) == mdval1 | |
| 112 assert k.get_metadata(mdkey2) == mdval2 | |
| 113 assert k.get_metadata(mdkey3) == mdval3 | |
| 114 k = bucket.new_key() | |
| 115 k.name = 'has_metadata' | |
| 116 k.get_contents_as_string() | |
| 117 assert k.get_metadata(mdkey1) == mdval1 | |
| 118 assert k.get_metadata(mdkey2) == mdval2 | |
| 119 assert k.get_metadata(mdkey3) == mdval3 | |
| 120 bucket.delete_key(k) | |
| 121 # test list and iterator | |
| 122 rs1 = bucket.list() | |
| 123 num_iter = 0 | |
| 124 for r in rs1: | |
| 125 num_iter = num_iter + 1 | |
| 126 rs = bucket.get_all_keys() | |
| 127 num_keys = len(rs) | |
| 128 assert num_iter == num_keys | |
| 129 # try some acl stuff | |
| 130 bucket.set_acl('public-read') | |
| 131 acl = bucket.get_acl() | |
| 132 assert len(acl.entries.entry_list) == 2 | |
| 133 bucket.set_acl('private') | |
| 134 acl = bucket.get_acl() | |
| 135 assert len(acl.entries.entry_list) == 1 | |
| 136 k = bucket.lookup('foo/bar') | |
| 137 k.set_acl('public-read') | |
| 138 acl = k.get_acl() | |
| 139 assert len(acl.entries.entry_list) == 2 | |
| 140 k.set_acl('private') | |
| 141 acl = k.get_acl() | |
| 142 assert len(acl.entries.entry_list) == 1 | |
| 143 # now delete all keys in bucket | |
| 144 for k in bucket: | |
| 145 bucket.delete_key(k) | |
| 146 # now delete bucket | |
| 147 time.sleep(5) | |
| 148 c.delete_bucket(bucket) | |
| 149 print '--- tests completed ---' | |
| OLD | NEW |
|---|
Chromium Code Reviews
