OLD | NEW |
| (Empty) |
1 #!/usr/bin/env python | |
2 | |
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 SDBConnection | |
28 """ | |
29 | |
30 import unittest | |
31 import time | |
32 from boto.sdb.connection import SDBConnection | |
33 from boto.exception import SDBResponseError | |
34 | |
35 class SDBConnectionTest (unittest.TestCase): | |
36 | |
37 def test_1_basic(self): | |
38 print '--- running SDBConnection tests ---' | |
39 c = SDBConnection() | |
40 rs = c.get_all_domains() | |
41 num_domains = len(rs) | |
42 | |
43 # try illegal name | |
44 try: | |
45 domain = c.create_domain('bad:domain:name') | |
46 except SDBResponseError: | |
47 pass | |
48 | |
49 # now create one that should work and should be unique (i.e. a new one) | |
50 domain_name = 'test%d' % int(time.time()) | |
51 domain = c.create_domain(domain_name) | |
52 rs = c.get_all_domains() | |
53 assert len(rs) == num_domains + 1 | |
54 | |
55 # now let's a couple of items and attributes | |
56 item_1 = 'item1' | |
57 same_value = 'same_value' | |
58 attrs_1 = {'name1' : same_value, 'name2' : 'diff_value_1'} | |
59 domain.put_attributes(item_1, attrs_1) | |
60 item_2 = 'item2' | |
61 attrs_2 = {'name1' : same_value, 'name2' : 'diff_value_2'} | |
62 domain.put_attributes(item_2, attrs_2) | |
63 | |
64 # try to get the attributes and see if they match | |
65 item = domain.get_attributes(item_1, consistent_read=True) | |
66 assert len(item.keys()) == len(attrs_1.keys()) | |
67 assert item['name1'] == attrs_1['name1'] | |
68 assert item['name2'] == attrs_1['name2'] | |
69 | |
70 # try a search or two | |
71 query = 'select * from %s where name1="%s"' % (domain_name, same_value) | |
72 rs = domain.select(query, consistent_read=True) | |
73 n = 0 | |
74 for item in rs: | |
75 n += 1 | |
76 assert n == 2 | |
77 query = 'select * from %s where name2="diff_value_2"' % domain_name | |
78 rs = domain.select(query, consistent_read=True) | |
79 n = 0 | |
80 for item in rs: | |
81 n += 1 | |
82 assert n == 1 | |
83 | |
84 # delete all attributes associated with item_1 | |
85 stat = domain.delete_attributes(item_1) | |
86 assert stat | |
87 | |
88 # now try a batch put operation on the domain | |
89 item3 = {'name3_1' : 'value3_1', | |
90 'name3_2' : 'value3_2', | |
91 'name3_3' : ['value3_3_1', 'value3_3_2']} | |
92 | |
93 item4 = {'name4_1' : 'value4_1', | |
94 'name4_2' : ['value4_2_1', 'value4_2_2'], | |
95 'name4_3' : 'value4_3'} | |
96 items = {'item3' : item3, 'item4' : item4} | |
97 domain.batch_put_attributes(items) | |
98 | |
99 item = domain.get_attributes('item3', consistent_read=True) | |
100 assert item['name3_2'] == 'value3_2' | |
101 | |
102 # now try a batch delete operation (variation #1) | |
103 items = {'item3' : item3} | |
104 stat = domain.batch_delete_attributes(items) | |
105 | |
106 item = domain.get_attributes('item3', consistent_read=True) | |
107 assert not item | |
108 | |
109 # now try a batch delete operation (variation #2) | |
110 stat = domain.batch_delete_attributes({'item4' : None}) | |
111 | |
112 item = domain.get_attributes('item4', consistent_read=True) | |
113 assert not item | |
114 | |
115 # now delete the domain | |
116 stat = c.delete_domain(domain) | |
117 assert stat | |
118 | |
119 print '--- tests completed ---' | |
120 | |
OLD | NEW |