bpo-44822: Don't truncate strs with embedded NULL chars returned by sqlite3 UDF callbacks (GH-27588)

(cherry picked from commit 8f010dc920)

Co-authored-by: Erlend Egeberg Aasland <erlend.aasland@innova.no>
This commit is contained in:
Miss Islington (bot) 2021-08-06 10:59:51 -07:00 committed by GitHub
parent a5d9963276
commit 2b1e713f87
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
3 changed files with 41 additions and 3 deletions

View file

@ -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):