1 """

     2 TestCases for exercising a Queue DB.

     3 """

     4 

     5 import os, string

     6 from pprint import pprint

     7 import unittest

     8 

     9 from test_all import db, verbose, get_new_database_path

    10 

    11 #----------------------------------------------------------------------

    12 

    13 class SimpleQueueTestCase(unittest.TestCase):

    14     def setUp(self):

    15         self.filename = get_new_database_path()

    16 

    17     def tearDown(self):

    18         try:

    19             os.remove(self.filename)

    20         except os.error:

    21             pass

    22 

    23 

    24     def test01_basic(self):

    25         # Basic Queue tests using the deprecated DBCursor.consume method.

    26 

    27         if verbose:

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

    29             print "Running %s.test01_basic..." % self.__class__.__name__

    30 

    31         d = db.DB()

    32         d.set_re_len(40)  # Queues must be fixed length

    33         d.open(self.filename, db.DB_QUEUE, db.DB_CREATE)

    34 

    35         if verbose:

    36             print "before appends" + '-' * 30

    37             pprint(d.stat())

    38 

    39         for x in string.letters:

    40             d.append(x * 40)

    41 

    42         self.assertEqual(len(d), len(string.letters))

    43 

    44         d.put(100, "some more data")

    45         d.put(101, "and some more ")

    46         d.put(75,  "out of order")

    47         d.put(1,   "replacement data")

    48 

    49         self.assertEqual(len(d), len(string.letters)+3)

    50 

    51         if verbose:

    52             print "before close" + '-' * 30

    53             pprint(d.stat())

    54 

    55         d.close()

    56         del d

    57         d = db.DB()

    58         d.open(self.filename)

    59 

    60         if verbose:

    61             print "after open" + '-' * 30

    62             pprint(d.stat())

    63 

    64         # Test "txn" as a positional parameter

    65         d.append("one more", None)

    66         # Test "txn" as a keyword parameter

    67         d.append("another one", txn=None)

    68 

    69         c = d.cursor()

    70 

    71         if verbose:

    72             print "after append" + '-' * 30

    73             pprint(d.stat())

    74 

    75         rec = c.consume()

    76         while rec:

    77             if verbose:

    78                 print rec

    79             rec = c.consume()

    80         c.close()

    81 

    82         if verbose:

    83             print "after consume loop" + '-' * 30

    84             pprint(d.stat())

    85 

    86         self.assertEqual(len(d), 0, \

    87                "if you see this message then you need to rebuild " \

    88                "Berkeley DB 3.1.17 with the patch in patches/qam_stat.diff")

    89 

    90         d.close()

    91 

    92 

    93 

    94     def test02_basicPost32(self):

    95         # Basic Queue tests using the new DB.consume method in DB 3.2+

    96         # (No cursor needed)

    97 

    98         if verbose:

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

   100             print "Running %s.test02_basicPost32..." % self.__class__.__name__

   101 

   102         if db.version() < (3, 2, 0):

   103             if verbose:

   104                 print "Test not run, DB not new enough..."

   105             return

   106 

   107         d = db.DB()

   108         d.set_re_len(40)  # Queues must be fixed length

   109         d.open(self.filename, db.DB_QUEUE, db.DB_CREATE)

   110 

   111         if verbose:

   112             print "before appends" + '-' * 30

   113             pprint(d.stat())

   114 

   115         for x in string.letters:

   116             d.append(x * 40)

   117 

   118         self.assertEqual(len(d), len(string.letters))

   119 

   120         d.put(100, "some more data")

   121         d.put(101, "and some more ")

   122         d.put(75,  "out of order")

   123         d.put(1,   "replacement data")

   124 

   125         self.assertEqual(len(d), len(string.letters)+3)

   126 

   127         if verbose:

   128             print "before close" + '-' * 30

   129             pprint(d.stat())

   130 

   131         d.close()

   132         del d

   133         d = db.DB()

   134         d.open(self.filename)

   135         #d.set_get_returns_none(true)

   136 

   137         if verbose:

   138             print "after open" + '-' * 30

   139             pprint(d.stat())

   140 

   141         d.append("one more")

   142 

   143         if verbose:

   144             print "after append" + '-' * 30

   145             pprint(d.stat())

   146 

   147         rec = d.consume()

   148         while rec:

   149             if verbose:

   150                 print rec

   151             rec = d.consume()

   152 

   153         if verbose:

   154             print "after consume loop" + '-' * 30

   155             pprint(d.stat())

   156 

   157         d.close()

   158 

   159 

   160 

   161 #----------------------------------------------------------------------

   162 

   163 def test_suite():

   164     return unittest.makeSuite(SimpleQueueTestCase)

   165 

   166 

   167 if __name__ == '__main__':

   168     unittest.main(defaultTest='test_suite')