| OLD | NEW |
| (Empty) |
| 1 # Copyright (c) 2006-2010 Mitch Garnaat http://garnaat.org/ | |
| 2 # Copyright (c) 2010, Eucalyptus Systems, Inc. | |
| 3 # All rights reserved. | |
| 4 # | |
| 5 # Permission is hereby granted, free of charge, to any person obtaining a | |
| 6 # copy of this software and associated documentation files (the | |
| 7 # "Software"), to deal in the Software without restriction, including | |
| 8 # without limitation the rights to use, copy, modify, merge, publish, dis- | |
| 9 # tribute, sublicense, and/or sell copies of the Software, and to permit | |
| 10 # persons to whom the Software is furnished to do so, subject to the fol- | |
| 11 # lowing conditions: | |
| 12 # | |
| 13 # The above copyright notice and this permission notice shall be included | |
| 14 # in all copies or substantial portions of the Software. | |
| 15 # | |
| 16 # THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS | |
| 17 # OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABIL- | |
| 18 # ITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT | |
| 19 # SHALL THE AUTHOR BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, | |
| 20 # WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, | |
| 21 # OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS | |
| 22 # IN THE SOFTWARE. | |
| 23 | |
| 24 """ | |
| 25 Some unit tests for the SDBConnection | |
| 26 """ | |
| 27 | |
| 28 import unittest | |
| 29 import time | |
| 30 from boto.sdb.connection import SDBConnection | |
| 31 from boto.exception import SDBResponseError | |
| 32 | |
| 33 class SDBConnectionTest (unittest.TestCase): | |
| 34 | |
| 35 def test_1_basic(self): | |
| 36 print '--- running SDBConnection tests ---' | |
| 37 c = SDBConnection() | |
| 38 rs = c.get_all_domains() | |
| 39 num_domains = len(rs) | |
| 40 | |
| 41 # try illegal name | |
| 42 try: | |
| 43 domain = c.create_domain('bad:domain:name') | |
| 44 except SDBResponseError: | |
| 45 pass | |
| 46 | |
| 47 # now create one that should work and should be unique (i.e. a new one) | |
| 48 domain_name = 'test%d' % int(time.time()) | |
| 49 domain = c.create_domain(domain_name) | |
| 50 rs = c.get_all_domains() | |
| 51 assert len(rs) == num_domains + 1 | |
| 52 | |
| 53 # now let's a couple of items and attributes | |
| 54 item_1 = 'item1' | |
| 55 same_value = 'same_value' | |
| 56 attrs_1 = {'name1' : same_value, 'name2' : 'diff_value_1'} | |
| 57 domain.put_attributes(item_1, attrs_1) | |
| 58 item_2 = 'item2' | |
| 59 attrs_2 = {'name1' : same_value, 'name2' : 'diff_value_2'} | |
| 60 domain.put_attributes(item_2, attrs_2) | |
| 61 | |
| 62 # try to get the attributes and see if they match | |
| 63 item = domain.get_attributes(item_1, consistent_read=True) | |
| 64 assert len(item.keys()) == len(attrs_1.keys()) | |
| 65 assert item['name1'] == attrs_1['name1'] | |
| 66 assert item['name2'] == attrs_1['name2'] | |
| 67 | |
| 68 # try a search or two | |
| 69 query = 'select * from %s where name1="%s"' % (domain_name, same_value) | |
| 70 rs = domain.select(query, consistent_read=True) | |
| 71 n = 0 | |
| 72 for item in rs: | |
| 73 n += 1 | |
| 74 assert n == 2 | |
| 75 query = 'select * from %s where name2="diff_value_2"' % domain_name | |
| 76 rs = domain.select(query, consistent_read=True) | |
| 77 n = 0 | |
| 78 for item in rs: | |
| 79 n += 1 | |
| 80 assert n == 1 | |
| 81 | |
| 82 # delete all attributes associated with item_1 | |
| 83 stat = domain.delete_attributes(item_1) | |
| 84 assert stat | |
| 85 | |
| 86 # now try a batch put operation on the domain | |
| 87 item3 = {'name3_1' : 'value3_1', | |
| 88 'name3_2' : 'value3_2', | |
| 89 'name3_3' : ['value3_3_1', 'value3_3_2']} | |
| 90 | |
| 91 item4 = {'name4_1' : 'value4_1', | |
| 92 'name4_2' : ['value4_2_1', 'value4_2_2'], | |
| 93 'name4_3' : 'value4_3'} | |
| 94 items = {'item3' : item3, 'item4' : item4} | |
| 95 domain.batch_put_attributes(items) | |
| 96 | |
| 97 item = domain.get_attributes('item3', consistent_read=True) | |
| 98 assert item['name3_2'] == 'value3_2' | |
| 99 | |
| 100 # now try a batch delete operation (variation #1) | |
| 101 items = {'item3' : item3} | |
| 102 stat = domain.batch_delete_attributes(items) | |
| 103 | |
| 104 item = domain.get_attributes('item3', consistent_read=True) | |
| 105 assert not item | |
| 106 | |
| 107 # now try a batch delete operation (variation #2) | |
| 108 stat = domain.batch_delete_attributes({'item4' : None}) | |
| 109 | |
| 110 item = domain.get_attributes('item4', consistent_read=True) | |
| 111 assert not item | |
| 112 | |
| 113 # now delete the domain | |
| 114 stat = c.delete_domain(domain) | |
| 115 assert stat | |
| 116 | |
| 117 print '--- tests completed ---' | |
| 118 | |
| OLD | NEW |