mirror of https://github.com/python/cpython
bpo-45243: Use connection limits to simplify `sqlite3` tests (GH-29356)
This commit is contained in:
parent
71e8a3e76a
commit
3d42cd9461
|
@ -29,7 +29,6 @@ import unittest
|
|||
|
||||
from test.support import (
|
||||
SHORT_TIMEOUT,
|
||||
bigmemtest,
|
||||
check_disallow_instantiation,
|
||||
threading_helper,
|
||||
)
|
||||
|
@ -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",
|
||||
|
@ -650,6 +665,15 @@ class CursorTests(unittest.TestCase):
|
|||
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"})
|
||||
|
@ -1036,14 +1060,12 @@ class ExtensionTests(unittest.TestCase):
|
|||
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:")
|
||||
|
|
|
@ -28,7 +28,8 @@ import weakref
|
|||
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):
|
||||
|
@ -356,17 +357,18 @@ class RegressionTests(unittest.TestCase):
|
|||
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):
|
||||
"""
|
||||
|
|
|
@ -31,6 +31,7 @@ import unittest.mock
|
|||
import sqlite3 as sqlite
|
||||
|
||||
from test.support import bigmemtest
|
||||
from .test_dbapi import cx_limit
|
||||
|
||||
|
||||
def with_tracebacks(strings, traceback=True):
|
||||
|
@ -223,6 +224,14 @@ class FunctionTests(unittest.TestCase):
|
|||
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():
|
||||
|
|
Loading…
Reference in New Issue