bpo-45243: Use connection limits to simplify `sqlite3` tests (GH-29356)

3 files changed, 54 insertions, 21 deletions

diff --git a/Lib/test/test_sqlite3/test_dbapi.py b/Lib/test/test_sqlite3/test_dbapi.py
index ba3652a..6628eee 100644
--- a/Lib/test/test_sqlite3/test_dbapi.py
+++ b/Lib/test/test_sqlite3/test_dbapi.py
@@ -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:")
diff --git a/Lib/test/test_sqlite3/test_regression.py b/Lib/test/test_sqlite3/test_regression.py
index 3d71809..158f4cf 100644
--- a/Lib/test/test_sqlite3/test_regression.py
+++ b/Lib/test/test_sqlite3/test_regression.py
@@ -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):
         """
diff --git a/Lib/test/test_sqlite3/test_userfunctions.py b/Lib/test/test_sqlite3/test_userfunctions.py
index ad40847..62a11a5 100644
--- a/Lib/test/test_sqlite3/test_userfunctions.py
+++ b/Lib/test/test_sqlite3/test_userfunctions.py
@@ -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():