equal
deleted
inserted
replaced
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() |
|