bpo-44822: Don't truncate `str`s with embedded NULL chars returned by `sqlite3` UDF callbacks (GH-27588)
(cherry picked from commit 8f010dc920e1f6dc6a357e7cc1460a7a567c05c6)
Co-authored-by: Erlend Egeberg Aasland <erlend.aasland@innova.no>
diff --git a/Lib/sqlite3/test/userfunctions.py b/Lib/sqlite3/test/userfunctions.py
index 4290890..75e8035 100644
--- a/Lib/sqlite3/test/userfunctions.py
+++ b/Lib/sqlite3/test/userfunctions.py
@@ -28,6 +28,8 @@
def func_returntext():
return "foo"
+def func_returntextwithnull():
+ return "1\x002"
def func_returnunicode():
return "bar"
def func_returnint():
@@ -138,11 +140,21 @@ def step(self, val):
def finalize(self):
return self.val
+class AggrText:
+ def __init__(self):
+ self.txt = ""
+ def step(self, txt):
+ self.txt = self.txt + txt
+ def finalize(self):
+ return self.txt
+
+
class FunctionTests(unittest.TestCase):
def setUp(self):
self.con = sqlite.connect(":memory:")
self.con.create_function("returntext", 0, func_returntext)
+ self.con.create_function("returntextwithnull", 0, func_returntextwithnull)
self.con.create_function("returnunicode", 0, func_returnunicode)
self.con.create_function("returnint", 0, func_returnint)
self.con.create_function("returnfloat", 0, func_returnfloat)
@@ -186,6 +198,12 @@ def test_func_return_text(self):
self.assertEqual(type(val), str)
self.assertEqual(val, "foo")
+ def test_func_return_text_with_null_char(self):
+ cur = self.con.cursor()
+ res = cur.execute("select returntextwithnull()").fetchone()[0]
+ self.assertEqual(type(res), str)
+ self.assertEqual(res, "1\x002")
+
def test_func_return_unicode(self):
cur = self.con.cursor()
cur.execute("select returnunicode()")
@@ -364,6 +382,7 @@ def setUp(self):
self.con.create_aggregate("checkType", 2, AggrCheckType)
self.con.create_aggregate("checkTypes", -1, AggrCheckTypes)
self.con.create_aggregate("mysum", 1, AggrSum)
+ self.con.create_aggregate("aggtxt", 1, AggrText)
def tearDown(self):
#self.cur.close()
@@ -457,6 +476,15 @@ def test_aggr_no_match(self):
val = cur.fetchone()[0]
self.assertIsNone(val)
+ def test_aggr_text(self):
+ cur = self.con.cursor()
+ for txt in ["foo", "1\x002"]:
+ with self.subTest(txt=txt):
+ cur.execute("select aggtxt(?) from test", (txt,))
+ val = cur.fetchone()[0]
+ self.assertEqual(val, txt)
+
+
class AuthorizerTests(unittest.TestCase):
@staticmethod
def authorizer_cb(action, arg1, arg2, dbname, source):