Skip to content

bpo-45243: Use connection limits to simplify sqlite3 tests #29356

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
40 changes: 31 additions & 9 deletions Lib/test/test_sqlite3/test_dbapi.py
Original file line number Diff line number Diff line change
Expand Up @@ -29,7 +29,6 @@

from test.support import (
SHORT_TIMEOUT,
bigmemtest,
check_disallow_instantiation,
threading_helper,
)
Expand All @@ -48,6 +47,22 @@ def managed_connect(*args, in_mem=False, **kwargs):
unlink(TESTFN)


# Helper for temporary memory databases
def memory_database():
cx = sqlite.connect(":memory:")
return contextlib.closing(cx)


# Temporarily limit a database connection parameter
@contextlib.contextmanager
def cx_limit(cx, category=sqlite.SQLITE_LIMIT_LENGTH, limit=128):
try:
_prev = cx.setlimit(category, limit)
yield limit
finally:
cx.setlimit(category, _prev)


class ModuleTests(unittest.TestCase):
def test_api_level(self):
self.assertEqual(sqlite.apilevel, "2.0",
Expand Down Expand Up @@ -650,6 +665,15 @@ def __getitem__(slf, x):
with self.assertRaises(ZeroDivisionError):
self.cu.execute("select name from test where name=?", L())

def test_execute_too_many_params(self):
category = sqlite.SQLITE_LIMIT_VARIABLE_NUMBER
msg = "too many SQL variables"
with cx_limit(self.cx, category=category, limit=1):
self.cu.execute("select * from test where id=?", (1,))
with self.assertRaisesRegex(sqlite.OperationalError, msg):
self.cu.execute("select * from test where id!=? and id!=?",
(1, 2))

def test_execute_dict_mapping(self):
self.cu.execute("insert into test(name) values ('foo')")
self.cu.execute("select name from test where name=:name", {"name": "foo"})
Expand Down Expand Up @@ -1036,14 +1060,12 @@ def test_cursor_executescript_with_surrogates(self):
insert into a(s) values ('\ud8ff');
""")

@unittest.skipUnless(sys.maxsize > 2**32, 'requires 64bit platform')
@bigmemtest(size=2**31, memuse=3, dry_run=False)
def test_cursor_executescript_too_large_script(self, maxsize):
con = sqlite.connect(":memory:")
cur = con.cursor()
for size in 2**31-1, 2**31:
with self.assertRaises(sqlite.DataError):
cur.executescript("create table a(s);".ljust(size))
def test_cursor_executescript_too_large_script(self):
msg = "query string is too large"
with memory_database() as cx, cx_limit(cx) as lim:
cx.executescript("select 'almost too large'".ljust(lim-1))
with self.assertRaisesRegex(sqlite.DataError, msg):
cx.executescript("select 'too large'".ljust(lim))

def test_cursor_executescript_tx_control(self):
con = sqlite.connect(":memory:")
Expand Down
26 changes: 14 additions & 12 deletions Lib/test/test_sqlite3/test_regression.py
Original file line number Diff line number Diff line change
Expand Up @@ -28,7 +28,8 @@
import functools
from test import support

from .test_dbapi import managed_connect
from .test_dbapi import memory_database, managed_connect, cx_limit


class RegressionTests(unittest.TestCase):
def setUp(self):
Expand Down Expand Up @@ -356,17 +357,18 @@ def test_surrogates(self):
self.assertRaises(UnicodeEncodeError, cur.execute, "select '\ud8ff'")
self.assertRaises(UnicodeEncodeError, cur.execute, "select '\udcff'")

@unittest.skipUnless(sys.maxsize > 2**32, 'requires 64bit platform')
@support.bigmemtest(size=2**31, memuse=4, dry_run=False)
def test_large_sql(self, maxsize):
# Test two cases: size+1 > INT_MAX and size+1 <= INT_MAX.
for size in (2**31, 2**31-2):
con = sqlite.connect(":memory:")
sql = "select 1".ljust(size)
self.assertRaises(sqlite.DataError, con, sql)
cur = con.cursor()
self.assertRaises(sqlite.DataError, cur.execute, sql)
del sql
def test_large_sql(self):
msg = "query string is too large"
with memory_database() as cx, cx_limit(cx) as lim:
cu = cx.cursor()

cx("select 1".ljust(lim-1))
# use a different SQL statement; don't reuse from the LRU cache
cu.execute("select 2".ljust(lim-1))

sql = "select 3".ljust(lim)
self.assertRaisesRegex(sqlite.DataError, msg, cx, sql)
self.assertRaisesRegex(sqlite.DataError, msg, cu.execute, sql)

def test_commit_cursor_reset(self):
"""
Expand Down
9 changes: 9 additions & 0 deletions Lib/test/test_sqlite3/test_userfunctions.py
Original file line number Diff line number Diff line change
Expand Up @@ -31,6 +31,7 @@
import sqlite3 as sqlite

from test.support import bigmemtest
from .test_dbapi import cx_limit


def with_tracebacks(strings, traceback=True):
Expand Down Expand Up @@ -223,6 +224,14 @@ def test_func_error_on_create(self):
with self.assertRaises(sqlite.OperationalError):
self.con.create_function("bla", -100, lambda x: 2*x)

def test_func_too_many_args(self):
category = sqlite.SQLITE_LIMIT_FUNCTION_ARG
msg = "too many arguments on function"
with cx_limit(self.con, category=category, limit=1):
self.con.execute("select abs(-1)");
with self.assertRaisesRegex(sqlite.OperationalError, msg):
self.con.execute("select max(1, 2)");

def test_func_ref_count(self):
def getfunc():
def f():
Expand Down