mirror of https://github.com/python/cpython
bpo-43853: Expand test suite for SQLite UDF's (GH-27642)
This commit is contained in:
parent
ac0c6e128c
commit
3eb3b4f270
|
@ -23,7 +23,6 @@
|
|||
|
||||
import contextlib
|
||||
import functools
|
||||
import gc
|
||||
import io
|
||||
import re
|
||||
import sys
|
||||
|
@ -31,7 +30,8 @@ import unittest
|
|||
import unittest.mock
|
||||
import sqlite3 as sqlite
|
||||
|
||||
from test.support import bigmemtest, catch_unraisable_exception
|
||||
from test.support import bigmemtest, catch_unraisable_exception, gc_collect
|
||||
|
||||
from test.test_sqlite3.test_dbapi import cx_limit
|
||||
|
||||
|
||||
|
@ -94,22 +94,6 @@ def func_memoryerror():
|
|||
def func_overflowerror():
|
||||
raise OverflowError
|
||||
|
||||
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
|
||||
|
@ -210,17 +194,15 @@ 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("memoryerror", 0, func_memoryerror)
|
||||
self.con.create_function("overflowerror", 0, func_overflowerror)
|
||||
|
||||
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):
|
||||
|
@ -305,6 +287,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()
|
||||
self.assertRaisesRegex(sqlite.DataError, "string or blob too big",
|
||||
self.con.execute, "select returntoolargeint()")
|
||||
|
||||
@with_tracebacks(ZeroDivisionError, name="func_raiseexception")
|
||||
def test_func_exception(self):
|
||||
cur = self.con.cursor()
|
||||
|
@ -327,44 +319,6 @@ class FunctionTests(unittest.TestCase):
|
|||
cur.execute("select overflowerror()")
|
||||
cur.fetchone()
|
||||
|
||||
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))
|
||||
|
@ -375,6 +329,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
|
||||
|
@ -430,7 +430,7 @@ class FunctionTests(unittest.TestCase):
|
|||
y.append(y)
|
||||
|
||||
del x,y
|
||||
gc.collect()
|
||||
gc_collect()
|
||||
|
||||
@with_tracebacks(OverflowError)
|
||||
def test_func_return_too_large_int(self):
|
||||
|
|
|
@ -559,7 +559,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);
|
||||
|
|
|
@ -166,9 +166,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)
|
||||
|
|
Loading…
Reference in New Issue