home *** CD-ROM | disk | FTP | other *** search
/ Maximum CD 2010 November / maximum-cd-2010-11.iso / DiscContents / calibre-0.7.13.msi / file_2793 (.txt) < prev    next >
Encoding:
Python Compiled Bytecode  |  2010-08-06  |  6.7 KB  |  187 lines

  1. # Source Generated with Decompyle++
  2. # File: in.pyc (Python 2.6)
  3.  
  4. import sys
  5. import os
  6. import unittest
  7. import sqlite3 as sqlite
  8.  
  9. def get_db_path():
  10.     return 'sqlite_testdb'
  11.  
  12.  
  13. class TransactionTests(unittest.TestCase):
  14.     
  15.     def setUp(self):
  16.         
  17.         try:
  18.             os.remove(get_db_path())
  19.         except OSError:
  20.             pass
  21.  
  22.         self.con1 = sqlite.connect(get_db_path(), timeout = 0.1)
  23.         self.cur1 = self.con1.cursor()
  24.         self.con2 = sqlite.connect(get_db_path(), timeout = 0.1)
  25.         self.cur2 = self.con2.cursor()
  26.  
  27.     
  28.     def tearDown(self):
  29.         self.cur1.close()
  30.         self.con1.close()
  31.         self.cur2.close()
  32.         self.con2.close()
  33.         
  34.         try:
  35.             os.unlink(get_db_path())
  36.         except OSError:
  37.             pass
  38.  
  39.  
  40.     
  41.     def CheckDMLdoesAutoCommitBefore(self):
  42.         self.cur1.execute('create table test(i)')
  43.         self.cur1.execute('insert into test(i) values (5)')
  44.         self.cur1.execute('create table test2(j)')
  45.         self.cur2.execute('select i from test')
  46.         res = self.cur2.fetchall()
  47.         self.failUnlessEqual(len(res), 1)
  48.  
  49.     
  50.     def CheckInsertStartsTransaction(self):
  51.         self.cur1.execute('create table test(i)')
  52.         self.cur1.execute('insert into test(i) values (5)')
  53.         self.cur2.execute('select i from test')
  54.         res = self.cur2.fetchall()
  55.         self.failUnlessEqual(len(res), 0)
  56.  
  57.     
  58.     def CheckUpdateStartsTransaction(self):
  59.         self.cur1.execute('create table test(i)')
  60.         self.cur1.execute('insert into test(i) values (5)')
  61.         self.con1.commit()
  62.         self.cur1.execute('update test set i=6')
  63.         self.cur2.execute('select i from test')
  64.         res = self.cur2.fetchone()[0]
  65.         self.failUnlessEqual(res, 5)
  66.  
  67.     
  68.     def CheckDeleteStartsTransaction(self):
  69.         self.cur1.execute('create table test(i)')
  70.         self.cur1.execute('insert into test(i) values (5)')
  71.         self.con1.commit()
  72.         self.cur1.execute('delete from test')
  73.         self.cur2.execute('select i from test')
  74.         res = self.cur2.fetchall()
  75.         self.failUnlessEqual(len(res), 1)
  76.  
  77.     
  78.     def CheckReplaceStartsTransaction(self):
  79.         self.cur1.execute('create table test(i)')
  80.         self.cur1.execute('insert into test(i) values (5)')
  81.         self.con1.commit()
  82.         self.cur1.execute('replace into test(i) values (6)')
  83.         self.cur2.execute('select i from test')
  84.         res = self.cur2.fetchall()
  85.         self.failUnlessEqual(len(res), 1)
  86.         self.failUnlessEqual(res[0][0], 5)
  87.  
  88.     
  89.     def CheckToggleAutoCommit(self):
  90.         self.cur1.execute('create table test(i)')
  91.         self.cur1.execute('insert into test(i) values (5)')
  92.         self.con1.isolation_level = None
  93.         self.failUnlessEqual(self.con1.isolation_level, None)
  94.         self.cur2.execute('select i from test')
  95.         res = self.cur2.fetchall()
  96.         self.failUnlessEqual(len(res), 1)
  97.         self.con1.isolation_level = 'DEFERRED'
  98.         self.failUnlessEqual(self.con1.isolation_level, 'DEFERRED')
  99.         self.cur1.execute('insert into test(i) values (5)')
  100.         self.cur2.execute('select i from test')
  101.         res = self.cur2.fetchall()
  102.         self.failUnlessEqual(len(res), 1)
  103.  
  104.     
  105.     def CheckRaiseTimeout(self):
  106.         if sqlite.sqlite_version_info < (3, 2, 2):
  107.             return None
  108.         self.cur1.execute('create table test(i)')
  109.         self.cur1.execute('insert into test(i) values (5)')
  110.         
  111.         try:
  112.             self.cur2.execute('insert into test(i) values (5)')
  113.             self.fail('should have raised an OperationalError')
  114.         except sqlite.OperationalError:
  115.             sqlite.sqlite_version_info < (3, 2, 2)
  116.             sqlite.sqlite_version_info < (3, 2, 2)
  117.         except:
  118.             sqlite.sqlite_version_info < (3, 2, 2)
  119.             self.fail('should have raised an OperationalError')
  120.  
  121.  
  122.     
  123.     def CheckLocking(self):
  124.         if sqlite.sqlite_version_info < (3, 2, 2):
  125.             return None
  126.         self.cur1.execute('create table test(i)')
  127.         self.cur1.execute('insert into test(i) values (5)')
  128.         
  129.         try:
  130.             self.cur2.execute('insert into test(i) values (5)')
  131.             self.fail('should have raised an OperationalError')
  132.         except sqlite.OperationalError:
  133.             sqlite.sqlite_version_info < (3, 2, 2)
  134.             sqlite.sqlite_version_info < (3, 2, 2)
  135.         except:
  136.             sqlite.sqlite_version_info < (3, 2, 2)
  137.             self.fail('should have raised an OperationalError')
  138.  
  139.         self.con1.commit()
  140.  
  141.  
  142.  
  143. class SpecialCommandTests(unittest.TestCase):
  144.     
  145.     def setUp(self):
  146.         self.con = sqlite.connect(':memory:')
  147.         self.cur = self.con.cursor()
  148.  
  149.     
  150.     def CheckVacuum(self):
  151.         self.cur.execute('create table test(i)')
  152.         self.cur.execute('insert into test(i) values (5)')
  153.         self.cur.execute('vacuum')
  154.  
  155.     
  156.     def CheckDropTable(self):
  157.         self.cur.execute('create table test(i)')
  158.         self.cur.execute('insert into test(i) values (5)')
  159.         self.cur.execute('drop table test')
  160.  
  161.     
  162.     def CheckPragma(self):
  163.         self.cur.execute('create table test(i)')
  164.         self.cur.execute('insert into test(i) values (5)')
  165.         self.cur.execute('pragma count_changes=1')
  166.  
  167.     
  168.     def tearDown(self):
  169.         self.cur.close()
  170.         self.con.close()
  171.  
  172.  
  173.  
  174. def suite():
  175.     default_suite = unittest.makeSuite(TransactionTests, 'Check')
  176.     special_command_suite = unittest.makeSuite(SpecialCommandTests, 'Check')
  177.     return unittest.TestSuite((default_suite, special_command_suite))
  178.  
  179.  
  180. def test():
  181.     runner = unittest.TextTestRunner()
  182.     runner.run(suite())
  183.  
  184. if __name__ == '__main__':
  185.     test()
  186.  
  187.