import unittest
import sqlite3 as sqlite
import weakref
+import functools
from test import support
class RegressionTests(unittest.TestCase):
with self.assertRaises(AttributeError):
del self.con.isolation_level
+ def CheckBpo37347(self):
+ class Printer:
+ def log(self, *args):
+ return sqlite.SQLITE_OK
-class UnhashableFunc:
- __hash__ = None
+ for method in [self.con.set_trace_callback,
+ functools.partial(self.con.set_progress_handler, n=1),
+ self.con.set_authorizer]:
+ printer_instance = Printer()
+ method(printer_instance.log)
+ method(printer_instance.log)
+ self.con.execute("select 1") # trigger seg fault
+ method(None)
- def __init__(self, return_value=None):
- self.calls = 0
- self.return_value = return_value
-
- def __call__(self, *args, **kwargs):
- self.calls += 1
- return self.return_value
-
-
-class UnhashableCallbacksTestCase(unittest.TestCase):
- """
- https://bugs.python.org/issue34052
-
- Registering unhashable callbacks raises TypeError, callbacks are not
- registered in SQLite after such registration attempt.
- """
- def setUp(self):
- self.con = sqlite.connect(':memory:')
-
- def tearDown(self):
- self.con.close()
-
- def test_progress_handler(self):
- f = UnhashableFunc(return_value=0)
- with self.assertRaisesRegex(TypeError, 'unhashable type'):
- self.con.set_progress_handler(f, 1)
- self.con.execute('SELECT 1')
- self.assertFalse(f.calls)
-
- def test_func(self):
- func_name = 'func_name'
- f = UnhashableFunc()
- with self.assertRaisesRegex(TypeError, 'unhashable type'):
- self.con.create_function(func_name, 0, f)
- msg = 'no such function: %s' % func_name
- with self.assertRaisesRegex(sqlite.OperationalError, msg):
- self.con.execute('SELECT %s()' % func_name)
- self.assertFalse(f.calls)
-
- def test_authorizer(self):
- f = UnhashableFunc(return_value=sqlite.SQLITE_DENY)
- with self.assertRaisesRegex(TypeError, 'unhashable type'):
- self.con.set_authorizer(f)
- self.con.execute('SELECT 1')
- self.assertFalse(f.calls)
-
- def test_aggr(self):
- class UnhashableType(type):
- __hash__ = None
- aggr_name = 'aggr_name'
- with self.assertRaisesRegex(TypeError, 'unhashable type'):
- self.con.create_aggregate(aggr_name, 0, UnhashableType('Aggr', (), {}))
- msg = 'no such function: %s' % aggr_name
- with self.assertRaisesRegex(sqlite.OperationalError, msg):
- self.con.execute('SELECT %s()' % aggr_name)
def suite():
regression_suite = unittest.makeSuite(RegressionTests, "Check")
return unittest.TestSuite((
regression_suite,
- unittest.makeSuite(UnhashableCallbacksTestCase),
))
def test():
}
self->check_same_thread = check_same_thread;
- Py_XSETREF(self->function_pinboard, PyDict_New());
- if (!self->function_pinboard) {
- return -1;
- }
+ self->function_pinboard_trace_callback = NULL;
+ self->function_pinboard_progress_handler = NULL;
+ self->function_pinboard_authorizer_cb = NULL;
Py_XSETREF(self->collations, PyDict_New());
if (!self->collations) {
/* Clean up if user has not called .close() explicitly. */
if (self->db) {
- Py_BEGIN_ALLOW_THREADS
SQLITE3_CLOSE(self->db);
- Py_END_ALLOW_THREADS
}
Py_XDECREF(self->isolation_level);
- Py_XDECREF(self->function_pinboard);
+ Py_XDECREF(self->function_pinboard_trace_callback);
+ Py_XDECREF(self->function_pinboard_progress_handler);
+ Py_XDECREF(self->function_pinboard_authorizer_cb);
Py_XDECREF(self->row_factory);
Py_XDECREF(self->text_factory);
Py_XDECREF(self->collations);
Py_XDECREF(self->statements);
Py_XDECREF(self->cursors);
-
Py_TYPE(self)->tp_free((PyObject*)self);
}
pysqlite_do_all_statements(self, ACTION_FINALIZE, 1);
if (self->db) {
- Py_BEGIN_ALLOW_THREADS
rc = SQLITE3_CLOSE(self->db);
- Py_END_ALLOW_THREADS
if (rc != SQLITE_OK) {
_pysqlite_seterror(self->db, NULL);
Py_SETREF(self->cursors, new_list);
}
+static void _destructor(void* args)
+{
+ Py_DECREF((PyObject*)args);
+}
+
PyObject* pysqlite_connection_create_function(pysqlite_Connection* self, PyObject* args, PyObject* kwargs)
{
static char *kwlist[] = {"name", "narg", "func", "deterministic", NULL};
flags |= SQLITE_DETERMINISTIC;
#endif
}
- if (PyDict_SetItem(self->function_pinboard, func, Py_None) == -1) {
- return NULL;
- }
- rc = sqlite3_create_function(self->db,
- name,
- narg,
- flags,
- (void*)func,
- _pysqlite_func_callback,
- NULL,
- NULL);
+ Py_INCREF(func);
+ rc = sqlite3_create_function_v2(self->db,
+ name,
+ narg,
+ flags,
+ (void*)func,
+ _pysqlite_func_callback,
+ NULL,
+ NULL,
+ &_destructor); // will decref func
if (rc != SQLITE_OK) {
/* Workaround for SQLite bug: no error code or string is available here */
kwlist, &name, &n_arg, &aggregate_class)) {
return NULL;
}
-
- if (PyDict_SetItem(self->function_pinboard, aggregate_class, Py_None) == -1) {
- return NULL;
- }
- rc = sqlite3_create_function(self->db, name, n_arg, SQLITE_UTF8, (void*)aggregate_class, 0, &_pysqlite_step_callback, &_pysqlite_final_callback);
+ Py_INCREF(aggregate_class);
+ rc = sqlite3_create_function_v2(self->db,
+ name,
+ n_arg,
+ SQLITE_UTF8,
+ (void*)aggregate_class,
+ 0,
+ &_pysqlite_step_callback,
+ &_pysqlite_final_callback,
+ &_destructor); // will decref func
if (rc != SQLITE_OK) {
/* Workaround for SQLite bug: no error code or string is available here */
PyErr_SetString(pysqlite_OperationalError, "Error creating aggregate");
return NULL;
}
- if (PyDict_SetItem(self->function_pinboard, authorizer_cb, Py_None) == -1) {
- return NULL;
- }
rc = sqlite3_set_authorizer(self->db, _authorizer_callback, (void*)authorizer_cb);
if (rc != SQLITE_OK) {
PyErr_SetString(pysqlite_OperationalError, "Error setting authorizer callback");
+ Py_XSETREF(self->function_pinboard_authorizer_cb, NULL);
return NULL;
+ } else {
+ Py_INCREF(authorizer_cb);
+ Py_XSETREF(self->function_pinboard_authorizer_cb, authorizer_cb);
}
Py_RETURN_NONE;
}
if (progress_handler == Py_None) {
/* None clears the progress handler previously set */
sqlite3_progress_handler(self->db, 0, 0, (void*)0);
+ Py_XSETREF(self->function_pinboard_progress_handler, NULL);
} else {
- if (PyDict_SetItem(self->function_pinboard, progress_handler, Py_None) == -1)
- return NULL;
sqlite3_progress_handler(self->db, n, _progress_handler, progress_handler);
+ Py_INCREF(progress_handler);
+ Py_XSETREF(self->function_pinboard_progress_handler, progress_handler);
}
-
Py_RETURN_NONE;
}
if (trace_callback == Py_None) {
/* None clears the trace callback previously set */
sqlite3_trace(self->db, 0, (void*)0);
+ Py_XSETREF(self->function_pinboard_trace_callback, NULL);
} else {
- if (PyDict_SetItem(self->function_pinboard, trace_callback, Py_None) == -1)
- return NULL;
sqlite3_trace(self->db, _trace_callback, trace_callback);
+ Py_INCREF(trace_callback);
+ Py_XSETREF(self->function_pinboard_trace_callback, trace_callback);
}
Py_RETURN_NONE;
projects / python / commitdiff