diff options
Diffstat (limited to 'Lib/test/test_sqlite3/test_dbapi.py')
| -rw-r--r-- | Lib/test/test_sqlite3/test_dbapi.py | 1161 |
1 files changed, 1161 insertions, 0 deletions
diff --git a/Lib/test/test_sqlite3/test_dbapi.py b/Lib/test/test_sqlite3/test_dbapi.py new file mode 100644 index 0000000..d825436 --- /dev/null +++ b/Lib/test/test_sqlite3/test_dbapi.py @@ -0,0 +1,1161 @@ +# pysqlite2/test/dbapi.py: tests for DB-API compliance +# +# Copyright (C) 2004-2010 Gerhard Häring <gh@ghaering.de> +# +# This file is part of pysqlite. +# +# This software is provided 'as-is', without any express or implied +# warranty. In no event will the authors be held liable for any damages +# arising from the use of this software. +# +# Permission is granted to anyone to use this software for any purpose, +# including commercial applications, and to alter it and redistribute it +# freely, subject to the following restrictions: +# +# 1. The origin of this software must not be misrepresented; you must not +# claim that you wrote the original software. If you use this software +# in a product, an acknowledgment in the product documentation would be +# appreciated but is not required. +# 2. Altered source versions must be plainly marked as such, and must not be +# misrepresented as being the original software. +# 3. This notice may not be removed or altered from any source distribution. + +import contextlib +import sqlite3 as sqlite +import subprocess +import sys +import threading +import unittest + +from test.support import ( + SHORT_TIMEOUT, + bigmemtest, + check_disallow_instantiation, + threading_helper, +) +from test.support.os_helper import TESTFN, unlink, temp_dir + + +# Helper for tests using TESTFN +@contextlib.contextmanager +def managed_connect(*args, in_mem=False, **kwargs): + cx = sqlite.connect(*args, **kwargs) + try: + yield cx + finally: + cx.close() + if not in_mem: + unlink(TESTFN) + + +class ModuleTests(unittest.TestCase): + def test_api_level(self): + self.assertEqual(sqlite.apilevel, "2.0", + "apilevel is %s, should be 2.0" % sqlite.apilevel) + + def test_thread_safety(self): + self.assertEqual(sqlite.threadsafety, 1, + "threadsafety is %d, should be 1" % sqlite.threadsafety) + + def test_param_style(self): + self.assertEqual(sqlite.paramstyle, "qmark", + "paramstyle is '%s', should be 'qmark'" % + sqlite.paramstyle) + + def test_warning(self): + self.assertTrue(issubclass(sqlite.Warning, Exception), + "Warning is not a subclass of Exception") + + def test_error(self): + self.assertTrue(issubclass(sqlite.Error, Exception), + "Error is not a subclass of Exception") + + def test_interface_error(self): + self.assertTrue(issubclass(sqlite.InterfaceError, sqlite.Error), + "InterfaceError is not a subclass of Error") + + def test_database_error(self): + self.assertTrue(issubclass(sqlite.DatabaseError, sqlite.Error), + "DatabaseError is not a subclass of Error") + + def test_data_error(self): + self.assertTrue(issubclass(sqlite.DataError, sqlite.DatabaseError), + "DataError is not a subclass of DatabaseError") + + def test_operational_error(self): + self.assertTrue(issubclass(sqlite.OperationalError, sqlite.DatabaseError), + "OperationalError is not a subclass of DatabaseError") + + def test_integrity_error(self): + self.assertTrue(issubclass(sqlite.IntegrityError, sqlite.DatabaseError), + "IntegrityError is not a subclass of DatabaseError") + + def test_internal_error(self): + self.assertTrue(issubclass(sqlite.InternalError, sqlite.DatabaseError), + "InternalError is not a subclass of DatabaseError") + + def test_programming_error(self): + self.assertTrue(issubclass(sqlite.ProgrammingError, sqlite.DatabaseError), + "ProgrammingError is not a subclass of DatabaseError") + + def test_not_supported_error(self): + self.assertTrue(issubclass(sqlite.NotSupportedError, + sqlite.DatabaseError), + "NotSupportedError is not a subclass of DatabaseError") + + def test_module_constants(self): + consts = [ + "SQLITE_ABORT", + "SQLITE_ALTER_TABLE", + "SQLITE_ANALYZE", + "SQLITE_ATTACH", + "SQLITE_AUTH", + "SQLITE_BUSY", + "SQLITE_CANTOPEN", + "SQLITE_CONSTRAINT", + "SQLITE_CORRUPT", + "SQLITE_CREATE_INDEX", + "SQLITE_CREATE_TABLE", + "SQLITE_CREATE_TEMP_INDEX", + "SQLITE_CREATE_TEMP_TABLE", + "SQLITE_CREATE_TEMP_TRIGGER", + "SQLITE_CREATE_TEMP_VIEW", + "SQLITE_CREATE_TRIGGER", + "SQLITE_CREATE_VIEW", + "SQLITE_CREATE_VTABLE", + "SQLITE_DELETE", + "SQLITE_DENY", + "SQLITE_DETACH", + "SQLITE_DONE", + "SQLITE_DROP_INDEX", + "SQLITE_DROP_TABLE", + "SQLITE_DROP_TEMP_INDEX", + "SQLITE_DROP_TEMP_TABLE", + "SQLITE_DROP_TEMP_TRIGGER", + "SQLITE_DROP_TEMP_VIEW", + "SQLITE_DROP_TRIGGER", + "SQLITE_DROP_VIEW", + "SQLITE_DROP_VTABLE", + "SQLITE_EMPTY", + "SQLITE_ERROR", + "SQLITE_FORMAT", + "SQLITE_FULL", + "SQLITE_FUNCTION", + "SQLITE_IGNORE", + "SQLITE_INSERT", + "SQLITE_INTERNAL", + "SQLITE_INTERRUPT", + "SQLITE_IOERR", + "SQLITE_LOCKED", + "SQLITE_MISMATCH", + "SQLITE_MISUSE", + "SQLITE_NOLFS", + "SQLITE_NOMEM", + "SQLITE_NOTADB", + "SQLITE_NOTFOUND", + "SQLITE_OK", + "SQLITE_PERM", + "SQLITE_PRAGMA", + "SQLITE_PROTOCOL", + "SQLITE_READ", + "SQLITE_READONLY", + "SQLITE_REINDEX", + "SQLITE_ROW", + "SQLITE_SAVEPOINT", + "SQLITE_SCHEMA", + "SQLITE_SELECT", + "SQLITE_TOOBIG", + "SQLITE_TRANSACTION", + "SQLITE_UPDATE", + ] + if sqlite.sqlite_version_info >= (3, 7, 17): + consts += ["SQLITE_NOTICE", "SQLITE_WARNING"] + if sqlite.sqlite_version_info >= (3, 8, 3): + consts.append("SQLITE_RECURSIVE") + consts += ["PARSE_DECLTYPES", "PARSE_COLNAMES"] + for const in consts: + with self.subTest(const=const): + self.assertTrue(hasattr(sqlite, const)) + + def test_error_code_on_exception(self): + err_msg = "unable to open database file" + with temp_dir() as db: + with self.assertRaisesRegex(sqlite.Error, err_msg) as cm: + sqlite.connect(db) + e = cm.exception + self.assertEqual(e.sqlite_errorcode, sqlite.SQLITE_CANTOPEN) + self.assertEqual(e.sqlite_errorname, "SQLITE_CANTOPEN") + + # sqlite3_enable_shared_cache() is deprecated on macOS and calling it may raise + # OperationalError on some buildbots. + @unittest.skipIf(sys.platform == "darwin", "shared cache is deprecated on macOS") + def test_shared_cache_deprecated(self): + for enable in (True, False): + with self.assertWarns(DeprecationWarning) as cm: + sqlite.enable_shared_cache(enable) + self.assertIn("dbapi.py", cm.filename) + + def test_disallow_instantiation(self): + cx = sqlite.connect(":memory:") + check_disallow_instantiation(self, type(cx("select 1"))) + + def test_complete_statement(self): + self.assertFalse(sqlite.complete_statement("select t")) + self.assertTrue(sqlite.complete_statement("create table t(t);")) + + +class ConnectionTests(unittest.TestCase): + + def setUp(self): + self.cx = sqlite.connect(":memory:") + cu = self.cx.cursor() + cu.execute("create table test(id integer primary key, name text)") + cu.execute("insert into test(name) values (?)", ("foo",)) + + def tearDown(self): + self.cx.close() + + def test_commit(self): + self.cx.commit() + + def test_commit_after_no_changes(self): + """ + A commit should also work when no changes were made to the database. + """ + self.cx.commit() + self.cx.commit() + + def test_rollback(self): + self.cx.rollback() + + def test_rollback_after_no_changes(self): + """ + A rollback should also work when no changes were made to the database. + """ + self.cx.rollback() + self.cx.rollback() + + def test_cursor(self): + cu = self.cx.cursor() + + def test_failed_open(self): + YOU_CANNOT_OPEN_THIS = "/foo/bar/bla/23534/mydb.db" + with self.assertRaises(sqlite.OperationalError): + con = sqlite.connect(YOU_CANNOT_OPEN_THIS) + + def test_close(self): + self.cx.close() + + def test_use_after_close(self): + sql = "select 1" + cu = self.cx.cursor() + res = cu.execute(sql) + self.cx.close() + self.assertRaises(sqlite.ProgrammingError, res.fetchall) + self.assertRaises(sqlite.ProgrammingError, cu.execute, sql) + self.assertRaises(sqlite.ProgrammingError, cu.executemany, sql, []) + self.assertRaises(sqlite.ProgrammingError, cu.executescript, sql) + self.assertRaises(sqlite.ProgrammingError, self.cx.execute, sql) + self.assertRaises(sqlite.ProgrammingError, + self.cx.executemany, sql, []) + self.assertRaises(sqlite.ProgrammingError, self.cx.executescript, sql) + self.assertRaises(sqlite.ProgrammingError, + self.cx.create_function, "t", 1, lambda x: x) + self.assertRaises(sqlite.ProgrammingError, self.cx.cursor) + with self.assertRaises(sqlite.ProgrammingError): + with self.cx: + pass + + def test_exceptions(self): + # Optional DB-API extension. + self.assertEqual(self.cx.Warning, sqlite.Warning) + self.assertEqual(self.cx.Error, sqlite.Error) + self.assertEqual(self.cx.InterfaceError, sqlite.InterfaceError) + self.assertEqual(self.cx.DatabaseError, sqlite.DatabaseError) + self.assertEqual(self.cx.DataError, sqlite.DataError) + self.assertEqual(self.cx.OperationalError, sqlite.OperationalError) + self.assertEqual(self.cx.IntegrityError, sqlite.IntegrityError) + self.assertEqual(self.cx.InternalError, sqlite.InternalError) + self.assertEqual(self.cx.ProgrammingError, sqlite.ProgrammingError) + self.assertEqual(self.cx.NotSupportedError, sqlite.NotSupportedError) + + def test_in_transaction(self): + # Can't use db from setUp because we want to test initial state. + cx = sqlite.connect(":memory:") + cu = cx.cursor() + self.assertEqual(cx.in_transaction, False) + cu.execute("create table transactiontest(id integer primary key, name text)") + self.assertEqual(cx.in_transaction, False) + cu.execute("insert into transactiontest(name) values (?)", ("foo",)) + self.assertEqual(cx.in_transaction, True) + cu.execute("select name from transactiontest where name=?", ["foo"]) + row = cu.fetchone() + self.assertEqual(cx.in_transaction, True) + cx.commit() + self.assertEqual(cx.in_transaction, False) + cu.execute("select name from transactiontest where name=?", ["foo"]) + row = cu.fetchone() + self.assertEqual(cx.in_transaction, False) + + def test_in_transaction_ro(self): + with self.assertRaises(AttributeError): + self.cx.in_transaction = True + + def test_connection_exceptions(self): + exceptions = [ + "DataError", + "DatabaseError", + "Error", + "IntegrityError", + "InterfaceError", + "NotSupportedError", + "OperationalError", + "ProgrammingError", + "Warning", + ] + for exc in exceptions: + with self.subTest(exc=exc): + self.assertTrue(hasattr(self.cx, exc)) + self.assertIs(getattr(sqlite, exc), getattr(self.cx, exc)) + + def test_interrupt_on_closed_db(self): + cx = sqlite.connect(":memory:") + cx.close() + with self.assertRaises(sqlite.ProgrammingError): + cx.interrupt() + + def test_interrupt(self): + self.assertIsNone(self.cx.interrupt()) + + def test_drop_unused_refs(self): + for n in range(500): + cu = self.cx.execute(f"select {n}") + self.assertEqual(cu.fetchone()[0], n) + + +class UninitialisedConnectionTests(unittest.TestCase): + def setUp(self): + self.cx = sqlite.Connection.__new__(sqlite.Connection) + + def test_uninit_operations(self): + funcs = ( + lambda: self.cx.isolation_level, + lambda: self.cx.total_changes, + lambda: self.cx.in_transaction, + lambda: self.cx.iterdump(), + lambda: self.cx.cursor(), + lambda: self.cx.close(), + ) + for func in funcs: + with self.subTest(func=func): + self.assertRaisesRegex(sqlite.ProgrammingError, + "Base Connection.__init__ not called", + func) + + +class OpenTests(unittest.TestCase): + _sql = "create table test(id integer)" + + def test_open_with_path_like_object(self): + """ Checks that we can successfully connect to a database using an object that + is PathLike, i.e. has __fspath__(). """ + class Path: + def __fspath__(self): + return TESTFN + path = Path() + with managed_connect(path) as cx: + cx.execute(self._sql) + + def test_open_uri(self): + with managed_connect(TESTFN) as cx: + cx.execute(self._sql) + with managed_connect(f"file:{TESTFN}", uri=True) as cx: + cx.execute(self._sql) + with self.assertRaises(sqlite.OperationalError): + with managed_connect(f"file:{TESTFN}?mode=ro", uri=True) as cx: + cx.execute(self._sql) + + def test_database_keyword(self): + with sqlite.connect(database=":memory:") as cx: + self.assertEqual(type(cx), sqlite.Connection) + + +class CursorTests(unittest.TestCase): + def setUp(self): + self.cx = sqlite.connect(":memory:") + self.cu = self.cx.cursor() + self.cu.execute( + "create table test(id integer primary key, name text, " + "income number, unique_test text unique)" + ) + self.cu.execute("insert into test(name) values (?)", ("foo",)) + + def tearDown(self): + self.cu.close() + self.cx.close() + + def test_execute_no_args(self): + self.cu.execute("delete from test") + + def test_execute_illegal_sql(self): + with self.assertRaises(sqlite.OperationalError): + self.cu.execute("select asdf") + + def test_execute_too_much_sql(self): + with self.assertRaises(sqlite.Warning): + self.cu.execute("select 5+4; select 4+5") + + def test_execute_too_much_sql2(self): + self.cu.execute("select 5+4; -- foo bar") + + def test_execute_too_much_sql3(self): + self.cu.execute(""" + select 5+4; + + /* + foo + */ + """) + + def test_execute_wrong_sql_arg(self): + with self.assertRaises(TypeError): + self.cu.execute(42) + + def test_execute_arg_int(self): + self.cu.execute("insert into test(id) values (?)", (42,)) + + def test_execute_arg_float(self): + self.cu.execute("insert into test(income) values (?)", (2500.32,)) + + def test_execute_arg_string(self): + self.cu.execute("insert into test(name) values (?)", ("Hugo",)) + + def test_execute_arg_string_with_zero_byte(self): + self.cu.execute("insert into test(name) values (?)", ("Hu\x00go",)) + + self.cu.execute("select name from test where id=?", (self.cu.lastrowid,)) + row = self.cu.fetchone() + self.assertEqual(row[0], "Hu\x00go") + + def test_execute_non_iterable(self): + with self.assertRaises(ValueError) as cm: + self.cu.execute("insert into test(id) values (?)", 42) + self.assertEqual(str(cm.exception), 'parameters are of unsupported type') + + def test_execute_wrong_no_of_args1(self): + # too many parameters + with self.assertRaises(sqlite.ProgrammingError): + self.cu.execute("insert into test(id) values (?)", (17, "Egon")) + + def test_execute_wrong_no_of_args2(self): + # too little parameters + with self.assertRaises(sqlite.ProgrammingError): + self.cu.execute("insert into test(id) values (?)") + + def test_execute_wrong_no_of_args3(self): + # no parameters, parameters are needed + with self.assertRaises(sqlite.ProgrammingError): + self.cu.execute("insert into test(id) values (?)") + + def test_execute_param_list(self): + self.cu.execute("insert into test(name) values ('foo')") + self.cu.execute("select name from test where name=?", ["foo"]) + row = self.cu.fetchone() + self.assertEqual(row[0], "foo") + + def test_execute_param_sequence(self): + class L: + def __len__(self): + return 1 + def __getitem__(self, x): + assert x == 0 + return "foo" + + self.cu.execute("insert into test(name) values ('foo')") + self.cu.execute("select name from test where name=?", L()) + row = self.cu.fetchone() + self.assertEqual(row[0], "foo") + + def test_execute_param_sequence_bad_len(self): + # Issue41662: Error in __len__() was overridden with ProgrammingError. + class L: + def __len__(self): + 1/0 + def __getitem__(slf, x): + raise AssertionError + + self.cu.execute("insert into test(name) values ('foo')") + with self.assertRaises(ZeroDivisionError): + self.cu.execute("select name from test where name=?", L()) + + def test_execute_dict_mapping(self): + self.cu.execute("insert into test(name) values ('foo')") + self.cu.execute("select name from test where name=:name", {"name": "foo"}) + row = self.cu.fetchone() + self.assertEqual(row[0], "foo") + + def test_execute_dict_mapping_mapping(self): + class D(dict): + def __missing__(self, key): + return "foo" + + self.cu.execute("insert into test(name) values ('foo')") + self.cu.execute("select name from test where name=:name", D()) + row = self.cu.fetchone() + self.assertEqual(row[0], "foo") + + def test_execute_dict_mapping_too_little_args(self): + self.cu.execute("insert into test(name) values ('foo')") + with self.assertRaises(sqlite.ProgrammingError): + self.cu.execute("select name from test where name=:name and id=:id", {"name": "foo"}) + + def test_execute_dict_mapping_no_args(self): + self.cu.execute("insert into test(name) values ('foo')") + with self.assertRaises(sqlite.ProgrammingError): + self.cu.execute("select name from test where name=:name") + + def test_execute_dict_mapping_unnamed(self): + self.cu.execute("insert into test(name) values ('foo')") + with self.assertRaises(sqlite.ProgrammingError): + self.cu.execute("select name from test where name=?", {"name": "foo"}) + + def test_close(self): + self.cu.close() + + def test_rowcount_execute(self): + self.cu.execute("delete from test") + self.cu.execute("insert into test(name) values ('foo')") + self.cu.execute("insert into test(name) values ('foo')") + self.cu.execute("update test set name='bar'") + self.assertEqual(self.cu.rowcount, 2) + + def test_rowcount_select(self): + """ + pysqlite does not know the rowcount of SELECT statements, because we + don't fetch all rows after executing the select statement. The rowcount + has thus to be -1. + """ + self.cu.execute("select 5 union select 6") + self.assertEqual(self.cu.rowcount, -1) + + def test_rowcount_executemany(self): + self.cu.execute("delete from test") + self.cu.executemany("insert into test(name) values (?)", [(1,), (2,), (3,)]) + self.assertEqual(self.cu.rowcount, 3) + + def test_total_changes(self): + self.cu.execute("insert into test(name) values ('foo')") + self.cu.execute("insert into test(name) values ('foo')") + self.assertLess(2, self.cx.total_changes, msg='total changes reported wrong value') + + # Checks for executemany: + # Sequences are required by the DB-API, iterators + # enhancements in pysqlite. + + def test_execute_many_sequence(self): + self.cu.executemany("insert into test(income) values (?)", [(x,) for x in range(100, 110)]) + + def test_execute_many_iterator(self): + class MyIter: + def __init__(self): + self.value = 5 + + def __iter__(self): + return self + + def __next__(self): + if self.value == 10: + raise StopIteration + else: + self.value += 1 + return (self.value,) + + self.cu.executemany("insert into test(income) values (?)", MyIter()) + + def test_execute_many_generator(self): + def mygen(): + for i in range(5): + yield (i,) + + self.cu.executemany("insert into test(income) values (?)", mygen()) + + def test_execute_many_wrong_sql_arg(self): + with self.assertRaises(TypeError): + self.cu.executemany(42, [(3,)]) + + def test_execute_many_select(self): + with self.assertRaises(sqlite.ProgrammingError): + self.cu.executemany("select ?", [(3,)]) + + def test_execute_many_not_iterable(self): + with self.assertRaises(TypeError): + self.cu.executemany("insert into test(income) values (?)", 42) + + def test_fetch_iter(self): + # Optional DB-API extension. + self.cu.execute("delete from test") + self.cu.execute("insert into test(id) values (?)", (5,)) + self.cu.execute("insert into test(id) values (?)", (6,)) + self.cu.execute("select id from test order by id") + lst = [] + for row in self.cu: + lst.append(row[0]) + self.assertEqual(lst[0], 5) + self.assertEqual(lst[1], 6) + + def test_fetchone(self): + self.cu.execute("select name from test") + row = self.cu.fetchone() + self.assertEqual(row[0], "foo") + row = self.cu.fetchone() + self.assertEqual(row, None) + + def test_fetchone_no_statement(self): + cur = self.cx.cursor() + row = cur.fetchone() + self.assertEqual(row, None) + + def test_array_size(self): + # must default to 1 + self.assertEqual(self.cu.arraysize, 1) + + # now set to 2 + self.cu.arraysize = 2 + + # now make the query return 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')") + self.cu.execute("insert into test(name) values ('C')") + self.cu.execute("select name from test") + res = self.cu.fetchmany() + + self.assertEqual(len(res), 2) + + def test_fetchmany(self): + 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, []) + + def test_fetchmany_kw_arg(self): + """Checks if fetchmany works with keyword arguments""" + self.cu.execute("select name from test") + res = self.cu.fetchmany(size=100) + self.assertEqual(len(res), 1) + + def test_fetchall(self): + self.cu.execute("select name from test") + res = self.cu.fetchall() + self.assertEqual(len(res), 1) + res = self.cu.fetchall() + self.assertEqual(res, []) + + def test_setinputsizes(self): + self.cu.setinputsizes([3, 4, 5]) + + def test_setoutputsize(self): + self.cu.setoutputsize(5, 0) + + def test_setoutputsize_no_column(self): + self.cu.setoutputsize(42) + + def test_cursor_connection(self): + # Optional DB-API extension. + self.assertEqual(self.cu.connection, self.cx) + + def test_wrong_cursor_callable(self): + with self.assertRaises(TypeError): + def f(): pass + cur = self.cx.cursor(f) + + def test_cursor_wrong_class(self): + class Foo: pass + foo = Foo() + with self.assertRaises(TypeError): + cur = sqlite.Cursor(foo) + + def test_last_row_id_on_replace(self): + """ + INSERT OR REPLACE and REPLACE INTO should produce the same behavior. + """ + sql = '{} INTO test(id, unique_test) VALUES (?, ?)' + for statement in ('INSERT OR REPLACE', 'REPLACE'): + with self.subTest(statement=statement): + self.cu.execute(sql.format(statement), (1, 'foo')) + self.assertEqual(self.cu.lastrowid, 1) + + def test_last_row_id_on_ignore(self): + self.cu.execute( + "insert or ignore into test(unique_test) values (?)", + ('test',)) + self.assertEqual(self.cu.lastrowid, 2) + self.cu.execute( + "insert or ignore into test(unique_test) values (?)", + ('test',)) + self.assertEqual(self.cu.lastrowid, 2) + + def test_last_row_id_insert_o_r(self): + results = [] + for statement in ('FAIL', 'ABORT', 'ROLLBACK'): + sql = 'INSERT OR {} INTO test(unique_test) VALUES (?)' + with self.subTest(statement='INSERT OR {}'.format(statement)): + self.cu.execute(sql.format(statement), (statement,)) + results.append((statement, self.cu.lastrowid)) + with self.assertRaises(sqlite.IntegrityError): + self.cu.execute(sql.format(statement), (statement,)) + results.append((statement, self.cu.lastrowid)) + expected = [ + ('FAIL', 2), ('FAIL', 2), + ('ABORT', 3), ('ABORT', 3), + ('ROLLBACK', 4), ('ROLLBACK', 4), + ] + self.assertEqual(results, expected) + + def test_column_count(self): + # Check that column count is updated correctly for cached statements + select = "select * from test" + res = self.cu.execute(select) + old_count = len(res.description) + # Add a new column and execute the cached select query again + self.cu.execute("alter table test add newcol") + res = self.cu.execute(select) + new_count = len(res.description) + self.assertEqual(new_count - old_count, 1) + + def test_same_query_in_multiple_cursors(self): + cursors = [self.cx.execute("select 1") for _ in range(3)] + for cu in cursors: + self.assertEqual(cu.fetchall(), [(1,)]) + + +class ThreadTests(unittest.TestCase): + def setUp(self): + self.con = sqlite.connect(":memory:") + self.cur = self.con.cursor() + self.cur.execute("create table test(name text)") + + def tearDown(self): + self.cur.close() + self.con.close() + + @threading_helper.reap_threads + def _run_test(self, fn, *args, **kwds): + def run(err): + try: + fn(*args, **kwds) + err.append("did not raise ProgrammingError") + except sqlite.ProgrammingError: + pass + except: + err.append("raised wrong exception") + + err = [] + t = threading.Thread(target=run, kwargs={"err": err}) + t.start() + t.join() + if err: + self.fail("\n".join(err)) + + def test_check_connection_thread(self): + fns = [ + lambda: self.con.cursor(), + lambda: self.con.commit(), + lambda: self.con.rollback(), + lambda: self.con.close(), + lambda: self.con.set_trace_callback(None), + lambda: self.con.set_authorizer(None), + lambda: self.con.create_collation("foo", None), + ] + for fn in fns: + with self.subTest(fn=fn): + self._run_test(fn) + + def test_check_cursor_thread(self): + fns = [ + lambda: self.cur.execute("insert into test(name) values('a')"), + lambda: self.cur.close(), + lambda: self.cur.execute("select name from test"), + lambda: self.cur.fetchone(), + ] + for fn in fns: + with self.subTest(fn=fn): + self._run_test(fn) + + + @threading_helper.reap_threads + def test_dont_check_same_thread(self): + def run(con, err): + try: + con.execute("select 1") + except sqlite.Error: + err.append("multi-threading not allowed") + + con = sqlite.connect(":memory:", check_same_thread=False) + err = [] + t = threading.Thread(target=run, kwargs={"con": con, "err": err}) + t.start() + t.join() + self.assertEqual(len(err), 0, "\n".join(err)) + + +class ConstructorTests(unittest.TestCase): + def test_date(self): + d = sqlite.Date(2004, 10, 28) + + def test_time(self): + t = sqlite.Time(12, 39, 35) + + def test_timestamp(self): + ts = sqlite.Timestamp(2004, 10, 28, 12, 39, 35) + + def test_date_from_ticks(self): + d = sqlite.DateFromTicks(42) + + def test_time_from_ticks(self): + t = sqlite.TimeFromTicks(42) + + def test_timestamp_from_ticks(self): + ts = sqlite.TimestampFromTicks(42) + + def test_binary(self): + b = sqlite.Binary(b"\0'") + +class ExtensionTests(unittest.TestCase): + def test_script_string_sql(self): + con = sqlite.connect(":memory:") + cur = con.cursor() + cur.executescript(""" + -- bla bla + /* a stupid comment */ + create table a(i); + insert into a(i) values (5); + """) + cur.execute("select i from a") + res = cur.fetchone()[0] + self.assertEqual(res, 5) + + def test_script_syntax_error(self): + con = sqlite.connect(":memory:") + cur = con.cursor() + with self.assertRaises(sqlite.OperationalError): + cur.executescript("create table test(x); asdf; create table test2(x)") + + def test_script_error_normal(self): + con = sqlite.connect(":memory:") + cur = con.cursor() + with self.assertRaises(sqlite.OperationalError): + cur.executescript("create table test(sadfsadfdsa); select foo from hurz;") + + def test_cursor_executescript_as_bytes(self): + con = sqlite.connect(":memory:") + cur = con.cursor() + with self.assertRaises(TypeError): + cur.executescript(b"create table test(foo); insert into test(foo) values (5);") + + def test_cursor_executescript_with_null_characters(self): + con = sqlite.connect(":memory:") + cur = con.cursor() + with self.assertRaises(ValueError): + cur.executescript(""" + create table a(i);\0 + insert into a(i) values (5); + """) + + def test_cursor_executescript_with_surrogates(self): + con = sqlite.connect(":memory:") + cur = con.cursor() + with self.assertRaises(UnicodeEncodeError): + cur.executescript(""" + create table a(s); + insert into a(s) values ('\ud8ff'); + """) + + @unittest.skipUnless(sys.maxsize > 2**32, 'requires 64bit platform') + @bigmemtest(size=2**31, memuse=3, dry_run=False) + def test_cursor_executescript_too_large_script(self, maxsize): + con = sqlite.connect(":memory:") + cur = con.cursor() + for size in 2**31-1, 2**31: + with self.assertRaises(sqlite.DataError): + cur.executescript("create table a(s);".ljust(size)) + + def test_cursor_executescript_tx_control(self): + con = sqlite.connect(":memory:") + con.execute("begin") + self.assertTrue(con.in_transaction) + con.executescript("select 1") + self.assertFalse(con.in_transaction) + + def test_connection_execute(self): + con = sqlite.connect(":memory:") + result = con.execute("select 5").fetchone()[0] + self.assertEqual(result, 5, "Basic test of Connection.execute") + + def test_connection_executemany(self): + con = sqlite.connect(":memory:") + con.execute("create table test(foo)") + con.executemany("insert into test(foo) values (?)", [(3,), (4,)]) + result = con.execute("select foo from test order by foo").fetchall() + self.assertEqual(result[0][0], 3, "Basic test of Connection.executemany") + self.assertEqual(result[1][0], 4, "Basic test of Connection.executemany") + + def test_connection_executescript(self): + con = sqlite.connect(":memory:") + con.executescript("create table test(foo); insert into test(foo) values (5);") + result = con.execute("select foo from test").fetchone()[0] + self.assertEqual(result, 5, "Basic test of Connection.executescript") + +class ClosedConTests(unittest.TestCase): + def test_closed_con_cursor(self): + con = sqlite.connect(":memory:") + con.close() + with self.assertRaises(sqlite.ProgrammingError): + cur = con.cursor() + + def test_closed_con_commit(self): + con = sqlite.connect(":memory:") + con.close() + with self.assertRaises(sqlite.ProgrammingError): + con.commit() + + def test_closed_con_rollback(self): + con = sqlite.connect(":memory:") + con.close() + with self.assertRaises(sqlite.ProgrammingError): + con.rollback() + + def test_closed_cur_execute(self): + con = sqlite.connect(":memory:") + cur = con.cursor() + con.close() + with self.assertRaises(sqlite.ProgrammingError): + cur.execute("select 4") + + def test_closed_create_function(self): + con = sqlite.connect(":memory:") + con.close() + def f(x): return 17 + with self.assertRaises(sqlite.ProgrammingError): + con.create_function("foo", 1, f) + + def test_closed_create_aggregate(self): + con = sqlite.connect(":memory:") + con.close() + class Agg: + def __init__(self): + pass + def step(self, x): + pass + def finalize(self): + return 17 + with self.assertRaises(sqlite.ProgrammingError): + con.create_aggregate("foo", 1, Agg) + + def test_closed_set_authorizer(self): + con = sqlite.connect(":memory:") + con.close() + def authorizer(*args): + return sqlite.DENY + with self.assertRaises(sqlite.ProgrammingError): + con.set_authorizer(authorizer) + + def test_closed_set_progress_callback(self): + con = sqlite.connect(":memory:") + con.close() + def progress(): pass + with self.assertRaises(sqlite.ProgrammingError): + con.set_progress_handler(progress, 100) + + def test_closed_call(self): + con = sqlite.connect(":memory:") + con.close() + with self.assertRaises(sqlite.ProgrammingError): + con() + +class ClosedCurTests(unittest.TestCase): + def test_closed(self): + con = sqlite.connect(":memory:") + cur = con.cursor() + cur.close() + + for method_name in ("execute", "executemany", "executescript", "fetchall", "fetchmany", "fetchone"): + if method_name in ("execute", "executescript"): + params = ("select 4 union select 5",) + elif method_name == "executemany": + params = ("insert into foo(bar) values (?)", [(3,), (4,)]) + else: + params = [] + + with self.assertRaises(sqlite.ProgrammingError): + method = getattr(cur, method_name) + method(*params) + + +class SqliteOnConflictTests(unittest.TestCase): + """ + Tests for SQLite's "insert on conflict" feature. + + See https://www.sqlite.org/lang_conflict.html for details. + """ + + def setUp(self): + self.cx = sqlite.connect(":memory:") + self.cu = self.cx.cursor() + self.cu.execute(""" + CREATE TABLE test( + id INTEGER PRIMARY KEY, name TEXT, unique_name TEXT UNIQUE + ); + """) + + def tearDown(self): + self.cu.close() + self.cx.close() + + def test_on_conflict_rollback_with_explicit_transaction(self): + self.cx.isolation_level = None # autocommit mode + self.cu = self.cx.cursor() + # Start an explicit transaction. + self.cu.execute("BEGIN") + self.cu.execute("INSERT INTO test(name) VALUES ('abort_test')") + self.cu.execute("INSERT OR ROLLBACK INTO test(unique_name) VALUES ('foo')") + with self.assertRaises(sqlite.IntegrityError): + self.cu.execute("INSERT OR ROLLBACK INTO test(unique_name) VALUES ('foo')") + # Use connection to commit. + self.cx.commit() + self.cu.execute("SELECT name, unique_name from test") + # Transaction should have rolled back and nothing should be in table. + self.assertEqual(self.cu.fetchall(), []) + + def test_on_conflict_abort_raises_with_explicit_transactions(self): + # Abort cancels the current sql statement but doesn't change anything + # about the current transaction. + self.cx.isolation_level = None # autocommit mode + self.cu = self.cx.cursor() + # Start an explicit transaction. + self.cu.execute("BEGIN") + self.cu.execute("INSERT INTO test(name) VALUES ('abort_test')") + self.cu.execute("INSERT OR ABORT INTO test(unique_name) VALUES ('foo')") + with self.assertRaises(sqlite.IntegrityError): + self.cu.execute("INSERT OR ABORT INTO test(unique_name) VALUES ('foo')") + self.cx.commit() + self.cu.execute("SELECT name, unique_name FROM test") + # Expect the first two inserts to work, third to do nothing. + self.assertEqual(self.cu.fetchall(), [('abort_test', None), (None, 'foo',)]) + + def test_on_conflict_rollback_without_transaction(self): + # Start of implicit transaction + self.cu.execute("INSERT INTO test(name) VALUES ('abort_test')") + self.cu.execute("INSERT OR ROLLBACK INTO test(unique_name) VALUES ('foo')") + with self.assertRaises(sqlite.IntegrityError): + self.cu.execute("INSERT OR ROLLBACK INTO test(unique_name) VALUES ('foo')") + self.cu.execute("SELECT name, unique_name FROM test") + # Implicit transaction is rolled back on error. + self.assertEqual(self.cu.fetchall(), []) + + def test_on_conflict_abort_raises_without_transactions(self): + # Abort cancels the current sql statement but doesn't change anything + # about the current transaction. + self.cu.execute("INSERT INTO test(name) VALUES ('abort_test')") + self.cu.execute("INSERT OR ABORT INTO test(unique_name) VALUES ('foo')") + with self.assertRaises(sqlite.IntegrityError): + self.cu.execute("INSERT OR ABORT INTO test(unique_name) VALUES ('foo')") + # Make sure all other values were inserted. + self.cu.execute("SELECT name, unique_name FROM test") + self.assertEqual(self.cu.fetchall(), [('abort_test', None), (None, 'foo',)]) + + def test_on_conflict_fail(self): + self.cu.execute("INSERT OR FAIL INTO test(unique_name) VALUES ('foo')") + with self.assertRaises(sqlite.IntegrityError): + self.cu.execute("INSERT OR FAIL INTO test(unique_name) VALUES ('foo')") + self.assertEqual(self.cu.fetchall(), []) + + def test_on_conflict_ignore(self): + self.cu.execute("INSERT OR IGNORE INTO test(unique_name) VALUES ('foo')") + # Nothing should happen. + self.cu.execute("INSERT OR IGNORE INTO test(unique_name) VALUES ('foo')") + self.cu.execute("SELECT unique_name FROM test") + self.assertEqual(self.cu.fetchall(), [('foo',)]) + + def test_on_conflict_replace(self): + self.cu.execute("INSERT OR REPLACE INTO test(name, unique_name) VALUES ('Data!', 'foo')") + # There shouldn't be an IntegrityError exception. + self.cu.execute("INSERT OR REPLACE INTO test(name, unique_name) VALUES ('Very different data!', 'foo')") + self.cu.execute("SELECT name, unique_name FROM test") + self.assertEqual(self.cu.fetchall(), [('Very different data!', 'foo')]) + + +class MultiprocessTests(unittest.TestCase): + CONNECTION_TIMEOUT = SHORT_TIMEOUT / 1000. # Defaults to 30 ms + + def tearDown(self): + unlink(TESTFN) + + def test_ctx_mgr_rollback_if_commit_failed(self): + # bpo-27334: ctx manager does not rollback if commit fails + SCRIPT = f"""if 1: + import sqlite3 + def wait(): + print("started") + assert "database is locked" in input() + + cx = sqlite3.connect("{TESTFN}", timeout={self.CONNECTION_TIMEOUT}) + cx.create_function("wait", 0, wait) + with cx: + cx.execute("create table t(t)") + try: + # execute two transactions; both will try to lock the db + cx.executescript(''' + -- start a transaction and wait for parent + begin transaction; + select * from t; + select wait(); + rollback; + + -- start a new transaction; would fail if parent holds lock + begin transaction; + select * from t; + rollback; + ''') + finally: + cx.close() + """ + + # spawn child process + proc = subprocess.Popen( + [sys.executable, "-c", SCRIPT], + encoding="utf-8", + bufsize=0, + stdin=subprocess.PIPE, + stdout=subprocess.PIPE, + ) + self.addCleanup(proc.communicate) + + # wait for child process to start + self.assertEqual("started", proc.stdout.readline().strip()) + + cx = sqlite.connect(TESTFN, timeout=self.CONNECTION_TIMEOUT) + try: # context manager should correctly release the db lock + with cx: + cx.execute("insert into t values('test')") + except sqlite.OperationalError as exc: + proc.stdin.write(str(exc)) + else: + proc.stdin.write("no error") + finally: + cx.close() + + # terminate child process + self.assertIsNone(proc.returncode) + try: + proc.communicate(input="end", timeout=SHORT_TIMEOUT) + except subprocess.TimeoutExpired: + proc.kill() + proc.communicate() + raise + self.assertEqual(proc.returncode, 0) + + +if __name__ == "__main__": + unittest.main() |