home *** CD-ROM | disk | FTP | other *** search
/ PC World 2005 June / PCWorld_2005-06_cd.bin / software / vyzkuste / firewally / firewally.exe / framework-2.3.exe / test_join.py < prev    next >
Text File  |  2003-12-30  |  4KB  |  121 lines
  1. """TestCases for using the DB.join and DBCursor.join_item methods.
  2. """
  3.  
  4. import sys, os, string
  5. import tempfile
  6. import time
  7. from pprint import pprint
  8.  
  9. try:
  10.     from threading import Thread, currentThread
  11.     have_threads = 1
  12. except ImportError:
  13.     have_threads = 0
  14.  
  15. import unittest
  16. from test_all import verbose
  17.  
  18. try:
  19.     # For Python 2.3
  20.     from bsddb import db, dbshelve
  21. except ImportError:
  22.     # For earlier Pythons w/distutils pybsddb
  23.     from bsddb3 import db, dbshelve
  24.  
  25.  
  26. #----------------------------------------------------------------------
  27.  
  28. ProductIndex = [
  29.     ('apple', "Convenience Store"),
  30.     ('blueberry', "Farmer's Market"),
  31.     ('shotgun', "S-Mart"),              # Aisle 12
  32.     ('pear', "Farmer's Market"),
  33.     ('chainsaw', "S-Mart"),             # "Shop smart.  Shop S-Mart!"
  34.     ('strawberry', "Farmer's Market"),
  35. ]
  36.  
  37. ColorIndex = [
  38.     ('blue', "blueberry"),
  39.     ('red', "apple"),
  40.     ('red', "chainsaw"),
  41.     ('red', "strawberry"),
  42.     ('yellow', "peach"),
  43.     ('yellow', "pear"),
  44.     ('black', "shotgun"),
  45. ]
  46.  
  47. class JoinTestCase(unittest.TestCase):
  48.     keytype = ''
  49.  
  50.     def setUp(self):
  51.         self.filename = self.__class__.__name__ + '.db'
  52.         homeDir = os.path.join(os.path.dirname(sys.argv[0]), 'db_home')
  53.         self.homeDir = homeDir
  54.         try: os.mkdir(homeDir)
  55.         except os.error: pass
  56.         self.env = db.DBEnv()
  57.         self.env.open(homeDir, db.DB_CREATE | db.DB_INIT_MPOOL | db.DB_INIT_LOCK )
  58.  
  59.     def tearDown(self):
  60.         self.env.close()
  61.         import glob
  62.         files = glob.glob(os.path.join(self.homeDir, '*'))
  63.         for file in files:
  64.             os.remove(file)
  65.  
  66.     def test01_join(self):
  67.         if verbose:
  68.             print '\n', '-=' * 30
  69.             print "Running %s.test01_join..." % \
  70.                   self.__class__.__name__
  71.  
  72.         # create and populate primary index
  73.         priDB = db.DB(self.env)
  74.         priDB.open(self.filename, "primary", db.DB_BTREE, db.DB_CREATE)
  75.         map(lambda t, priDB=priDB: apply(priDB.put, t), ProductIndex)
  76.  
  77.         # create and populate secondary index
  78.         secDB = db.DB(self.env)
  79.         secDB.set_flags(db.DB_DUP | db.DB_DUPSORT)
  80.         secDB.open(self.filename, "secondary", db.DB_BTREE, db.DB_CREATE)
  81.         map(lambda t, secDB=secDB: apply(secDB.put, t), ColorIndex)
  82.  
  83.         sCursor = None
  84.         jCursor = None
  85.         try:
  86.             # lets look up all of the red Products
  87.             sCursor = secDB.cursor()
  88.             # Don't do the .set() in an assert, or you can get a bogus failure
  89.             # when running python -O
  90.             tmp = sCursor.set('red')
  91.             assert tmp
  92.  
  93.             # FIXME: jCursor doesn't properly hold a reference to its
  94.             # cursors, if they are closed before jcursor is used it
  95.             # can cause a crash.
  96.             jCursor = priDB.join([sCursor])
  97.  
  98.             if jCursor.get(0) != ('apple', "Convenience Store"):
  99.                 self.fail("join cursor positioned wrong")
  100.             if jCursor.join_item() != 'chainsaw':
  101.                 self.fail("DBCursor.join_item returned wrong item")
  102.             if jCursor.get(0)[0] != 'strawberry':
  103.                 self.fail("join cursor returned wrong thing")
  104.             if jCursor.get(0):  # there were only three red items to return
  105.                 self.fail("join cursor returned too many items")
  106.         finally:
  107.             if jCursor:
  108.                 jCursor.close()
  109.             if sCursor:
  110.                 sCursor.close()
  111.             priDB.close()
  112.             secDB.close()
  113.  
  114.  
  115. def test_suite():
  116.     suite = unittest.TestSuite()
  117.  
  118.     suite.addTest(unittest.makeSuite(JoinTestCase))
  119.  
  120.     return suite
  121.