1 import threading, os, time

     2 

     3 from logilab.common.testlib import TestCase, unittest_main

     4 from logilab.common.db import get_connection

     5 

     6 class SQLiteTC(TestCase):

     7     sqlite_file = '_extlite_test.sqlite'

     8     def setUp(self):

     9         cnx1 = get_connection('sqlite', database=self.sqlite_file)

    10         print 'SET IP'

    11         cu = cnx1.cursor()

    12         cu.execute('CREATE TABLE toto(name integer);')

    13         cnx1.commit()

    14         cnx1.close()

    15         

    16     def tearDown(self):

    17         try:

    18             os.remove(self.sqlite_file)

    19         except:

    20             pass

    21     def test(self):

    22         lock = threading.Lock()

    23         

    24         def run_thread():

    25             print 'run_thread'

    26             cnx2 = get_connection('sqlite', database=self.sqlite_file)

    27             lock.acquire()

    28             print 't2 sel1'

    29             cu = cnx2.cursor()

    30             cu.execute('SELECT name FROM toto')

    31             self.failIf(cu.fetchall())

    32             cnx2.commit()

    33             print 'done'

    34             lock.release()

    35             time.sleep(0.1)

    36             lock.acquire()

    37             print 't2 sel2'

    38             cu.execute('SELECT name FROM toto')

    39             self.failUnless(cu.fetchall())

    40             print 'done'

    41             lock.release()

    42 

    43         cnx1 = get_connection('sqlite', database=self.sqlite_file)

    44         lock.acquire()

    45         thread = threading.Thread(target=run_thread)

    46         thread.start()

    47         cu = cnx1.cursor()

    48         print 't1 sel'

    49         cu.execute('SELECT name FROM toto')

    50         print 'done'

    51         lock.release()

    52         time.sleep(0.1)

    53         cnx1.commit()

    54         lock.acquire()

    55         print 't1 insert'

    56         cu.execute("INSERT INTO toto(name) VALUES ('toto')")

    57         cnx1.commit()

    58         print 'done'

    59         lock.release()

    60 

    61 if __name__ == '__main__':

    62     unittest_main()