mirror of
https://github.com/python/cpython.git
synced 2025-12-08 06:10:17 +00:00
[3.14] gh-139283: correctly handle size limit in cursor.fetchmany() (GH-139296) (GH-139441)
Passing a negative or zero size to `cursor.fetchmany()` made it fetch all rows
instead of none.
While this could be considered a security vulnerability, it was decided to treat
this issue as a regular bug as passing a non-sanitized *size* value in the first
place is not recommended.
(cherry picked from commit bc172ee830)
Co-authored-by: Bénédikt Tran <10796600+picnixz@users.noreply.github.com>
Co-authored-by: Petr Viktorin <encukou@gmail.com>
This commit is contained in:
parent
cd8fc3aad3
commit
cde02ae782
6 changed files with 134 additions and 19 deletions
|
|
@ -21,6 +21,7 @@
|
|||
# 3. This notice may not be removed or altered from any source distribution.
|
||||
|
||||
import contextlib
|
||||
import functools
|
||||
import os
|
||||
import sqlite3 as sqlite
|
||||
import subprocess
|
||||
|
|
@ -1060,7 +1061,7 @@ def test_array_size(self):
|
|||
# now set to 2
|
||||
self.cu.arraysize = 2
|
||||
|
||||
# now make the query return 3 rows
|
||||
# now make the query return 2 rows from a table of 3 rows
|
||||
self.cu.execute("delete from test")
|
||||
self.cu.execute("insert into test(name) values ('A')")
|
||||
self.cu.execute("insert into test(name) values ('B')")
|
||||
|
|
@ -1070,13 +1071,50 @@ def test_array_size(self):
|
|||
|
||||
self.assertEqual(len(res), 2)
|
||||
|
||||
def test_invalid_array_size(self):
|
||||
UINT32_MAX = (1 << 32) - 1
|
||||
setter = functools.partial(setattr, self.cu, 'arraysize')
|
||||
|
||||
self.assertRaises(TypeError, setter, 1.0)
|
||||
self.assertRaises(ValueError, setter, -3)
|
||||
self.assertRaises(OverflowError, setter, UINT32_MAX + 1)
|
||||
|
||||
def test_fetchmany(self):
|
||||
# no active SQL statement
|
||||
res = self.cu.fetchmany()
|
||||
self.assertEqual(res, [])
|
||||
res = self.cu.fetchmany(1000)
|
||||
self.assertEqual(res, [])
|
||||
|
||||
# test default parameter
|
||||
self.cu.execute("select name from test")
|
||||
res = self.cu.fetchmany()
|
||||
self.assertEqual(len(res), 1)
|
||||
|
||||
# test when the number of requested rows exceeds the actual count
|
||||
self.cu.execute("select name from test")
|
||||
res = self.cu.fetchmany(100)
|
||||
self.assertEqual(len(res), 1)
|
||||
res = self.cu.fetchmany(100)
|
||||
self.assertEqual(res, [])
|
||||
|
||||
# test when size = 0
|
||||
self.cu.execute("select name from test")
|
||||
res = self.cu.fetchmany(0)
|
||||
self.assertEqual(res, [])
|
||||
res = self.cu.fetchmany(100)
|
||||
self.assertEqual(len(res), 1)
|
||||
res = self.cu.fetchmany(100)
|
||||
self.assertEqual(res, [])
|
||||
|
||||
def test_invalid_fetchmany(self):
|
||||
UINT32_MAX = (1 << 32) - 1
|
||||
fetchmany = self.cu.fetchmany
|
||||
|
||||
self.assertRaises(TypeError, fetchmany, 1.0)
|
||||
self.assertRaises(ValueError, fetchmany, -3)
|
||||
self.assertRaises(OverflowError, fetchmany, UINT32_MAX + 1)
|
||||
|
||||
def test_fetchmany_kw_arg(self):
|
||||
"""Checks if fetchmany works with keyword arguments"""
|
||||
self.cu.execute("select name from test")
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue