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 

     9     def _cleanup(self):

    10         try:

    11             os.remove(self.sqlite_file)

    12         except:

    13             pass

    14 

    15     def setUp(self):

    16         self._cleanup()

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

    18         cu = cnx1.cursor()

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

    20         cnx1.commit()

    21         cnx1.close()

    22 

    23     def tearDown(self):

    24         self._cleanup()

    25 

    26     def test(self):

    27         lock = threading.Lock()

    28 

    29         def run_thread():

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

    31             lock.acquire()

    32             cu = cnx2.cursor()

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

    34             self.failIf(cu.fetchall())

    35             cnx2.commit()

    36             lock.release()

    37             time.sleep(0.1)

    38             lock.acquire()

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

    40             self.failUnless(cu.fetchall())

    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         cu.execute('SELECT name FROM toto')

    49         lock.release()

    50         time.sleep(0.1)

    51         cnx1.commit()

    52         lock.acquire()

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

    54         cnx1.commit()

    55         lock.release()

    56 

    57 if __name__ == '__main__':

    58     unittest_main()