1 """

     2 TestCases for checking dbShelve objects.

     3 """

     4 

     5 import os, string

     6 import random

     7 import unittest

     8 

     9 

    10 from test_all import db, dbshelve, test_support, verbose, \

    11         get_new_environment_path, get_new_database_path

    12 

    13 

    14 

    15 #----------------------------------------------------------------------

    16 

    17 # We want the objects to be comparable so we can test dbshelve.values

    18 # later on.

    19 class DataClass:

    20     def __init__(self):

    21         self.value = random.random()

    22 

    23     def __repr__(self) :  # For Python 3.0 comparison

    24         return "DataClass %f" %self.value

    25 

    26     def __cmp__(self, other):  # For Python 2.x comparison

    27         return cmp(self.value, other)

    28 

    29 

    30 class DBShelveTestCase(unittest.TestCase):

    31     def setUp(self):

    32         import sys

    33         if sys.version_info[0] >= 3 :

    34             from test_all import do_proxy_db_py3k

    35             self._flag_proxy_db_py3k = do_proxy_db_py3k(False)

    36         self.filename = get_new_database_path()

    37         self.do_open()

    38 

    39     def tearDown(self):

    40         import sys

    41         if sys.version_info[0] >= 3 :

    42             from test_all import do_proxy_db_py3k

    43             do_proxy_db_py3k(self._flag_proxy_db_py3k)

    44         self.do_close()

    45         test_support.unlink(self.filename)

    46 

    47     def mk(self, key):

    48         """Turn key into an appropriate key type for this db"""

    49         # override in child class for RECNO

    50         import sys

    51         if sys.version_info[0] < 3 :

    52             return key

    53         else :

    54             return bytes(key, "iso8859-1")  # 8 bits

    55 

    56     def populateDB(self, d):

    57         for x in string.letters:

    58             d[self.mk('S' + x)] = 10 * x           # add a string

    59             d[self.mk('I' + x)] = ord(x)           # add an integer

    60             d[self.mk('L' + x)] = [x] * 10         # add a list

    61 

    62             inst = DataClass()            # add an instance

    63             inst.S = 10 * x

    64             inst.I = ord(x)

    65             inst.L = [x] * 10

    66             d[self.mk('O' + x)] = inst

    67 

    68 

    69     # overridable in derived classes to affect how the shelf is created/opened

    70     def do_open(self):

    71         self.d = dbshelve.open(self.filename)

    72 

    73     # and closed...

    74     def do_close(self):

    75         self.d.close()

    76 

    77 

    78 

    79     def test01_basics(self):

    80         if verbose:

    81             print '\n', '-=' * 30

    82             print "Running %s.test01_basics..." % self.__class__.__name__

    83 

    84         self.populateDB(self.d)

    85         self.d.sync()

    86         self.do_close()

    87         self.do_open()

    88         d = self.d

    89 

    90         l = len(d)

    91         k = d.keys()

    92         s = d.stat()

    93         f = d.fd()

    94 

    95         if verbose:

    96             print "length:", l

    97             print "keys:", k

    98             print "stats:", s

    99 

   100         self.assertEqual(0, d.has_key(self.mk('bad key')))

   101         self.assertEqual(1, d.has_key(self.mk('IA')))

   102         self.assertEqual(1, d.has_key(self.mk('OA')))

   103 

   104         d.delete(self.mk('IA'))

   105         del d[self.mk('OA')]

   106         self.assertEqual(0, d.has_key(self.mk('IA')))

   107         self.assertEqual(0, d.has_key(self.mk('OA')))

   108         self.assertEqual(len(d), l-2)

   109 

   110         values = []

   111         for key in d.keys():

   112             value = d[key]

   113             values.append(value)

   114             if verbose:

   115                 print "%s: %s" % (key, value)

   116             self.checkrec(key, value)

   117 

   118         dbvalues = d.values()

   119         self.assertEqual(len(dbvalues), len(d.keys()))

   120         import sys

   121         if sys.version_info[0] < 3 :

   122             values.sort()

   123             dbvalues.sort()

   124             self.assertEqual(values, dbvalues)

   125         else :  # XXX: Convert all to strings. Please, improve

   126             values.sort(key=lambda x : str(x))

   127             dbvalues.sort(key=lambda x : str(x))

   128             self.assertEqual(repr(values), repr(dbvalues))

   129 

   130         items = d.items()

   131         self.assertEqual(len(items), len(values))

   132 

   133         for key, value in items:

   134             self.checkrec(key, value)

   135 

   136         self.assertEqual(d.get(self.mk('bad key')), None)

   137         self.assertEqual(d.get(self.mk('bad key'), None), None)

   138         self.assertEqual(d.get(self.mk('bad key'), 'a string'), 'a string')

   139         self.assertEqual(d.get(self.mk('bad key'), [1, 2, 3]), [1, 2, 3])

   140 

   141         d.set_get_returns_none(0)

   142         self.assertRaises(db.DBNotFoundError, d.get, self.mk('bad key'))

   143         d.set_get_returns_none(1)

   144 

   145         d.put(self.mk('new key'), 'new data')

   146         self.assertEqual(d.get(self.mk('new key')), 'new data')

   147         self.assertEqual(d[self.mk('new key')], 'new data')

   148 

   149 

   150 

   151     def test02_cursors(self):

   152         if verbose:

   153             print '\n', '-=' * 30

   154             print "Running %s.test02_cursors..." % self.__class__.__name__

   155 

   156         self.populateDB(self.d)

   157         d = self.d

   158 

   159         count = 0

   160         c = d.cursor()

   161         rec = c.first()

   162         while rec is not None:

   163             count = count + 1

   164             if verbose:

   165                 print rec

   166             key, value = rec

   167             self.checkrec(key, value)

   168             # Hack to avoid conversion by 2to3 tool

   169             rec = getattr(c, "next")()

   170         del c

   171 

   172         self.assertEqual(count, len(d))

   173 

   174         count = 0

   175         c = d.cursor()

   176         rec = c.last()

   177         while rec is not None:

   178             count = count + 1

   179             if verbose:

   180                 print rec

   181             key, value = rec

   182             self.checkrec(key, value)

   183             rec = c.prev()

   184 

   185         self.assertEqual(count, len(d))

   186 

   187         c.set(self.mk('SS'))

   188         key, value = c.current()

   189         self.checkrec(key, value)

   190         del c

   191 

   192 

   193     def test03_append(self):

   194         # NOTE: this is overridden in RECNO subclass, don't change its name.

   195         if verbose:

   196             print '\n', '-=' * 30

   197             print "Running %s.test03_append..." % self.__class__.__name__

   198 

   199         self.assertRaises(dbshelve.DBShelveError,

   200                           self.d.append, 'unit test was here')

   201 

   202 

   203     def checkrec(self, key, value):

   204         # override this in a subclass if the key type is different

   205 

   206         import sys

   207         if sys.version_info[0] >= 3 :

   208             if isinstance(key, bytes) :

   209                 key = key.decode("iso8859-1")  # 8 bits

   210 

   211         x = key[1]

   212         if key[0] == 'S':

   213             self.assertEqual(type(value), str)

   214             self.assertEqual(value, 10 * x)

   215 

   216         elif key[0] == 'I':

   217             self.assertEqual(type(value), int)

   218             self.assertEqual(value, ord(x))

   219 

   220         elif key[0] == 'L':

   221             self.assertEqual(type(value), list)

   222             self.assertEqual(value, [x] * 10)

   223 

   224         elif key[0] == 'O':

   225             import sys

   226             if sys.version_info[0] < 3 :

   227                 from types import InstanceType

   228                 self.assertEqual(type(value), InstanceType)

   229             else :

   230                 self.assertEqual(type(value), DataClass)

   231 

   232             self.assertEqual(value.S, 10 * x)

   233             self.assertEqual(value.I, ord(x))

   234             self.assertEqual(value.L, [x] * 10)

   235 

   236         else:

   237             self.assert_(0, 'Unknown key type, fix the test')

   238 

   239 #----------------------------------------------------------------------

   240 

   241 class BasicShelveTestCase(DBShelveTestCase):

   242     def do_open(self):

   243         self.d = dbshelve.DBShelf()

   244         self.d.open(self.filename, self.dbtype, self.dbflags)

   245 

   246     def do_close(self):

   247         self.d.close()

   248 

   249 

   250 class BTreeShelveTestCase(BasicShelveTestCase):

   251     dbtype = db.DB_BTREE

   252     dbflags = db.DB_CREATE

   253 

   254 

   255 class HashShelveTestCase(BasicShelveTestCase):

   256     dbtype = db.DB_HASH

   257     dbflags = db.DB_CREATE

   258 

   259 

   260 class ThreadBTreeShelveTestCase(BasicShelveTestCase):

   261     dbtype = db.DB_BTREE

   262     dbflags = db.DB_CREATE | db.DB_THREAD

   263 

   264 

   265 class ThreadHashShelveTestCase(BasicShelveTestCase):

   266     dbtype = db.DB_HASH

   267     dbflags = db.DB_CREATE | db.DB_THREAD

   268 

   269 

   270 #----------------------------------------------------------------------

   271 

   272 class BasicEnvShelveTestCase(DBShelveTestCase):

   273     def do_open(self):

   274         self.env = db.DBEnv()

   275         self.env.open(self.homeDir,

   276                 self.envflags | db.DB_INIT_MPOOL | db.DB_CREATE)

   277 

   278         self.filename = os.path.split(self.filename)[1]

   279         self.d = dbshelve.DBShelf(self.env)

   280         self.d.open(self.filename, self.dbtype, self.dbflags)

   281 

   282 

   283     def do_close(self):

   284         self.d.close()

   285         self.env.close()

   286 

   287 

   288     def setUp(self) :

   289         self.homeDir = get_new_environment_path()

   290         DBShelveTestCase.setUp(self)

   291 

   292     def tearDown(self):

   293         import sys

   294         if sys.version_info[0] >= 3 :

   295             from test_all import do_proxy_db_py3k

   296             do_proxy_db_py3k(self._flag_proxy_db_py3k)

   297         self.do_close()

   298         test_support.rmtree(self.homeDir)

   299 

   300 

   301 class EnvBTreeShelveTestCase(BasicEnvShelveTestCase):

   302     envflags = 0

   303     dbtype = db.DB_BTREE

   304     dbflags = db.DB_CREATE

   305 

   306 

   307 class EnvHashShelveTestCase(BasicEnvShelveTestCase):

   308     envflags = 0

   309     dbtype = db.DB_HASH

   310     dbflags = db.DB_CREATE

   311 

   312 

   313 class EnvThreadBTreeShelveTestCase(BasicEnvShelveTestCase):

   314     envflags = db.DB_THREAD

   315     dbtype = db.DB_BTREE

   316     dbflags = db.DB_CREATE | db.DB_THREAD

   317 

   318 

   319 class EnvThreadHashShelveTestCase(BasicEnvShelveTestCase):

   320     envflags = db.DB_THREAD

   321     dbtype = db.DB_HASH

   322     dbflags = db.DB_CREATE | db.DB_THREAD

   323 

   324 

   325 #----------------------------------------------------------------------

   326 # test cases for a DBShelf in a RECNO DB.

   327 

   328 class RecNoShelveTestCase(BasicShelveTestCase):

   329     dbtype = db.DB_RECNO

   330     dbflags = db.DB_CREATE

   331 

   332     def setUp(self):

   333         BasicShelveTestCase.setUp(self)

   334 

   335         # pool to assign integer key values out of

   336         self.key_pool = list(range(1, 5000))

   337         self.key_map = {}     # map string keys to the number we gave them

   338         self.intkey_map = {}  # reverse map of above

   339 

   340     def mk(self, key):

   341         if key not in self.key_map:

   342             self.key_map[key] = self.key_pool.pop(0)

   343             self.intkey_map[self.key_map[key]] = key

   344         return self.key_map[key]

   345 

   346     def checkrec(self, intkey, value):

   347         key = self.intkey_map[intkey]

   348         BasicShelveTestCase.checkrec(self, key, value)

   349 

   350     def test03_append(self):

   351         if verbose:

   352             print '\n', '-=' * 30

   353             print "Running %s.test03_append..." % self.__class__.__name__

   354 

   355         self.d[1] = 'spam'

   356         self.d[5] = 'eggs'

   357         self.assertEqual(6, self.d.append('spam'))

   358         self.assertEqual(7, self.d.append('baked beans'))

   359         self.assertEqual('spam', self.d.get(6))

   360         self.assertEqual('spam', self.d.get(1))

   361         self.assertEqual('baked beans', self.d.get(7))

   362         self.assertEqual('eggs', self.d.get(5))

   363 

   364 

   365 #----------------------------------------------------------------------

   366 

   367 def test_suite():

   368     suite = unittest.TestSuite()

   369 

   370     suite.addTest(unittest.makeSuite(DBShelveTestCase))

   371     suite.addTest(unittest.makeSuite(BTreeShelveTestCase))

   372     suite.addTest(unittest.makeSuite(HashShelveTestCase))

   373     suite.addTest(unittest.makeSuite(ThreadBTreeShelveTestCase))

   374     suite.addTest(unittest.makeSuite(ThreadHashShelveTestCase))

   375     suite.addTest(unittest.makeSuite(EnvBTreeShelveTestCase))

   376     suite.addTest(unittest.makeSuite(EnvHashShelveTestCase))

   377     suite.addTest(unittest.makeSuite(EnvThreadBTreeShelveTestCase))

   378     suite.addTest(unittest.makeSuite(EnvThreadHashShelveTestCase))

   379     suite.addTest(unittest.makeSuite(RecNoShelveTestCase))

   380 

   381     return suite

   382 

   383 

   384 if __name__ == '__main__':

   385     unittest.main(defaultTest='test_suite')