| 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 |
|---|
Chromium Code Reviews
