Lib/bsddb/test/test_recno.py

   1 """TestCases for exercising a Recno DB.
   2 """
   3
   4 import os
   5 import sys
   6 import errno
   7 import tempfile
   8 from pprint import pprint
   9 import unittest
  10
  11 from bsddb import db
  12 from test_all import verbose
  13
  14 letters = 'abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ'
  15
  16
  17 #----------------------------------------------------------------------
  18
  19 class SimpleRecnoTestCase(unittest.TestCase):
  20     def setUp(self):
  21         self.filename = tempfile.mktemp()
  22
  23     def tearDown(self):
  24         try:
  25             os.remove(self.filename)
  26         except OSError, e:
  27             if e.errno <> errno.EEXIST: raise
  28
  29     def test01_basic(self):
  30         d = db.DB()
  31         d.open(self.filename, db.DB_RECNO, db.DB_CREATE)
  32
  33         for x in letters:
  34             recno = d.append(x * 60)
  35             assert type(recno) == type(0)
  36             assert recno >= 1
  37             if verbose:
  38                 print recno,
  39
  40         if verbose: print
  41
  42         stat = d.stat()
  43         if verbose:
  44             pprint(stat)
  45
  46         for recno in range(1, len(d)+1):
  47             data = d[recno]
  48             if verbose:
  49                 print data
  50
  51             assert type(data) == type("")
  52             assert data == d.get(recno)
  53
  54         try:
  55             data = d[0]  # This should raise a KeyError!?!?!
  56         except db.DBInvalidArgError, val:
  57             assert val[0] == db.EINVAL
  58             if verbose: print val
  59         else:
  60             self.fail("expected exception")
  61
  62         try:
  63             data = d[100]
  64         except KeyError:
  65             pass
  66         else:
  67             self.fail("expected exception")
  68
  69         data = d.get(100)
  70         assert data == None
  71
  72         keys = d.keys()
  73         if verbose:
  74             print keys
  75         assert type(keys) == type([])
  76         assert type(keys[0]) == type(123)
  77         assert len(keys) == len(d)
  78
  79         items = d.items()
  80         if verbose:
  81             pprint(items)
  82         assert type(items) == type([])
  83         assert type(items[0]) == type(())
  84         assert len(items[0]) == 2
  85         assert type(items[0][0]) == type(123)
  86         assert type(items[0][1]) == type("")
  87         assert len(items) == len(d)
  88
  89         assert d.has_key(25)
  90
  91         del d[25]
  92         assert not d.has_key(25)
  93
  94         d.delete(13)
  95         assert not d.has_key(13)
  96
  97         data = d.get_both(26, "z" * 60)
  98         assert data == "z" * 60
  99         if verbose:
 100             print data
 101
 102         fd = d.fd()
 103         if verbose:
 104             print fd
 105
 106         c = d.cursor()
 107         rec = c.first()
 108         while rec:
 109             if verbose:
 110                 print rec
 111             rec = c.next()
 112
 113         c.set(50)
 114         rec = c.current()
 115         if verbose:
 116             print rec
 117
 118         c.put(-1, "a replacement record", db.DB_CURRENT)
 119
 120         c.set(50)
 121         rec = c.current()
 122         assert rec == (50, "a replacement record")
 123         if verbose:
 124             print rec
 125
 126         rec = c.set_range(30)
 127         if verbose:
 128             print rec
 129
 130         c.close()
 131         d.close()
 132
 133         d = db.DB()
 134         d.open(self.filename)
 135         c = d.cursor()
 136
 137         # put a record beyond the consecutive end of the recno's
 138         d[100] = "way out there"
 139         assert d[100] == "way out there"
 140
 141         try:
 142             data = d[99]
 143         except KeyError:
 144             pass
 145         else:
 146             self.fail("expected exception")
 147
 148         try:
 149             d.get(99)
 150         except db.DBKeyEmptyError, val:
 151             assert val[0] == db.DB_KEYEMPTY
 152             if verbose: print val
 153         else:
 154             self.fail("expected exception")
 155
 156         rec = c.set(40)
 157         while rec:
 158             if verbose:
 159                 print rec
 160             rec = c.next()
 161
 162         c.close()
 163         d.close()
 164
 165     def test02_WithSource(self):
 166         """
 167         A Recno file that is given a "backing source file" is essentially a
 168         simple ASCII file.  Normally each record is delimited by \n and so is
 169         just a line in the file, but you can set a different record delimiter
 170         if needed.
 171         """
 172         source = os.path.join(os.path.dirname(sys.argv[0]),
 173                               'db_home/test_recno.txt')
 174         f = open(source, 'w') # create the file
 175         f.close()
 176
 177         d = db.DB()
 178         # This is the default value, just checking if both int
 179         d.set_re_delim(0x0A)
 180         d.set_re_delim('\n')  # and char can be used...
 181         d.set_re_source(source)
 182         d.open(self.filename, db.DB_RECNO, db.DB_CREATE)
 183
 184         data = "The quick brown fox jumped over the lazy dog".split()
 185         for datum in data:
 186             d.append(datum)
 187         d.sync()
 188         d.close()
 189
 190         # get the text from the backing source
 191         text = open(source, 'r').read()
 192         text = text.strip()
 193         if verbose:
 194             print text
 195             print data
 196             print text.split('\n')
 197
 198         assert text.split('\n') == data
 199
 200         # open as a DB again
 201         d = db.DB()
 202         d.set_re_source(source)
 203         d.open(self.filename, db.DB_RECNO)
 204
 205         d[3] = 'reddish-brown'
 206         d[8] = 'comatose'
 207
 208         d.sync()
 209         d.close()
 210
 211         text = open(source, 'r').read()
 212         text = text.strip()
 213         if verbose:
 214             print text
 215             print text.split('\n')
 216
 217         assert text.split('\n') == \
 218              "The quick reddish-brown fox jumped over the comatose dog".split()
 219
 220     def test03_FixedLength(self):
 221         d = db.DB()
 222         d.set_re_len(40)  # fixed length records, 40 bytes long
 223         d.set_re_pad('-') # sets the pad character...
 224         d.set_re_pad(45)  # ...test both int and char
 225         d.open(self.filename, db.DB_RECNO, db.DB_CREATE)
 226
 227         for x in letters:
 228             d.append(x * 35)    # These will be padded
 229
 230         d.append('.' * 40)      # this one will be exact
 231
 232         try:                    # this one will fail
 233             d.append('bad' * 20)
 234         except db.DBInvalidArgError, val:
 235             assert val[0] == db.EINVAL
 236             if verbose: print val
 237         else:
 238             self.fail("expected exception")
 239
 240         c = d.cursor()
 241         rec = c.first()
 242         while rec:
 243             if verbose:
 244                 print rec
 245             rec = c.next()
 246
 247         c.close()
 248         d.close()
 249
 250
 251 #----------------------------------------------------------------------
 252
 253
 254 def test_suite():
 255     return unittest.makeSuite(SimpleRecnoTestCase)
 256
 257
 258 if __name__ == '__main__':
 259     unittest.main(defaultTest='test_suite')