mirrors/cpython
1
0
Fork 0
mirror of https://github.com/python/cpython.git synced 2025-09-27 10:50:04 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 13

bpo-44430: Refactor sqlite3 threading tests (GH-26748)

Browse source 
1. Rewrite ThreadTests with a _run_test() helper method that does the heavy lifting
2. Add test.support.threading_helper.reap_threads to _run_test()
3. Use _run_test() in all threading tests
4. Add test case for sqlite3.Connection.set_trace_callback
5. Add test case for sqlite3.Connection.create_collation
This commit is contained in:
Erlend Egeberg Aasland 2021-06-20 21:26:36 +02:00 • committed by GitHub
parent 7d0a47e1af
commit 5f0fc30de4
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
1 changed files with 34 additions and 138 deletions
Download patch file Download diff file Expand all files Collapse all files

172
Lib/sqlite3/test/dbapi.py
View file

@ -581,158 +581,54 @@ class ThreadTests(unittest.TestCase):
def setUp(self): def setUp(self):
self.con = sqlite.connect(":memory:") self.con = sqlite.connect(":memory:")
self.cur = self.con.cursor() self.cur = self.con.cursor()
self.cur.execute("create table test(id integer primary key, name text, bin binary, ratio number, ts timestamp)") self.cur.execute("create table test(name text)")
def tearDown(self): def tearDown(self):
self.cur.close() self.cur.close()
self.con.close() self.con.close()
def test_con_cursor(self): @threading_helper.reap_threads
def run(con, errors): def _run_test(self, fn, *args, **kwds):
def run(err):
try: try:
cur = con.cursor() fn(*args, **kwds)
errors.append("did not raise ProgrammingError") err.append("did not raise ProgrammingError")
return
except sqlite.ProgrammingError: except sqlite.ProgrammingError:
return pass
except: except:
errors.append("raised wrong exception") err.append("raised wrong exception")
errors = [] err = []
t = threading.Thread(target=run, kwargs={"con": self.con, "errors": errors}) t = threading.Thread(target=run, kwargs={"err": err})
t.start() t.start()
t.join() t.join()
if len(errors) > 0: if err:
self.fail("\n".join(errors)) self.fail("\n".join(err))
def test_con_commit(self): def test_check_connection_thread(self):
def run(con, errors): fns = [
try: lambda: self.con.cursor(),
con.commit() lambda: self.con.commit(),
errors.append("did not raise ProgrammingError") lambda: self.con.rollback(),
return lambda: self.con.close(),
except sqlite.ProgrammingError: lambda: self.con.set_trace_callback(None),
return lambda: self.con.create_collation("foo", None),
except: ]
errors.append("raised wrong exception") for fn in fns:
with self.subTest(fn=fn):
self._run_test(fn)
errors = [] def test_check_cursor_thread(self):
t = threading.Thread(target=run, kwargs={"con": self.con, "errors": errors}) fns = [
t.start() lambda: self.cur.execute("insert into test(name) values('a')"),
t.join() lambda: self.cur.close(),
if len(errors) > 0: lambda: self.cur.execute("select name from test"),
self.fail("\n".join(errors)) lambda: self.cur.fetchone(),
]
for fn in fns:
with self.subTest(fn=fn):
self._run_test(fn)
def test_con_rollback(self):
def run(con, errors):
try:
con.rollback()
errors.append("did not raise ProgrammingError")
return
except sqlite.ProgrammingError:
return
except:
errors.append("raised wrong exception")
errors = []
t = threading.Thread(target=run, kwargs={"con": self.con, "errors": errors})
t.start()
t.join()
if len(errors) > 0:
self.fail("\n".join(errors))
def test_con_close(self):
def run(con, errors):
try:
con.close()
errors.append("did not raise ProgrammingError")
return
except sqlite.ProgrammingError:
return
except:
errors.append("raised wrong exception")
errors = []
t = threading.Thread(target=run, kwargs={"con": self.con, "errors": errors})
t.start()
t.join()
if len(errors) > 0:
self.fail("\n".join(errors))
def test_cur_implicit_begin(self):
def run(cur, errors):
try:
cur.execute("insert into test(name) values ('a')")
errors.append("did not raise ProgrammingError")
return
except sqlite.ProgrammingError:
return
except:
errors.append("raised wrong exception")
errors = []
t = threading.Thread(target=run, kwargs={"cur": self.cur, "errors": errors})
t.start()
t.join()
if len(errors) > 0:
self.fail("\n".join(errors))
def test_cur_close(self):
def run(cur, errors):
try:
cur.close()
errors.append("did not raise ProgrammingError")
return
except sqlite.ProgrammingError:
return
except:
errors.append("raised wrong exception")
errors = []
t = threading.Thread(target=run, kwargs={"cur": self.cur, "errors": errors})
t.start()
t.join()
if len(errors) > 0:
self.fail("\n".join(errors))
def test_cur_execute(self):
def run(cur, errors):
try:
cur.execute("select name from test")
errors.append("did not raise ProgrammingError")
return
except sqlite.ProgrammingError:
return
except:
errors.append("raised wrong exception")
errors = []
self.cur.execute("insert into test(name) values ('a')")
t = threading.Thread(target=run, kwargs={"cur": self.cur, "errors": errors})
t.start()
t.join()
if len(errors) > 0:
self.fail("\n".join(errors))
def test_cur_iter_next(self):
def run(cur, errors):
try:
row = cur.fetchone()
errors.append("did not raise ProgrammingError")
return
except sqlite.ProgrammingError:
return
except:
errors.append("raised wrong exception")
errors = []
self.cur.execute("insert into test(name) values ('a')")
self.cur.execute("select name from test")
t = threading.Thread(target=run, kwargs={"cur": self.cur, "errors": errors})
t.start()
t.join()
if len(errors) > 0:
self.fail("\n".join(errors))
@threading_helper.reap_threads @threading_helper.reap_threads
def test_dont_check_same_thread(self): def test_dont_check_same_thread(self):