Lib/bsddb/test/test_dbtables.py

   1 #!/usr/bin/env python
   2 #
   3 #-----------------------------------------------------------------------
   4 # A test suite for the table interface built on bsddb.db
   5 #-----------------------------------------------------------------------
   6 #
   7 # Copyright (C) 2000, 2001 by Autonomous Zone Industries
   8 # Copyright (C) 2002 Gregory P. Smith
   9 #
  10 # March 20, 2000
  11 #
  12 # License:      This is free software.  You may use this software for any
  13 #               purpose including modification/redistribution, so long as
  14 #               this header remains intact and that you do not claim any
  15 #               rights of ownership or authorship of this software.  This
  16 #               software has been tested, but no warranty is expressed or
  17 #               implied.
  18 #
  19 #   --  Gregory P. Smith <greg@krypto.org>
  20 #
  21 # $Id$
  22
  23 import os, re
  24 try:
  25     import cPickle
  26     pickle = cPickle
  27 except ImportError:
  28     import pickle
  29
  30 import unittest
  31 from test_all import db, dbtables, test_support, verbose, \
  32         get_new_environment_path, get_new_database_path
  33
  34 #----------------------------------------------------------------------
  35
  36 class TableDBTestCase(unittest.TestCase):
  37     db_name = 'test-table.db'
  38
  39     def setUp(self):
  40         import sys
  41         if sys.version_info[0] >= 3 :
  42             from test_all import do_proxy_db_py3k
  43             self._flag_proxy_db_py3k = do_proxy_db_py3k(False)
  44
  45         self.testHomeDir = get_new_environment_path()
  46         self.tdb = dbtables.bsdTableDB(
  47             filename='tabletest.db', dbhome=self.testHomeDir, create=1)
  48
  49     def tearDown(self):
  50         self.tdb.close()
  51         import sys
  52         if sys.version_info[0] >= 3 :
  53             from test_all import do_proxy_db_py3k
  54             do_proxy_db_py3k(self._flag_proxy_db_py3k)
  55         test_support.rmtree(self.testHomeDir)
  56
  57     def test01(self):
  58         tabname = "test01"
  59         colname = 'cool numbers'
  60         try:
  61             self.tdb.Drop(tabname)
  62         except dbtables.TableDBError:
  63             pass
  64         self.tdb.CreateTable(tabname, [colname])
  65         import sys
  66         if sys.version_info[0] < 3 :
  67             self.tdb.Insert(tabname, {colname: pickle.dumps(3.14159, 1)})
  68         else :
  69             self.tdb.Insert(tabname, {colname: pickle.dumps(3.14159,
  70                 1).decode("iso8859-1")})  # 8 bits
  71
  72         if verbose:
  73             self.tdb._db_print()
  74
  75         values = self.tdb.Select(
  76             tabname, [colname], conditions={colname: None})
  77
  78         import sys
  79         if sys.version_info[0] < 3 :
  80             colval = pickle.loads(values[0][colname])
  81         else :
  82             colval = pickle.loads(bytes(values[0][colname], "iso8859-1"))
  83         self.assert_(colval > 3.141)
  84         self.assert_(colval < 3.142)
  85
  86
  87     def test02(self):
  88         tabname = "test02"
  89         col0 = 'coolness factor'
  90         col1 = 'but can it fly?'
  91         col2 = 'Species'
  92
  93         import sys
  94         if sys.version_info[0] < 3 :
  95             testinfo = [
  96                 {col0: pickle.dumps(8, 1), col1: 'no', col2: 'Penguin'},
  97                 {col0: pickle.dumps(-1, 1), col1: 'no', col2: 'Turkey'},
  98                 {col0: pickle.dumps(9, 1), col1: 'yes', col2: 'SR-71A Blackbird'}
  99             ]
 100         else :
 101             testinfo = [
 102                 {col0: pickle.dumps(8, 1).decode("iso8859-1"),
 103                     col1: 'no', col2: 'Penguin'},
 104                 {col0: pickle.dumps(-1, 1).decode("iso8859-1"),
 105                     col1: 'no', col2: 'Turkey'},
 106                 {col0: pickle.dumps(9, 1).decode("iso8859-1"),
 107                     col1: 'yes', col2: 'SR-71A Blackbird'}
 108             ]
 109
 110         try:
 111             self.tdb.Drop(tabname)
 112         except dbtables.TableDBError:
 113             pass
 114         self.tdb.CreateTable(tabname, [col0, col1, col2])
 115         for row in testinfo :
 116             self.tdb.Insert(tabname, row)
 117
 118         import sys
 119         if sys.version_info[0] < 3 :
 120             values = self.tdb.Select(tabname, [col2],
 121                 conditions={col0: lambda x: pickle.loads(x) >= 8})
 122         else :
 123             values = self.tdb.Select(tabname, [col2],
 124                 conditions={col0: lambda x:
 125                     pickle.loads(bytes(x, "iso8859-1")) >= 8})
 126
 127         self.assertEqual(len(values), 2)
 128         if values[0]['Species'] == 'Penguin' :
 129             self.assertEqual(values[1]['Species'], 'SR-71A Blackbird')
 130         elif values[0]['Species'] == 'SR-71A Blackbird' :
 131             self.assertEqual(values[1]['Species'], 'Penguin')
 132         else :
 133             if verbose:
 134                 print "values= %r" % (values,)
 135             raise RuntimeError("Wrong values returned!")
 136
 137     def test03(self):
 138         tabname = "test03"
 139         try:
 140             self.tdb.Drop(tabname)
 141         except dbtables.TableDBError:
 142             pass
 143         if verbose:
 144             print '...before CreateTable...'
 145             self.tdb._db_print()
 146         self.tdb.CreateTable(tabname, ['a', 'b', 'c', 'd', 'e'])
 147         if verbose:
 148             print '...after CreateTable...'
 149             self.tdb._db_print()
 150         self.tdb.Drop(tabname)
 151         if verbose:
 152             print '...after Drop...'
 153             self.tdb._db_print()
 154         self.tdb.CreateTable(tabname, ['a', 'b', 'c', 'd', 'e'])
 155
 156         try:
 157             self.tdb.Insert(tabname,
 158                             {'a': "",
 159                              'e': pickle.dumps([{4:5, 6:7}, 'foo'], 1),
 160                              'f': "Zero"})
 161             self.fail('Expected an exception')
 162         except dbtables.TableDBError:
 163             pass
 164
 165         try:
 166             self.tdb.Select(tabname, [], conditions={'foo': '123'})
 167             self.fail('Expected an exception')
 168         except dbtables.TableDBError:
 169             pass
 170
 171         self.tdb.Insert(tabname,
 172                         {'a': '42',
 173                          'b': "bad",
 174                          'c': "meep",
 175                          'e': 'Fuzzy wuzzy was a bear'})
 176         self.tdb.Insert(tabname,
 177                         {'a': '581750',
 178                          'b': "good",
 179                          'd': "bla",
 180                          'c': "black",
 181                          'e': 'fuzzy was here'})
 182         self.tdb.Insert(tabname,
 183                         {'a': '800000',
 184                          'b': "good",
 185                          'd': "bla",
 186                          'c': "black",
 187                          'e': 'Fuzzy wuzzy is a bear'})
 188
 189         if verbose:
 190             self.tdb._db_print()
 191
 192         # this should return two rows
 193         values = self.tdb.Select(tabname, ['b', 'a', 'd'],
 194             conditions={'e': re.compile('wuzzy').search,
 195                         'a': re.compile('^[0-9]+$').match})
 196         self.assertEqual(len(values), 2)
 197
 198         # now lets delete one of them and try again
 199         self.tdb.Delete(tabname, conditions={'b': dbtables.ExactCond('good')})
 200         values = self.tdb.Select(
 201             tabname, ['a', 'd', 'b'],
 202             conditions={'e': dbtables.PrefixCond('Fuzzy')})
 203         self.assertEqual(len(values), 1)
 204         self.assertEqual(values[0]['d'], None)
 205
 206         values = self.tdb.Select(tabname, ['b'],
 207             conditions={'c': lambda c: c == 'meep'})
 208         self.assertEqual(len(values), 1)
 209         self.assertEqual(values[0]['b'], "bad")
 210
 211
 212     def test04_MultiCondSelect(self):
 213         tabname = "test04_MultiCondSelect"
 214         try:
 215             self.tdb.Drop(tabname)
 216         except dbtables.TableDBError:
 217             pass
 218         self.tdb.CreateTable(tabname, ['a', 'b', 'c', 'd', 'e'])
 219
 220         try:
 221             self.tdb.Insert(tabname,
 222                             {'a': "",
 223                              'e': pickle.dumps([{4:5, 6:7}, 'foo'], 1),
 224                              'f': "Zero"})
 225             self.fail('Expected an exception')
 226         except dbtables.TableDBError:
 227             pass
 228
 229         self.tdb.Insert(tabname, {'a': "A", 'b': "B", 'c': "C", 'd': "D",
 230                                   'e': "E"})
 231         self.tdb.Insert(tabname, {'a': "-A", 'b': "-B", 'c': "-C", 'd': "-D",
 232                                   'e': "-E"})
 233         self.tdb.Insert(tabname, {'a': "A-", 'b': "B-", 'c': "C-", 'd': "D-",
 234                                   'e': "E-"})
 235
 236         if verbose:
 237             self.tdb._db_print()
 238
 239         # This select should return 0 rows.  it is designed to test
 240         # the bug identified and fixed in sourceforge bug # 590449
 241         # (Big Thanks to "Rob Tillotson (n9mtb)" for tracking this down
 242         # and supplying a fix!!  This one caused many headaches to say
 243         # the least...)
 244         values = self.tdb.Select(tabname, ['b', 'a', 'd'],
 245             conditions={'e': dbtables.ExactCond('E'),
 246                         'a': dbtables.ExactCond('A'),
 247                         'd': dbtables.PrefixCond('-')
 248                        } )
 249         self.assertEqual(len(values), 0, values)
 250
 251
 252     def test_CreateOrExtend(self):
 253         tabname = "test_CreateOrExtend"
 254
 255         self.tdb.CreateOrExtendTable(
 256             tabname, ['name', 'taste', 'filling', 'alcohol content', 'price'])
 257         try:
 258             self.tdb.Insert(tabname,
 259                             {'taste': 'crap',
 260                              'filling': 'no',
 261                              'is it Guinness?': 'no'})
 262             self.fail("Insert should've failed due to bad column name")
 263         except:
 264             pass
 265         self.tdb.CreateOrExtendTable(tabname,
 266                                      ['name', 'taste', 'is it Guinness?'])
 267
 268         # these should both succeed as the table should contain the union of both sets of columns.
 269         self.tdb.Insert(tabname, {'taste': 'crap', 'filling': 'no',
 270                                   'is it Guinness?': 'no'})
 271         self.tdb.Insert(tabname, {'taste': 'great', 'filling': 'yes',
 272                                   'is it Guinness?': 'yes',
 273                                   'name': 'Guinness'})
 274
 275
 276     def test_CondObjs(self):
 277         tabname = "test_CondObjs"
 278
 279         self.tdb.CreateTable(tabname, ['a', 'b', 'c', 'd', 'e', 'p'])
 280
 281         self.tdb.Insert(tabname, {'a': "the letter A",
 282                                   'b': "the letter B",
 283                                   'c': "is for cookie"})
 284         self.tdb.Insert(tabname, {'a': "is for aardvark",
 285                                   'e': "the letter E",
 286                                   'c': "is for cookie",
 287                                   'd': "is for dog"})
 288         self.tdb.Insert(tabname, {'a': "the letter A",
 289                                   'e': "the letter E",
 290                                   'c': "is for cookie",
 291                                   'p': "is for Python"})
 292
 293         values = self.tdb.Select(
 294             tabname, ['p', 'e'],
 295             conditions={'e': dbtables.PrefixCond('the l')})
 296         self.assertEqual(len(values), 2, values)
 297         self.assertEqual(values[0]['e'], values[1]['e'], values)
 298         self.assertNotEqual(values[0]['p'], values[1]['p'], values)
 299
 300         values = self.tdb.Select(
 301             tabname, ['d', 'a'],
 302             conditions={'a': dbtables.LikeCond('%aardvark%')})
 303         self.assertEqual(len(values), 1, values)
 304         self.assertEqual(values[0]['d'], "is for dog", values)
 305         self.assertEqual(values[0]['a'], "is for aardvark", values)
 306
 307         values = self.tdb.Select(tabname, None,
 308                                  {'b': dbtables.Cond(),
 309                                   'e':dbtables.LikeCond('%letter%'),
 310                                   'a':dbtables.PrefixCond('is'),
 311                                   'd':dbtables.ExactCond('is for dog'),
 312                                   'c':dbtables.PrefixCond('is for'),
 313                                   'p':lambda s: not s})
 314         self.assertEqual(len(values), 1, values)
 315         self.assertEqual(values[0]['d'], "is for dog", values)
 316         self.assertEqual(values[0]['a'], "is for aardvark", values)
 317
 318     def test_Delete(self):
 319         tabname = "test_Delete"
 320         self.tdb.CreateTable(tabname, ['x', 'y', 'z'])
 321
 322         # prior to 2001-05-09 there was a bug where Delete() would
 323         # fail if it encountered any rows that did not have values in
 324         # every column.
 325         # Hunted and Squashed by <Donwulff> (Jukka Santala - donwulff@nic.fi)
 326         self.tdb.Insert(tabname, {'x': 'X1', 'y':'Y1'})
 327         self.tdb.Insert(tabname, {'x': 'X2', 'y':'Y2', 'z': 'Z2'})
 328
 329         self.tdb.Delete(tabname, conditions={'x': dbtables.PrefixCond('X')})
 330         values = self.tdb.Select(tabname, ['y'],
 331                                  conditions={'x': dbtables.PrefixCond('X')})
 332         self.assertEqual(len(values), 0)
 333
 334     def test_Modify(self):
 335         tabname = "test_Modify"
 336         self.tdb.CreateTable(tabname, ['Name', 'Type', 'Access'])
 337
 338         self.tdb.Insert(tabname, {'Name': 'Index to MP3 files.doc',
 339                                   'Type': 'Word', 'Access': '8'})
 340         self.tdb.Insert(tabname, {'Name': 'Nifty.MP3', 'Access': '1'})
 341         self.tdb.Insert(tabname, {'Type': 'Unknown', 'Access': '0'})
 342
 343         def set_type(type):
 344             if type == None:
 345                 return 'MP3'
 346             return type
 347
 348         def increment_access(count):
 349             return str(int(count)+1)
 350
 351         def remove_value(value):
 352             return None
 353
 354         self.tdb.Modify(tabname,
 355                         conditions={'Access': dbtables.ExactCond('0')},
 356                         mappings={'Access': remove_value})
 357         self.tdb.Modify(tabname,
 358                         conditions={'Name': dbtables.LikeCond('%MP3%')},
 359                         mappings={'Type': set_type})
 360         self.tdb.Modify(tabname,
 361                         conditions={'Name': dbtables.LikeCond('%')},
 362                         mappings={'Access': increment_access})
 363
 364         try:
 365             self.tdb.Modify(tabname,
 366                             conditions={'Name': dbtables.LikeCond('%')},
 367                             mappings={'Access': 'What is your quest?'})
 368         except TypeError:
 369             # success, the string value in mappings isn't callable
 370             pass
 371         else:
 372             raise RuntimeError, "why was TypeError not raised for bad callable?"
 373
 374         # Delete key in select conditions
 375         values = self.tdb.Select(
 376             tabname, None,
 377             conditions={'Type': dbtables.ExactCond('Unknown')})
 378         self.assertEqual(len(values), 1, values)
 379         self.assertEqual(values[0]['Name'], None, values)
 380         self.assertEqual(values[0]['Access'], None, values)
 381
 382         # Modify value by select conditions
 383         values = self.tdb.Select(
 384             tabname, None,
 385             conditions={'Name': dbtables.ExactCond('Nifty.MP3')})
 386         self.assertEqual(len(values), 1, values)
 387         self.assertEqual(values[0]['Type'], "MP3", values)
 388         self.assertEqual(values[0]['Access'], "2", values)
 389
 390         # Make sure change applied only to select conditions
 391         values = self.tdb.Select(
 392             tabname, None, conditions={'Name': dbtables.LikeCond('%doc%')})
 393         self.assertEqual(len(values), 1, values)
 394         self.assertEqual(values[0]['Type'], "Word", values)
 395         self.assertEqual(values[0]['Access'], "9", values)
 396
 397
 398 def test_suite():
 399     suite = unittest.TestSuite()
 400     suite.addTest(unittest.makeSuite(TableDBTestCase))
 401     return suite
 402
 403
 404 if __name__ == '__main__':
 405     unittest.main(defaultTest='test_suite')