mirrors/cpython
1
0
Fork 0
mirror of https://github.com/python/cpython.git synced 2025-09-26 18:29:57 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 10

Clean up the sqlite3 tests (GH-93056)

Browse source 
Remove helper managed_connect(). Use memory_database() or
contextlib.closing() + addCleanup(unlink) instead.
This commit is contained in:
Serhiy Storchaka 2022-05-21 19:06:12 +03:00 • committed by GitHub
parent f9d6c59917
commit e5d8dbdd30
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
2 changed files with 23 additions and 33 deletions
Download patch file Download diff file Expand all files Collapse all files

42
Lib/test/test_sqlite3/test_dbapi.py
View file

@ -36,18 +36,6 @@ from os import SEEK_SET, SEEK_CUR, SEEK_END
from test.support.os_helper import TESTFN, TESTFN_UNDECODABLE, unlink, temp_dir, FakePath
# Helper for tests using TESTFN
@contextlib.contextmanager
def managed_connect(*args, in_mem=False, **kwargs):
cx = sqlite.connect(*args, **kwargs)
try:
yield cx
finally:
cx.close()
if not in_mem:
unlink(TESTFN)
# Helper for temporary memory databases
def memory_database(*args, **kwargs):
cx = sqlite.connect(":memory:", *args, **kwargs)
@ -331,7 +319,7 @@ class ModuleTests(unittest.TestCase):
@unittest.skipIf(sqlite.sqlite_version_info <= (3, 7, 16),
"Requires SQLite 3.7.16 or newer")
def test_extended_error_code_on_exception(self):
with managed_connect(":memory:", in_mem=True) as con:
with memory_database() as con:
with con:
con.execute("create table t(t integer check(t > 0))")
errmsg = "constraint failed"
@ -389,7 +377,7 @@ class ConnectionTests(unittest.TestCase):
def test_failed_open(self):
YOU_CANNOT_OPEN_THIS = "/foo/bar/bla/23534/mydb.db"
with self.assertRaises(sqlite.OperationalError):
con = sqlite.connect(YOU_CANNOT_OPEN_THIS)
sqlite.connect(YOU_CANNOT_OPEN_THIS)
def test_close(self):
self.cx.close()
@ -653,7 +641,9 @@ class OpenTests(unittest.TestCase):
""" Checks that we can successfully connect to a database using an object that
is PathLike, i.e. has __fspath__(). """
path = FakePath(TESTFN)
with managed_connect(path) as cx:
self.addCleanup(unlink, path)
self.assertFalse(os.path.exists(path))
with contextlib.closing(sqlite.connect(path)) as cx:
self.assertTrue(os.path.exists(path))
cx.execute(self._sql)
@ -663,23 +653,26 @@ class OpenTests(unittest.TestCase):
def test_open_with_undecodable_path(self):
path = TESTFN_UNDECODABLE
self.addCleanup(unlink, path)
with managed_connect(path, in_mem=True) as cx:
self.assertFalse(os.path.exists(path))
with contextlib.closing(sqlite.connect(path)) as cx:
self.assertTrue(os.path.exists(path))
cx.execute(self._sql)
def test_open_uri(self):
path = TESTFN
self.addCleanup(unlink, path)
uri = "file:" + urllib.parse.quote(os.fsencode(path))
self.assertFalse(os.path.exists(path))
with managed_connect(uri, uri=True) as cx:
with contextlib.closing(sqlite.connect(uri, uri=True)) as cx:
self.assertTrue(os.path.exists(path))
cx.execute(self._sql)
def test_open_unquoted_uri(self):
path = TESTFN
self.addCleanup(unlink, path)
uri = "file:" + path
self.assertFalse(os.path.exists(path))
with managed_connect(uri, uri=True) as cx:
with contextlib.closing(sqlite.connect(uri, uri=True)) as cx:
self.assertTrue(os.path.exists(path))
cx.execute(self._sql)
@ -695,7 +688,7 @@ class OpenTests(unittest.TestCase):
sqlite.connect(path).close()
self.assertTrue(os.path.exists(path))
# Cannot modify new DB
with managed_connect(uri, uri=True) as cx:
with contextlib.closing(sqlite.connect(uri, uri=True)) as cx:
with self.assertRaises(sqlite.OperationalError):
cx.execute(self._sql)
@ -704,14 +697,12 @@ class OpenTests(unittest.TestCase):
@unittest.skipUnless(TESTFN_UNDECODABLE, "only works if there are undecodable paths")
def test_open_undecodable_uri(self):
path = TESTFN_UNDECODABLE
self.addCleanup(unlink, path)
uri = "file:" + urllib.parse.quote(path)
self.assertFalse(os.path.exists(path))
try:
with managed_connect(uri, uri=True, in_mem=True) as cx:
with contextlib.closing(sqlite.connect(uri, uri=True)) as cx:
self.assertTrue(os.path.exists(path))
cx.execute(self._sql)
finally:
unlink(path)
def test_factory_database_arg(self):
def factory(database, *args, **kwargs):
@ -722,12 +713,11 @@ class OpenTests(unittest.TestCase):
for database in (TESTFN, os.fsencode(TESTFN),
FakePath(TESTFN), FakePath(os.fsencode(TESTFN))):
database_arg = None
with sqlite.connect(database, factory=factory):
pass
sqlite.connect(database, factory=factory).close()
self.assertEqual(database_arg, database)
def test_database_keyword(self):
with sqlite.connect(database=":memory:") as cx:
with contextlib.closing(sqlite.connect(database=":memory:")) as cx:
self.assertEqual(type(cx), sqlite.Connection)

10
Lib/test/test_sqlite3/test_regression.py
View file

@ -28,7 +28,7 @@ import functools
from test import support
from unittest.mock import patch
from test.test_sqlite3.test_dbapi import memory_database, managed_connect, cx_limit
from test.test_sqlite3.test_dbapi import memory_database, cx_limit
class RegressionTests(unittest.TestCase):
@ -422,7 +422,7 @@ class RegressionTests(unittest.TestCase):
self.assertEqual(val, b'')
def test_table_lock_cursor_replace_stmt(self):
with managed_connect(":memory:", in_mem=True) as con:
with memory_database() as con:
cur = con.cursor()
cur.execute("create table t(t)")
cur.executemany("insert into t values(?)",
@ -433,7 +433,7 @@ class RegressionTests(unittest.TestCase):
con.commit()
def test_table_lock_cursor_dealloc(self):
with managed_connect(":memory:", in_mem=True) as con:
with memory_database() as con:
con.execute("create table t(t)")
con.executemany("insert into t values(?)",
((v,) for v in range(5)))
@ -444,7 +444,7 @@ class RegressionTests(unittest.TestCase):
con.commit()
def test_table_lock_cursor_non_readonly_select(self):
with managed_connect(":memory:", in_mem=True) as con:
with memory_database() as con:
con.execute("create table t(t)")
con.executemany("insert into t values(?)",
((v,) for v in range(5)))
@ -459,7 +459,7 @@ class RegressionTests(unittest.TestCase):
con.commit()
def test_executescript_step_through_select(self):
with managed_connect(":memory:", in_mem=True) as con:
with memory_database() as con:
values = [(v,) for v in range(5)]
with con:
con.execute("create table t(t)")