Mirrors
/
cpython
mirror of https://github.com/python/cpython
3
0
Fork 0
Code Releases Activity
Browse Source

[3.10] bpo-43853: Expand test suite for SQLite UDF's (GH-27642) (GH-31030) 

* [3.10] bpo-43853: Expand test suite for SQLite UDF's (GH-27642).
(cherry picked from commit 3eb3b4f270)

Co-authored-by: Erlend Egeberg Aasland <erlend.aasland@innova.no>

* Fix test_func_return_too_large_int

GH-27613 (bpo 44839) was not backported, so exceptions differ between
main (3.11) and older versions.
pull/31472/head
Erlend Egeberg Aasland 4 years ago
committed by GitHub
parent
f1916cde24
commit
ba457fe6f8
No known key found for this signature in database GPG Key ID: 4AEE18F83AFDEB23
3 changed files with 79 additions and 66 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 128
      Lib/sqlite3/test/userfunctions.py
  2. 6
      Modules/_sqlite/connection.c
  3. 11
      Modules/_sqlite/statement.c

128
Lib/sqlite3/test/userfunctions.py
View File

@ -23,9 +23,11 @@
import unittest
import unittest.mock
import gc
import sqlite3 as sqlite
from test.support import gc_collect
def func_returntext():
return "foo"
def func_returntextwithnull():
@ -45,22 +47,6 @@ def func_returnlonglong():
def func_raiseexception():
5/0
def func_isstring(v):
return type(v) is str
def func_isint(v):
return type(v) is int
def func_isfloat(v):
return type(v) is float
def func_isnone(v):
return type(v) is type(None)
def func_isblob(v):
return isinstance(v, (bytes, memoryview))
def func_islonglong(v):
return isinstance(v, int) and v >= 1<<31
def func(*args):
return len(args)
class AggrNoStep:
def __init__(self):
pass
@ -161,15 +147,13 @@ class FunctionTests(unittest.TestCase):
self.con.create_function("returnnull", 0, func_returnnull)
self.con.create_function("returnblob", 0, func_returnblob)
self.con.create_function("returnlonglong", 0, func_returnlonglong)
self.con.create_function("returnnan", 0, lambda: float("nan"))
self.con.create_function("returntoolargeint", 0, lambda: 1 << 65)
self.con.create_function("raiseexception", 0, func_raiseexception)
self.con.create_function("isstring", 1, func_isstring)
self.con.create_function("isint", 1, func_isint)
self.con.create_function("isfloat", 1, func_isfloat)
self.con.create_function("isnone", 1, func_isnone)
self.con.create_function("isblob", 1, func_isblob)
self.con.create_function("islonglong", 1, func_islonglong)
self.con.create_function("spam", -1, func)
self.con.create_function("isblob", 1, lambda x: isinstance(x, bytes))
self.con.create_function("isnone", 1, lambda x: x is None)
self.con.create_function("spam", -1, lambda *x: len(x))
self.con.execute("create table test(t text)")
def tearDown(self):
@ -246,6 +230,16 @@ class FunctionTests(unittest.TestCase):
val = cur.fetchone()[0]
self.assertEqual(val, 1<<31)
def test_func_return_nan(self):
cur = self.con.cursor()
cur.execute("select returnnan()")
self.assertIsNone(cur.fetchone()[0])
def test_func_return_too_large_int(self):
cur = self.con.cursor()
with self.assertRaises(sqlite.OperationalError):
self.con.execute("select returntoolargeint()")
def test_func_exception(self):
cur = self.con.cursor()
with self.assertRaises(sqlite.OperationalError) as cm:
@ -253,44 +247,6 @@ class FunctionTests(unittest.TestCase):
cur.fetchone()
self.assertEqual(str(cm.exception), 'user-defined function raised exception')
def test_param_string(self):
cur = self.con.cursor()
for text in ["foo", str()]:
with self.subTest(text=text):
cur.execute("select isstring(?)", (text,))
val = cur.fetchone()[0]
self.assertEqual(val, 1)
def test_param_int(self):
cur = self.con.cursor()
cur.execute("select isint(?)", (42,))
val = cur.fetchone()[0]
self.assertEqual(val, 1)
def test_param_float(self):
cur = self.con.cursor()
cur.execute("select isfloat(?)", (3.14,))
val = cur.fetchone()[0]
self.assertEqual(val, 1)
def test_param_none(self):
cur = self.con.cursor()
cur.execute("select isnone(?)", (None,))
val = cur.fetchone()[0]
self.assertEqual(val, 1)
def test_param_blob(self):
cur = self.con.cursor()
cur.execute("select isblob(?)", (memoryview(b"blob"),))
val = cur.fetchone()[0]
self.assertEqual(val, 1)
def test_param_long_long(self):
cur = self.con.cursor()
cur.execute("select islonglong(?)", (1<<42,))
val = cur.fetchone()[0]
self.assertEqual(val, 1)
def test_any_arguments(self):
cur = self.con.cursor()
cur.execute("select spam(?, ?)", (1, 2))
@ -301,6 +257,52 @@ class FunctionTests(unittest.TestCase):
cur = self.con.execute("select isblob(x'')")
self.assertTrue(cur.fetchone()[0])
def test_nan_float(self):
cur = self.con.execute("select isnone(?)", (float("nan"),))
# SQLite has no concept of nan; it is converted to NULL
self.assertTrue(cur.fetchone()[0])
def test_too_large_int(self):
err = "Python int too large to convert to SQLite INTEGER"
self.assertRaisesRegex(OverflowError, err, self.con.execute,
"select spam(?)", (1 << 65,))
def test_non_contiguous_blob(self):
self.assertRaisesRegex(ValueError, "could not convert BLOB to buffer",
self.con.execute, "select spam(?)",
(memoryview(b"blob")[::2],))
def test_param_surrogates(self):
self.assertRaisesRegex(UnicodeEncodeError, "surrogates not allowed",
self.con.execute, "select spam(?)",
("\ud803\ude6d",))
def test_func_params(self):
results = []
def append_result(arg):
results.append((arg, type(arg)))
self.con.create_function("test_params", 1, append_result)
dataset = [
(42, int),
(-1, int),
(1234567890123456789, int),
(4611686018427387905, int), # 63-bit int with non-zero low bits
(3.14, float),
(float('inf'), float),
("text", str),
("1\x002", str),
("\u02e2q\u02e1\u2071\u1d57\u1d49", str),
(b"blob", bytes),
(bytearray(range(2)), bytes),
(memoryview(b"blob"), bytes),
(None, type(None)),
]
for val, _ in dataset:
cur = self.con.execute("select test_params(?)", (val,))
cur.fetchone()
self.assertEqual(dataset, results)
# Regarding deterministic functions:
#
# Between 3.8.3 and 3.15.0, deterministic functions were only used to
@ -356,7 +358,7 @@ class FunctionTests(unittest.TestCase):
y.append(y)
del x,y
gc.collect()
gc_collect()
class AggregateTests(unittest.TestCase):
def setUp(self):

