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 |