6
Modules/_sqlite/connection.c
View File

@ -552,7 +552,11 @@ _pysqlite_set_result(sqlite3_context* context, PyObject* py_val)
return -1;
sqlite3_result_int64(context, value);
} else if (PyFloat_Check(py_val)) {
sqlite3_result_double(context, PyFloat_AsDouble(py_val));
double value = PyFloat_AsDouble(py_val);
if (value == -1 && PyErr_Occurred()) {
return -1;
}
sqlite3_result_double(context, value);
} else if (PyUnicode_Check(py_val)) {
Py_ssize_t sz;
const char *str = PyUnicode_AsUTF8AndSize(py_val, &sz);

11
Modules/_sqlite/statement.c
View File

@ -172,9 +172,16 @@ int pysqlite_statement_bind_parameter(pysqlite_Statement* self, int pos, PyObjec
rc = sqlite3_bind_int64(self->st, pos, value);
break;
}
case TYPE_FLOAT:
rc = sqlite3_bind_double(self->st, pos, PyFloat_AsDouble(parameter));
case TYPE_FLOAT: {
double value = PyFloat_AsDouble(parameter);
if (value == -1 && PyErr_Occurred()) {
rc = -1;
}
else {
rc = sqlite3_bind_double(self->st, pos, value);
}
break;
}
case TYPE_UNICODE:
string = PyUnicode_AsUTF8AndSize(parameter, &buflen);
if (string == NULL)

Write Preview
Loading…
Cancel
Save