summaryrefslogtreecommitdiffstats
path: root/Lib/sqlite3
diff options
context:
space:
mode:
authorErlend Egeberg Aasland <erlend.aasland@innova.no>2021-10-29 22:08:19 (GMT)
committerGitHub <noreply@github.com>2021-10-29 22:08:19 (GMT)
commit62bf263a775f4444d8b5d5841cc09be3bd53e933 (patch)
treed92d1e79ebb70b5b5f92ab034aa45fecd2a368e4 /Lib/sqlite3
parentc2d0ba722a7b3839685af968cf0c304a24cdf525 (diff)
downloadcpython-62bf263a775f4444d8b5d5841cc09be3bd53e933.zip
cpython-62bf263a775f4444d8b5d5841cc09be3bd53e933.tar.gz
cpython-62bf263a775f4444d8b5d5841cc09be3bd53e933.tar.bz2
bpo-10572: Move `sqlite3` tests to `Lib/test` (GH-29304)
Automerge-Triggered-By: GH:brettcannon
Diffstat (limited to 'Lib/sqlite3')
-rw-r--r--Lib/sqlite3/test/__init__.py0
-rw-r--r--Lib/sqlite3/test/test_backup.py164
-rw-r--r--Lib/sqlite3/test/test_dbapi.py1161
-rw-r--r--Lib/sqlite3/test/test_dump.py75
-rw-r--r--Lib/sqlite3/test/test_factory.py310
-rw-r--r--Lib/sqlite3/test/test_hooks.py294
-rw-r--r--Lib/sqlite3/test/test_regression.py491
-rw-r--r--Lib/sqlite3/test/test_transactions.py200
-rw-r--r--Lib/sqlite3/test/test_types.py535
-rw-r--r--Lib/sqlite3/test/test_userfunctions.py665
10 files changed, 0 insertions, 3895 deletions
diff --git a/Lib/sqlite3/test/__init__.py b/Lib/sqlite3/test/__init__.py
deleted file mode 100644
index e69de29..0000000
--- a/Lib/sqlite3/test/__init__.py
+++ /dev/null
diff --git a/Lib/sqlite3/test/test_backup.py b/Lib/sqlite3/test/test_backup.py
deleted file mode 100644
index 384dd92..0000000
--- a/Lib/sqlite3/test/test_backup.py
+++ /dev/null
@@ -1,164 +0,0 @@
-import sqlite3 as sqlite
-import unittest
-
-
-class BackupTests(unittest.TestCase):
- def setUp(self):
- cx = self.cx = sqlite.connect(":memory:")
- cx.execute('CREATE TABLE foo (key INTEGER)')
- cx.executemany('INSERT INTO foo (key) VALUES (?)', [(3,), (4,)])
- cx.commit()
-
- def tearDown(self):
- self.cx.close()
-
- def verify_backup(self, bckcx):
- result = bckcx.execute("SELECT key FROM foo ORDER BY key").fetchall()
- self.assertEqual(result[0][0], 3)
- self.assertEqual(result[1][0], 4)
-
- def test_bad_target(self):
- with self.assertRaises(TypeError):
- self.cx.backup(None)
- with self.assertRaises(TypeError):
- self.cx.backup()
-
- def test_bad_target_filename(self):
- with self.assertRaises(TypeError):
- self.cx.backup('some_file_name.db')
-
- def test_bad_target_same_connection(self):
- with self.assertRaises(ValueError):
- self.cx.backup(self.cx)
-
- def test_bad_target_closed_connection(self):
- bck = sqlite.connect(':memory:')
- bck.close()
- with self.assertRaises(sqlite.ProgrammingError):
- self.cx.backup(bck)
-
- def test_bad_source_closed_connection(self):
- bck = sqlite.connect(':memory:')
- source = sqlite.connect(":memory:")
- source.close()
- with self.assertRaises(sqlite.ProgrammingError):
- source.backup(bck)
-
- def test_bad_target_in_transaction(self):
- bck = sqlite.connect(':memory:')
- bck.execute('CREATE TABLE bar (key INTEGER)')
- bck.executemany('INSERT INTO bar (key) VALUES (?)', [(3,), (4,)])
- with self.assertRaises(sqlite.OperationalError) as cm:
- self.cx.backup(bck)
- if sqlite.sqlite_version_info < (3, 8, 8):
- self.assertEqual(str(cm.exception), 'target is in transaction')
-
- def test_keyword_only_args(self):
- with self.assertRaises(TypeError):
- with sqlite.connect(':memory:') as bck:
- self.cx.backup(bck, 1)
-
- def test_simple(self):
- with sqlite.connect(':memory:') as bck:
- self.cx.backup(bck)
- self.verify_backup(bck)
-
- def test_progress(self):
- journal = []
-
- def progress(status, remaining, total):
- journal.append(status)
-
- with sqlite.connect(':memory:') as bck:
- self.cx.backup(bck, pages=1, progress=progress)
- self.verify_backup(bck)
-
- self.assertEqual(len(journal), 2)
- self.assertEqual(journal[0], sqlite.SQLITE_OK)
- self.assertEqual(journal[1], sqlite.SQLITE_DONE)
-
- def test_progress_all_pages_at_once_1(self):
- journal = []
-
- def progress(status, remaining, total):
- journal.append(remaining)
-
- with sqlite.connect(':memory:') as bck:
- self.cx.backup(bck, progress=progress)
- self.verify_backup(bck)
-
- self.assertEqual(len(journal), 1)
- self.assertEqual(journal[0], 0)
-
- def test_progress_all_pages_at_once_2(self):
- journal = []
-
- def progress(status, remaining, total):
- journal.append(remaining)
-
- with sqlite.connect(':memory:') as bck:
- self.cx.backup(bck, pages=-1, progress=progress)
- self.verify_backup(bck)
-
- self.assertEqual(len(journal), 1)
- self.assertEqual(journal[0], 0)
-
- def test_non_callable_progress(self):
- with self.assertRaises(TypeError) as cm:
- with sqlite.connect(':memory:') as bck:
- self.cx.backup(bck, pages=1, progress='bar')
- self.assertEqual(str(cm.exception), 'progress argument must be a callable')
-
- def test_modifying_progress(self):
- journal = []
-
- def progress(status, remaining, total):
- if not journal:
- self.cx.execute('INSERT INTO foo (key) VALUES (?)', (remaining+1000,))
- self.cx.commit()
- journal.append(remaining)
-
- with sqlite.connect(':memory:') as bck:
- self.cx.backup(bck, pages=1, progress=progress)
- self.verify_backup(bck)
-
- result = bck.execute("SELECT key FROM foo"
- " WHERE key >= 1000"
- " ORDER BY key").fetchall()
- self.assertEqual(result[0][0], 1001)
-
- self.assertEqual(len(journal), 3)
- self.assertEqual(journal[0], 1)
- self.assertEqual(journal[1], 1)
- self.assertEqual(journal[2], 0)
-
- def test_failing_progress(self):
- def progress(status, remaining, total):
- raise SystemError('nearly out of space')
-
- with self.assertRaises(SystemError) as err:
- with sqlite.connect(':memory:') as bck:
- self.cx.backup(bck, progress=progress)
- self.assertEqual(str(err.exception), 'nearly out of space')
-
- def test_database_source_name(self):
- with sqlite.connect(':memory:') as bck:
- self.cx.backup(bck, name='main')
- with sqlite.connect(':memory:') as bck:
- self.cx.backup(bck, name='temp')
- with self.assertRaises(sqlite.OperationalError) as cm:
- with sqlite.connect(':memory:') as bck:
- self.cx.backup(bck, name='non-existing')
- self.assertIn("unknown database", str(cm.exception))
-
- self.cx.execute("ATTACH DATABASE ':memory:' AS attached_db")
- self.cx.execute('CREATE TABLE attached_db.foo (key INTEGER)')
- self.cx.executemany('INSERT INTO attached_db.foo (key) VALUES (?)', [(3,), (4,)])
- self.cx.commit()
- with sqlite.connect(':memory:') as bck:
- self.cx.backup(bck, name='attached_db')
- self.verify_backup(bck)
-
-
-if __name__ == "__main__":
- unittest.main()
diff --git a/Lib/sqlite3/test/test_dbapi.py b/Lib/sqlite3/test/test_dbapi.py
deleted file mode 100644
index d825436..0000000
--- a/Lib/sqlite3/test/test_dbapi.py
+++ /dev/null
@@ -1,1161 +0,0 @@
-# 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()
diff --git a/Lib/sqlite3/test/test_dump.py b/Lib/sqlite3/test/test_dump.py
deleted file mode 100644
index 1f14c62..0000000
--- a/Lib/sqlite3/test/test_dump.py
+++ /dev/null
@@ -1,75 +0,0 @@
-# Author: Paul Kippes <kippesp@gmail.com>
-
-import unittest
-import sqlite3 as sqlite
-
-class DumpTests(unittest.TestCase):
- def setUp(self):
- self.cx = sqlite.connect(":memory:")
- self.cu = self.cx.cursor()
-
- def tearDown(self):
- self.cx.close()
-
- def test_table_dump(self):
- expected_sqls = [
- """CREATE TABLE "index"("index" blob);"""
- ,
- """INSERT INTO "index" VALUES(X'01');"""
- ,
- """CREATE TABLE "quoted""table"("quoted""field" text);"""
- ,
- """INSERT INTO "quoted""table" VALUES('quoted''value');"""
- ,
- "CREATE TABLE t1(id integer primary key, s1 text, " \
- "t1_i1 integer not null, i2 integer, unique (s1), " \
- "constraint t1_idx1 unique (i2));"
- ,
- "INSERT INTO \"t1\" VALUES(1,'foo',10,20);"
- ,
- "INSERT INTO \"t1\" VALUES(2,'foo2',30,30);"
- ,
- "CREATE TABLE t2(id integer, t2_i1 integer, " \
- "t2_i2 integer, primary key (id)," \
- "foreign key(t2_i1) references t1(t1_i1));"
- ,
- "CREATE TRIGGER trigger_1 update of t1_i1 on t1 " \
- "begin " \
- "update t2 set t2_i1 = new.t1_i1 where t2_i1 = old.t1_i1; " \
- "end;"
- ,
- "CREATE VIEW v1 as select * from t1 left join t2 " \
- "using (id);"
- ]
- [self.cu.execute(s) for s in expected_sqls]
- i = self.cx.iterdump()
- actual_sqls = [s for s in i]
- expected_sqls = ['BEGIN TRANSACTION;'] + expected_sqls + \
- ['COMMIT;']
- [self.assertEqual(expected_sqls[i], actual_sqls[i])
- for i in range(len(expected_sqls))]
-
- def test_unorderable_row(self):
- # iterdump() should be able to cope with unorderable row types (issue #15545)
- class UnorderableRow:
- def __init__(self, cursor, row):
- self.row = row
- def __getitem__(self, index):
- return self.row[index]
- self.cx.row_factory = UnorderableRow
- CREATE_ALPHA = """CREATE TABLE "alpha" ("one");"""
- CREATE_BETA = """CREATE TABLE "beta" ("two");"""
- expected = [
- "BEGIN TRANSACTION;",
- CREATE_ALPHA,
- CREATE_BETA,
- "COMMIT;"
- ]
- self.cu.execute(CREATE_BETA)
- self.cu.execute(CREATE_ALPHA)
- got = list(self.cx.iterdump())
- self.assertEqual(expected, got)
-
-
-if __name__ == "__main__":
- unittest.main()
diff --git a/Lib/sqlite3/test/test_factory.py b/Lib/sqlite3/test/test_factory.py
deleted file mode 100644
index 420855b..0000000
--- a/Lib/sqlite3/test/test_factory.py
+++ /dev/null
@@ -1,310 +0,0 @@
-# pysqlite2/test/factory.py: tests for the various factories in pysqlite
-#
-# Copyright (C) 2005-2007 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 unittest
-import sqlite3 as sqlite
-from collections.abc import Sequence
-
-
-def dict_factory(cursor, row):
- d = {}
- for idx, col in enumerate(cursor.description):
- d[col[0]] = row[idx]
- return d
-
-class MyCursor(sqlite.Cursor):
- def __init__(self, *args, **kwargs):
- sqlite.Cursor.__init__(self, *args, **kwargs)
- self.row_factory = dict_factory
-
-class ConnectionFactoryTests(unittest.TestCase):
- def test_connection_factories(self):
- class DefectFactory(sqlite.Connection):
- def __init__(self, *args, **kwargs):
- return None
- class OkFactory(sqlite.Connection):
- def __init__(self, *args, **kwargs):
- sqlite.Connection.__init__(self, *args, **kwargs)
-
- for factory in DefectFactory, OkFactory:
- with self.subTest(factory=factory):
- con = sqlite.connect(":memory:", factory=factory)
- self.assertIsInstance(con, factory)
-
-
-class CursorFactoryTests(unittest.TestCase):
- def setUp(self):
- self.con = sqlite.connect(":memory:")
-
- def tearDown(self):
- self.con.close()
-
- def test_is_instance(self):
- cur = self.con.cursor()
- self.assertIsInstance(cur, sqlite.Cursor)
- cur = self.con.cursor(MyCursor)
- self.assertIsInstance(cur, MyCursor)
- cur = self.con.cursor(factory=lambda con: MyCursor(con))
- self.assertIsInstance(cur, MyCursor)
-
- def test_invalid_factory(self):
- # not a callable at all
- self.assertRaises(TypeError, self.con.cursor, None)
- # invalid callable with not exact one argument
- self.assertRaises(TypeError, self.con.cursor, lambda: None)
- # invalid callable returning non-cursor
- self.assertRaises(TypeError, self.con.cursor, lambda con: None)
-
-class RowFactoryTestsBackwardsCompat(unittest.TestCase):
- def setUp(self):
- self.con = sqlite.connect(":memory:")
-
- def test_is_produced_by_factory(self):
- cur = self.con.cursor(factory=MyCursor)
- cur.execute("select 4+5 as foo")
- row = cur.fetchone()
- self.assertIsInstance(row, dict)
- cur.close()
-
- def tearDown(self):
- self.con.close()
-
-class RowFactoryTests(unittest.TestCase):
- def setUp(self):
- self.con = sqlite.connect(":memory:")
-
- def test_custom_factory(self):
- self.con.row_factory = lambda cur, row: list(row)
- row = self.con.execute("select 1, 2").fetchone()
- self.assertIsInstance(row, list)
-
- def test_sqlite_row_index(self):
- self.con.row_factory = sqlite.Row
- row = self.con.execute("select 1 as a_1, 2 as b").fetchone()
- self.assertIsInstance(row, sqlite.Row)
-
- self.assertEqual(row["a_1"], 1, "by name: wrong result for column 'a_1'")
- self.assertEqual(row["b"], 2, "by name: wrong result for column 'b'")
-
- self.assertEqual(row["A_1"], 1, "by name: wrong result for column 'A_1'")
- self.assertEqual(row["B"], 2, "by name: wrong result for column 'B'")
-
- self.assertEqual(row[0], 1, "by index: wrong result for column 0")
- self.assertEqual(row[1], 2, "by index: wrong result for column 1")
- self.assertEqual(row[-1], 2, "by index: wrong result for column -1")
- self.assertEqual(row[-2], 1, "by index: wrong result for column -2")
-
- with self.assertRaises(IndexError):
- row['c']
- with self.assertRaises(IndexError):
- row['a_\x11']
- with self.assertRaises(IndexError):
- row['a\x7f1']
- with self.assertRaises(IndexError):
- row[2]
- with self.assertRaises(IndexError):
- row[-3]
- with self.assertRaises(IndexError):
- row[2**1000]
- with self.assertRaises(IndexError):
- row[complex()] # index must be int or string
-
- def test_sqlite_row_index_unicode(self):
- self.con.row_factory = sqlite.Row
- row = self.con.execute("select 1 as \xff").fetchone()
- self.assertEqual(row["\xff"], 1)
- with self.assertRaises(IndexError):
- row['\u0178']
- with self.assertRaises(IndexError):
- row['\xdf']
-
- def test_sqlite_row_slice(self):
- # A sqlite.Row can be sliced like a list.
- self.con.row_factory = sqlite.Row
- row = self.con.execute("select 1, 2, 3, 4").fetchone()
- self.assertEqual(row[0:0], ())
- self.assertEqual(row[0:1], (1,))
- self.assertEqual(row[1:3], (2, 3))
- self.assertEqual(row[3:1], ())
- # Explicit bounds are optional.
- self.assertEqual(row[1:], (2, 3, 4))
- self.assertEqual(row[:3], (1, 2, 3))
- # Slices can use negative indices.
- self.assertEqual(row[-2:-1], (3,))
- self.assertEqual(row[-2:], (3, 4))
- # Slicing supports steps.
- self.assertEqual(row[0:4:2], (1, 3))
- self.assertEqual(row[3:0:-2], (4, 2))
-
- def test_sqlite_row_iter(self):
- """Checks if the row object is iterable"""
- self.con.row_factory = sqlite.Row
- row = self.con.execute("select 1 as a, 2 as b").fetchone()
- for col in row:
- pass
-
- def test_sqlite_row_as_tuple(self):
- """Checks if the row object can be converted to a tuple"""
- self.con.row_factory = sqlite.Row
- row = self.con.execute("select 1 as a, 2 as b").fetchone()
- t = tuple(row)
- self.assertEqual(t, (row['a'], row['b']))
-
- def test_sqlite_row_as_dict(self):
- """Checks if the row object can be correctly converted to a dictionary"""
- self.con.row_factory = sqlite.Row
- row = self.con.execute("select 1 as a, 2 as b").fetchone()
- d = dict(row)
- self.assertEqual(d["a"], row["a"])
- self.assertEqual(d["b"], row["b"])
-
- def test_sqlite_row_hash_cmp(self):
- """Checks if the row object compares and hashes correctly"""
- self.con.row_factory = sqlite.Row
- row_1 = self.con.execute("select 1 as a, 2 as b").fetchone()
- row_2 = self.con.execute("select 1 as a, 2 as b").fetchone()
- row_3 = self.con.execute("select 1 as a, 3 as b").fetchone()
- row_4 = self.con.execute("select 1 as b, 2 as a").fetchone()
- row_5 = self.con.execute("select 2 as b, 1 as a").fetchone()
-
- self.assertTrue(row_1 == row_1)
- self.assertTrue(row_1 == row_2)
- self.assertFalse(row_1 == row_3)
- self.assertFalse(row_1 == row_4)
- self.assertFalse(row_1 == row_5)
- self.assertFalse(row_1 == object())
-
- self.assertFalse(row_1 != row_1)
- self.assertFalse(row_1 != row_2)
- self.assertTrue(row_1 != row_3)
- self.assertTrue(row_1 != row_4)
- self.assertTrue(row_1 != row_5)
- self.assertTrue(row_1 != object())
-
- with self.assertRaises(TypeError):
- row_1 > row_2
- with self.assertRaises(TypeError):
- row_1 < row_2
- with self.assertRaises(TypeError):
- row_1 >= row_2
- with self.assertRaises(TypeError):
- row_1 <= row_2
-
- self.assertEqual(hash(row_1), hash(row_2))
-
- def test_sqlite_row_as_sequence(self):
- """ Checks if the row object can act like a sequence """
- self.con.row_factory = sqlite.Row
- row = self.con.execute("select 1 as a, 2 as b").fetchone()
-
- as_tuple = tuple(row)
- self.assertEqual(list(reversed(row)), list(reversed(as_tuple)))
- self.assertIsInstance(row, Sequence)
-
- def test_fake_cursor_class(self):
- # Issue #24257: Incorrect use of PyObject_IsInstance() caused
- # segmentation fault.
- # Issue #27861: Also applies for cursor factory.
- class FakeCursor(str):
- __class__ = sqlite.Cursor
- self.con.row_factory = sqlite.Row
- self.assertRaises(TypeError, self.con.cursor, FakeCursor)
- self.assertRaises(TypeError, sqlite.Row, FakeCursor(), ())
-
- def tearDown(self):
- self.con.close()
-
-class TextFactoryTests(unittest.TestCase):
- def setUp(self):
- self.con = sqlite.connect(":memory:")
-
- def test_unicode(self):
- austria = "Österreich"
- row = self.con.execute("select ?", (austria,)).fetchone()
- self.assertEqual(type(row[0]), str, "type of row[0] must be unicode")
-
- def test_string(self):
- self.con.text_factory = bytes
- austria = "Österreich"
- row = self.con.execute("select ?", (austria,)).fetchone()
- self.assertEqual(type(row[0]), bytes, "type of row[0] must be bytes")
- self.assertEqual(row[0], austria.encode("utf-8"), "column must equal original data in UTF-8")
-
- def test_custom(self):
- self.con.text_factory = lambda x: str(x, "utf-8", "ignore")
- austria = "Österreich"
- row = self.con.execute("select ?", (austria,)).fetchone()
- self.assertEqual(type(row[0]), str, "type of row[0] must be unicode")
- self.assertTrue(row[0].endswith("reich"), "column must contain original data")
-
- def test_optimized_unicode(self):
- # OptimizedUnicode is deprecated as of Python 3.10
- with self.assertWarns(DeprecationWarning) as cm:
- self.con.text_factory = sqlite.OptimizedUnicode
- self.assertIn("factory.py", cm.filename)
- austria = "Österreich"
- germany = "Deutchland"
- a_row = self.con.execute("select ?", (austria,)).fetchone()
- d_row = self.con.execute("select ?", (germany,)).fetchone()
- self.assertEqual(type(a_row[0]), str, "type of non-ASCII row must be str")
- self.assertEqual(type(d_row[0]), str, "type of ASCII-only row must be str")
-
- def tearDown(self):
- self.con.close()
-
-class TextFactoryTestsWithEmbeddedZeroBytes(unittest.TestCase):
- def setUp(self):
- self.con = sqlite.connect(":memory:")
- self.con.execute("create table test (value text)")
- self.con.execute("insert into test (value) values (?)", ("a\x00b",))
-
- def test_string(self):
- # text_factory defaults to str
- row = self.con.execute("select value from test").fetchone()
- self.assertIs(type(row[0]), str)
- self.assertEqual(row[0], "a\x00b")
-
- def test_bytes(self):
- self.con.text_factory = bytes
- row = self.con.execute("select value from test").fetchone()
- self.assertIs(type(row[0]), bytes)
- self.assertEqual(row[0], b"a\x00b")
-
- def test_bytearray(self):
- self.con.text_factory = bytearray
- row = self.con.execute("select value from test").fetchone()
- self.assertIs(type(row[0]), bytearray)
- self.assertEqual(row[0], b"a\x00b")
-
- def test_custom(self):
- # A custom factory should receive a bytes argument
- self.con.text_factory = lambda x: x
- row = self.con.execute("select value from test").fetchone()
- self.assertIs(type(row[0]), bytes)
- self.assertEqual(row[0], b"a\x00b")
-
- def tearDown(self):
- self.con.close()
-
-
-if __name__ == "__main__":
- unittest.main()
diff --git a/Lib/sqlite3/test/test_hooks.py b/Lib/sqlite3/test/test_hooks.py
deleted file mode 100644
index bf454b2..0000000
--- a/Lib/sqlite3/test/test_hooks.py
+++ /dev/null
@@ -1,294 +0,0 @@
-# pysqlite2/test/hooks.py: tests for various SQLite-specific hooks
-#
-# Copyright (C) 2006-2007 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 unittest
-import sqlite3 as sqlite
-
-from test.support.os_helper import TESTFN, unlink
-from .test_userfunctions import with_tracebacks
-
-class CollationTests(unittest.TestCase):
- def test_create_collation_not_string(self):
- con = sqlite.connect(":memory:")
- with self.assertRaises(TypeError):
- con.create_collation(None, lambda x, y: (x > y) - (x < y))
-
- def test_create_collation_not_callable(self):
- con = sqlite.connect(":memory:")
- with self.assertRaises(TypeError) as cm:
- con.create_collation("X", 42)
- self.assertEqual(str(cm.exception), 'parameter must be callable')
-
- def test_create_collation_not_ascii(self):
- con = sqlite.connect(":memory:")
- con.create_collation("collä", lambda x, y: (x > y) - (x < y))
-
- def test_create_collation_bad_upper(self):
- class BadUpperStr(str):
- def upper(self):
- return None
- con = sqlite.connect(":memory:")
- mycoll = lambda x, y: -((x > y) - (x < y))
- con.create_collation(BadUpperStr("mycoll"), mycoll)
- result = con.execute("""
- select x from (
- select 'a' as x
- union
- select 'b' as x
- ) order by x collate mycoll
- """).fetchall()
- self.assertEqual(result[0][0], 'b')
- self.assertEqual(result[1][0], 'a')
-
- def test_collation_is_used(self):
- def mycoll(x, y):
- # reverse order
- return -((x > y) - (x < y))
-
- con = sqlite.connect(":memory:")
- con.create_collation("mycoll", mycoll)
- sql = """
- select x from (
- select 'a' as x
- union
- select 'b' as x
- union
- select 'c' as x
- ) order by x collate mycoll
- """
- result = con.execute(sql).fetchall()
- self.assertEqual(result, [('c',), ('b',), ('a',)],
- msg='the expected order was not returned')
-
- con.create_collation("mycoll", None)
- with self.assertRaises(sqlite.OperationalError) as cm:
- result = con.execute(sql).fetchall()
- self.assertEqual(str(cm.exception), 'no such collation sequence: mycoll')
-
- def test_collation_returns_large_integer(self):
- def mycoll(x, y):
- # reverse order
- return -((x > y) - (x < y)) * 2**32
- con = sqlite.connect(":memory:")
- con.create_collation("mycoll", mycoll)
- sql = """
- select x from (
- select 'a' as x
- union
- select 'b' as x
- union
- select 'c' as x
- ) order by x collate mycoll
- """
- result = con.execute(sql).fetchall()
- self.assertEqual(result, [('c',), ('b',), ('a',)],
- msg="the expected order was not returned")
-
- def test_collation_register_twice(self):
- """
- Register two different collation functions under the same name.
- Verify that the last one is actually used.
- """
- con = sqlite.connect(":memory:")
- con.create_collation("mycoll", lambda x, y: (x > y) - (x < y))
- con.create_collation("mycoll", lambda x, y: -((x > y) - (x < y)))
- result = con.execute("""
- select x from (select 'a' as x union select 'b' as x) order by x collate mycoll
- """).fetchall()
- self.assertEqual(result[0][0], 'b')
- self.assertEqual(result[1][0], 'a')
-
- def test_deregister_collation(self):
- """
- Register a collation, then deregister it. Make sure an error is raised if we try
- to use it.
- """
- con = sqlite.connect(":memory:")
- con.create_collation("mycoll", lambda x, y: (x > y) - (x < y))
- con.create_collation("mycoll", None)
- with self.assertRaises(sqlite.OperationalError) as cm:
- con.execute("select 'a' as x union select 'b' as x order by x collate mycoll")
- self.assertEqual(str(cm.exception), 'no such collation sequence: mycoll')
-
-class ProgressTests(unittest.TestCase):
- def test_progress_handler_used(self):
- """
- Test that the progress handler is invoked once it is set.
- """
- con = sqlite.connect(":memory:")
- progress_calls = []
- def progress():
- progress_calls.append(None)
- return 0
- con.set_progress_handler(progress, 1)
- con.execute("""
- create table foo(a, b)
- """)
- self.assertTrue(progress_calls)
-
- def test_opcode_count(self):
- """
- Test that the opcode argument is respected.
- """
- con = sqlite.connect(":memory:")
- progress_calls = []
- def progress():
- progress_calls.append(None)
- return 0
- con.set_progress_handler(progress, 1)
- curs = con.cursor()
- curs.execute("""
- create table foo (a, b)
- """)
- first_count = len(progress_calls)
- progress_calls = []
- con.set_progress_handler(progress, 2)
- curs.execute("""
- create table bar (a, b)
- """)
- second_count = len(progress_calls)
- self.assertGreaterEqual(first_count, second_count)
-
- def test_cancel_operation(self):
- """
- Test that returning a non-zero value stops the operation in progress.
- """
- con = sqlite.connect(":memory:")
- def progress():
- return 1
- con.set_progress_handler(progress, 1)
- curs = con.cursor()
- self.assertRaises(
- sqlite.OperationalError,
- curs.execute,
- "create table bar (a, b)")
-
- def test_clear_handler(self):
- """
- Test that setting the progress handler to None clears the previously set handler.
- """
- con = sqlite.connect(":memory:")
- action = 0
- def progress():
- nonlocal action
- action = 1
- return 0
- con.set_progress_handler(progress, 1)
- con.set_progress_handler(None, 1)
- con.execute("select 1 union select 2 union select 3").fetchall()
- self.assertEqual(action, 0, "progress handler was not cleared")
-
- @with_tracebacks(['bad_progress', 'ZeroDivisionError'])
- def test_error_in_progress_handler(self):
- con = sqlite.connect(":memory:")
- def bad_progress():
- 1 / 0
- con.set_progress_handler(bad_progress, 1)
- with self.assertRaises(sqlite.OperationalError):
- con.execute("""
- create table foo(a, b)
- """)
-
- @with_tracebacks(['__bool__', 'ZeroDivisionError'])
- def test_error_in_progress_handler_result(self):
- con = sqlite.connect(":memory:")
- class BadBool:
- def __bool__(self):
- 1 / 0
- def bad_progress():
- return BadBool()
- con.set_progress_handler(bad_progress, 1)
- with self.assertRaises(sqlite.OperationalError):
- con.execute("""
- create table foo(a, b)
- """)
-
-
-class TraceCallbackTests(unittest.TestCase):
- def test_trace_callback_used(self):
- """
- Test that the trace callback is invoked once it is set.
- """
- con = sqlite.connect(":memory:")
- traced_statements = []
- def trace(statement):
- traced_statements.append(statement)
- con.set_trace_callback(trace)
- con.execute("create table foo(a, b)")
- self.assertTrue(traced_statements)
- self.assertTrue(any("create table foo" in stmt for stmt in traced_statements))
-
- def test_clear_trace_callback(self):
- """
- Test that setting the trace callback to None clears the previously set callback.
- """
- con = sqlite.connect(":memory:")
- traced_statements = []
- def trace(statement):
- traced_statements.append(statement)
- con.set_trace_callback(trace)
- con.set_trace_callback(None)
- con.execute("create table foo(a, b)")
- self.assertFalse(traced_statements, "trace callback was not cleared")
-
- def test_unicode_content(self):
- """
- Test that the statement can contain unicode literals.
- """
- unicode_value = '\xf6\xe4\xfc\xd6\xc4\xdc\xdf\u20ac'
- con = sqlite.connect(":memory:")
- traced_statements = []
- def trace(statement):
- traced_statements.append(statement)
- con.set_trace_callback(trace)
- con.execute("create table foo(x)")
- con.execute("insert into foo(x) values ('%s')" % unicode_value)
- con.commit()
- self.assertTrue(any(unicode_value in stmt for stmt in traced_statements),
- "Unicode data %s garbled in trace callback: %s"
- % (ascii(unicode_value), ', '.join(map(ascii, traced_statements))))
-
- def test_trace_callback_content(self):
- # set_trace_callback() shouldn't produce duplicate content (bpo-26187)
- traced_statements = []
- def trace(statement):
- traced_statements.append(statement)
-
- queries = ["create table foo(x)",
- "insert into foo(x) values(1)"]
- self.addCleanup(unlink, TESTFN)
- con1 = sqlite.connect(TESTFN, isolation_level=None)
- con2 = sqlite.connect(TESTFN)
- try:
- con1.set_trace_callback(trace)
- cur = con1.cursor()
- cur.execute(queries[0])
- con2.execute("create table bar(x)")
- cur.execute(queries[1])
- finally:
- con1.close()
- con2.close()
- self.assertEqual(traced_statements, queries)
-
-
-if __name__ == "__main__":
- unittest.main()
diff --git a/Lib/sqlite3/test/test_regression.py b/Lib/sqlite3/test/test_regression.py
deleted file mode 100644
index 3d71809..0000000
--- a/Lib/sqlite3/test/test_regression.py
+++ /dev/null
@@ -1,491 +0,0 @@
-# pysqlite2/test/regression.py: pysqlite regression tests
-#
-# Copyright (C) 2006-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 datetime
-import sys
-import unittest
-import sqlite3 as sqlite
-import weakref
-import functools
-from test import support
-
-from .test_dbapi import managed_connect
-
-class RegressionTests(unittest.TestCase):
- def setUp(self):
- self.con = sqlite.connect(":memory:")
-
- def tearDown(self):
- self.con.close()
-
- def test_pragma_user_version(self):
- # This used to crash pysqlite because this pragma command returns NULL for the column name
- cur = self.con.cursor()
- cur.execute("pragma user_version")
-
- def test_pragma_schema_version(self):
- # This still crashed pysqlite <= 2.2.1
- con = sqlite.connect(":memory:", detect_types=sqlite.PARSE_COLNAMES)
- try:
- cur = self.con.cursor()
- cur.execute("pragma schema_version")
- finally:
- cur.close()
- con.close()
-
- def test_statement_reset(self):
- # pysqlite 2.1.0 to 2.2.0 have the problem that not all statements are
- # reset before a rollback, but only those that are still in the
- # statement cache. The others are not accessible from the connection object.
- con = sqlite.connect(":memory:", cached_statements=5)
- cursors = [con.cursor() for x in range(5)]
- cursors[0].execute("create table test(x)")
- for i in range(10):
- cursors[0].executemany("insert into test(x) values (?)", [(x,) for x in range(10)])
-
- for i in range(5):
- cursors[i].execute(" " * i + "select x from test")
-
- con.rollback()
-
- def test_column_name_with_spaces(self):
- cur = self.con.cursor()
- cur.execute('select 1 as "foo bar [datetime]"')
- self.assertEqual(cur.description[0][0], "foo bar [datetime]")
-
- cur.execute('select 1 as "foo baz"')
- self.assertEqual(cur.description[0][0], "foo baz")
-
- def test_statement_finalization_on_close_db(self):
- # pysqlite versions <= 2.3.3 only finalized statements in the statement
- # cache when closing the database. statements that were still
- # referenced in cursors weren't closed and could provoke "
- # "OperationalError: Unable to close due to unfinalised statements".
- con = sqlite.connect(":memory:")
- cursors = []
- # default statement cache size is 100
- for i in range(105):
- cur = con.cursor()
- cursors.append(cur)
- cur.execute("select 1 x union select " + str(i))
- con.close()
-
- def test_on_conflict_rollback(self):
- con = sqlite.connect(":memory:")
- con.execute("create table foo(x, unique(x) on conflict rollback)")
- con.execute("insert into foo(x) values (1)")
- try:
- con.execute("insert into foo(x) values (1)")
- except sqlite.DatabaseError:
- pass
- con.execute("insert into foo(x) values (2)")
- try:
- con.commit()
- except sqlite.OperationalError:
- self.fail("pysqlite knew nothing about the implicit ROLLBACK")
-
- def test_workaround_for_buggy_sqlite_transfer_bindings(self):
- """
- pysqlite would crash with older SQLite versions unless
- a workaround is implemented.
- """
- self.con.execute("create table foo(bar)")
- self.con.execute("drop table foo")
- self.con.execute("create table foo(bar)")
-
- def test_empty_statement(self):
- """
- pysqlite used to segfault with SQLite versions 3.5.x. These return NULL
- for "no-operation" statements
- """
- self.con.execute("")
-
- def test_type_map_usage(self):
- """
- pysqlite until 2.4.1 did not rebuild the row_cast_map when recompiling
- a statement. This test exhibits the problem.
- """
- SELECT = "select * from foo"
- con = sqlite.connect(":memory:",detect_types=sqlite.PARSE_DECLTYPES)
- cur = con.cursor()
- cur.execute("create table foo(bar timestamp)")
- cur.execute("insert into foo(bar) values (?)", (datetime.datetime.now(),))
- cur.execute(SELECT)
- cur.execute("drop table foo")
- cur.execute("create table foo(bar integer)")
- cur.execute("insert into foo(bar) values (5)")
- cur.execute(SELECT)
-
- def test_bind_mutating_list(self):
- # Issue41662: Crash when mutate a list of parameters during iteration.
- class X:
- def __conform__(self, protocol):
- parameters.clear()
- return "..."
- parameters = [X(), 0]
- con = sqlite.connect(":memory:",detect_types=sqlite.PARSE_DECLTYPES)
- con.execute("create table foo(bar X, baz integer)")
- # Should not crash
- with self.assertRaises(IndexError):
- con.execute("insert into foo(bar, baz) values (?, ?)", parameters)
-
- def test_error_msg_decode_error(self):
- # When porting the module to Python 3.0, the error message about
- # decoding errors disappeared. This verifies they're back again.
- with self.assertRaises(sqlite.OperationalError) as cm:
- self.con.execute("select 'xxx' || ? || 'yyy' colname",
- (bytes(bytearray([250])),)).fetchone()
- msg = "Could not decode to UTF-8 column 'colname' with text 'xxx"
- self.assertIn(msg, str(cm.exception))
-
- def test_register_adapter(self):
- """
- See issue 3312.
- """
- self.assertRaises(TypeError, sqlite.register_adapter, {}, None)
-
- def test_set_isolation_level(self):
- # See issue 27881.
- class CustomStr(str):
- def upper(self):
- return None
- def __del__(self):
- con.isolation_level = ""
-
- con = sqlite.connect(":memory:")
- con.isolation_level = None
- for level in "", "DEFERRED", "IMMEDIATE", "EXCLUSIVE":
- with self.subTest(level=level):
- con.isolation_level = level
- con.isolation_level = level.lower()
- con.isolation_level = level.capitalize()
- con.isolation_level = CustomStr(level)
-
- # setting isolation_level failure should not alter previous state
- con.isolation_level = None
- con.isolation_level = "DEFERRED"
- pairs = [
- (1, TypeError), (b'', TypeError), ("abc", ValueError),
- ("IMMEDIATE\0EXCLUSIVE", ValueError), ("\xe9", ValueError),
- ]
- for value, exc in pairs:
- with self.subTest(level=value):
- with self.assertRaises(exc):
- con.isolation_level = value
- self.assertEqual(con.isolation_level, "DEFERRED")
-
- def test_cursor_constructor_call_check(self):
- """
- Verifies that cursor methods check whether base class __init__ was
- called.
- """
- class Cursor(sqlite.Cursor):
- def __init__(self, con):
- pass
-
- con = sqlite.connect(":memory:")
- cur = Cursor(con)
- with self.assertRaises(sqlite.ProgrammingError):
- cur.execute("select 4+5").fetchall()
- with self.assertRaisesRegex(sqlite.ProgrammingError,
- r'^Base Cursor\.__init__ not called\.$'):
- cur.close()
-
- def test_str_subclass(self):
- """
- The Python 3.0 port of the module didn't cope with values of subclasses of str.
- """
- class MyStr(str): pass
- self.con.execute("select ?", (MyStr("abc"),))
-
- def test_connection_constructor_call_check(self):
- """
- Verifies that connection methods check whether base class __init__ was
- called.
- """
- class Connection(sqlite.Connection):
- def __init__(self, name):
- pass
-
- con = Connection(":memory:")
- with self.assertRaises(sqlite.ProgrammingError):
- cur = con.cursor()
-
- def test_cursor_registration(self):
- """
- Verifies that subclassed cursor classes are correctly registered with
- the connection object, too. (fetch-across-rollback problem)
- """
- class Connection(sqlite.Connection):
- def cursor(self):
- return Cursor(self)
-
- class Cursor(sqlite.Cursor):
- def __init__(self, con):
- sqlite.Cursor.__init__(self, con)
-
- con = Connection(":memory:")
- cur = con.cursor()
- cur.execute("create table foo(x)")
- cur.executemany("insert into foo(x) values (?)", [(3,), (4,), (5,)])
- cur.execute("select x from foo")
- con.rollback()
- with self.assertRaises(sqlite.InterfaceError):
- cur.fetchall()
-
- def test_auto_commit(self):
- """
- Verifies that creating a connection in autocommit mode works.
- 2.5.3 introduced a regression so that these could no longer
- be created.
- """
- con = sqlite.connect(":memory:", isolation_level=None)
-
- def test_pragma_autocommit(self):
- """
- Verifies that running a PRAGMA statement that does an autocommit does
- work. This did not work in 2.5.3/2.5.4.
- """
- cur = self.con.cursor()
- cur.execute("create table foo(bar)")
- cur.execute("insert into foo(bar) values (5)")
-
- cur.execute("pragma page_size")
- row = cur.fetchone()
-
- def test_connection_call(self):
- """
- Call a connection with a non-string SQL request: check error handling
- of the statement constructor.
- """
- self.assertRaises(TypeError, self.con, b"select 1")
-
- def test_collation(self):
- def collation_cb(a, b):
- return 1
- self.assertRaises(UnicodeEncodeError, self.con.create_collation,
- # Lone surrogate cannot be encoded to the default encoding (utf8)
- "\uDC80", collation_cb)
-
- def test_recursive_cursor_use(self):
- """
- http://bugs.python.org/issue10811
-
- Recursively using a cursor, such as when reusing it from a generator led to segfaults.
- Now we catch recursive cursor usage and raise a ProgrammingError.
- """
- con = sqlite.connect(":memory:")
-
- cur = con.cursor()
- cur.execute("create table a (bar)")
- cur.execute("create table b (baz)")
-
- def foo():
- cur.execute("insert into a (bar) values (?)", (1,))
- yield 1
-
- with self.assertRaises(sqlite.ProgrammingError):
- cur.executemany("insert into b (baz) values (?)",
- ((i,) for i in foo()))
-
- def test_convert_timestamp_microsecond_padding(self):
- """
- http://bugs.python.org/issue14720
-
- The microsecond parsing of convert_timestamp() should pad with zeros,
- since the microsecond string "456" actually represents "456000".
- """
-
- con = sqlite.connect(":memory:", detect_types=sqlite.PARSE_DECLTYPES)
- cur = con.cursor()
- cur.execute("CREATE TABLE t (x TIMESTAMP)")
-
- # Microseconds should be 456000
- cur.execute("INSERT INTO t (x) VALUES ('2012-04-04 15:06:00.456')")
-
- # Microseconds should be truncated to 123456
- cur.execute("INSERT INTO t (x) VALUES ('2012-04-04 15:06:00.123456789')")
-
- cur.execute("SELECT * FROM t")
- values = [x[0] for x in cur.fetchall()]
-
- self.assertEqual(values, [
- datetime.datetime(2012, 4, 4, 15, 6, 0, 456000),
- datetime.datetime(2012, 4, 4, 15, 6, 0, 123456),
- ])
-
- def test_invalid_isolation_level_type(self):
- # isolation level is a string, not an integer
- self.assertRaises(TypeError,
- sqlite.connect, ":memory:", isolation_level=123)
-
-
- def test_null_character(self):
- # Issue #21147
- con = sqlite.connect(":memory:")
- self.assertRaises(ValueError, con, "\0select 1")
- self.assertRaises(ValueError, con, "select 1\0")
- cur = con.cursor()
- self.assertRaises(ValueError, cur.execute, " \0select 2")
- self.assertRaises(ValueError, cur.execute, "select 2\0")
-
- def test_surrogates(self):
- con = sqlite.connect(":memory:")
- self.assertRaises(UnicodeEncodeError, con, "select '\ud8ff'")
- self.assertRaises(UnicodeEncodeError, con, "select '\udcff'")
- cur = con.cursor()
- self.assertRaises(UnicodeEncodeError, cur.execute, "select '\ud8ff'")
- self.assertRaises(UnicodeEncodeError, cur.execute, "select '\udcff'")
-
- @unittest.skipUnless(sys.maxsize > 2**32, 'requires 64bit platform')
- @support.bigmemtest(size=2**31, memuse=4, dry_run=False)
- def test_large_sql(self, maxsize):
- # Test two cases: size+1 > INT_MAX and size+1 <= INT_MAX.
- for size in (2**31, 2**31-2):
- con = sqlite.connect(":memory:")
- sql = "select 1".ljust(size)
- self.assertRaises(sqlite.DataError, con, sql)
- cur = con.cursor()
- self.assertRaises(sqlite.DataError, cur.execute, sql)
- del sql
-
- def test_commit_cursor_reset(self):
- """
- Connection.commit() did reset cursors, which made sqlite3
- to return rows multiple times when fetched from cursors
- after commit. See issues 10513 and 23129 for details.
- """
- con = sqlite.connect(":memory:")
- con.executescript("""
- create table t(c);
- create table t2(c);
- insert into t values(0);
- insert into t values(1);
- insert into t values(2);
- """)
-
- self.assertEqual(con.isolation_level, "")
-
- counter = 0
- for i, row in enumerate(con.execute("select c from t")):
- with self.subTest(i=i, row=row):
- con.execute("insert into t2(c) values (?)", (i,))
- con.commit()
- if counter == 0:
- self.assertEqual(row[0], 0)
- elif counter == 1:
- self.assertEqual(row[0], 1)
- elif counter == 2:
- self.assertEqual(row[0], 2)
- counter += 1
- self.assertEqual(counter, 3, "should have returned exactly three rows")
-
- def test_bpo31770(self):
- """
- The interpreter shouldn't crash in case Cursor.__init__() is called
- more than once.
- """
- def callback(*args):
- pass
- con = sqlite.connect(":memory:")
- cur = sqlite.Cursor(con)
- ref = weakref.ref(cur, callback)
- cur.__init__(con)
- del cur
- # The interpreter shouldn't crash when ref is collected.
- del ref
- support.gc_collect()
-
- def test_del_isolation_level_segfault(self):
- with self.assertRaises(AttributeError):
- del self.con.isolation_level
-
- def test_bpo37347(self):
- class Printer:
- def log(self, *args):
- return sqlite.SQLITE_OK
-
- 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 test_return_empty_bytestring(self):
- cur = self.con.execute("select X''")
- val = cur.fetchone()[0]
- self.assertEqual(val, b'')
-
- def test_table_lock_cursor_replace_stmt(self):
- with managed_connect(":memory:", in_mem=True) as con:
- cur = con.cursor()
- cur.execute("create table t(t)")
- cur.executemany("insert into t values(?)",
- ((v,) for v in range(5)))
- con.commit()
- cur.execute("select t from t")
- cur.execute("drop table t")
- con.commit()
-
- def test_table_lock_cursor_dealloc(self):
- with managed_connect(":memory:", in_mem=True) as con:
- con.execute("create table t(t)")
- con.executemany("insert into t values(?)",
- ((v,) for v in range(5)))
- con.commit()
- cur = con.execute("select t from t")
- del cur
- con.execute("drop table t")
- con.commit()
-
- def test_table_lock_cursor_non_readonly_select(self):
- with managed_connect(":memory:", in_mem=True) as con:
- con.execute("create table t(t)")
- con.executemany("insert into t values(?)",
- ((v,) for v in range(5)))
- con.commit()
- def dup(v):
- con.execute("insert into t values(?)", (v,))
- return
- con.create_function("dup", 1, dup)
- cur = con.execute("select dup(t) from t")
- del cur
- con.execute("drop table t")
- con.commit()
-
- def test_executescript_step_through_select(self):
- with managed_connect(":memory:", in_mem=True) as con:
- values = [(v,) for v in range(5)]
- with con:
- con.execute("create table t(t)")
- con.executemany("insert into t values(?)", values)
- steps = []
- con.create_function("step", 1, lambda x: steps.append((x,)))
- con.executescript("select step(t) from t")
- self.assertEqual(steps, values)
-
-
-if __name__ == "__main__":
- unittest.main()
diff --git a/Lib/sqlite3/test/test_transactions.py b/Lib/sqlite3/test/test_transactions.py
deleted file mode 100644
index 8ea352b..0000000
--- a/Lib/sqlite3/test/test_transactions.py
+++ /dev/null
@@ -1,200 +0,0 @@
-# pysqlite2/test/transactions.py: tests transactions
-#
-# Copyright (C) 2005-2007 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 os, unittest
-import sqlite3 as sqlite
-
-def get_db_path():
- return "sqlite_testdb"
-
-class TransactionTests(unittest.TestCase):
- def setUp(self):
- try:
- os.remove(get_db_path())
- except OSError:
- pass
-
- self.con1 = sqlite.connect(get_db_path(), timeout=0.1)
- self.cur1 = self.con1.cursor()
-
- self.con2 = sqlite.connect(get_db_path(), timeout=0.1)
- self.cur2 = self.con2.cursor()
-
- def tearDown(self):
- self.cur1.close()
- self.con1.close()
-
- self.cur2.close()
- self.con2.close()
-
- try:
- os.unlink(get_db_path())
- except OSError:
- pass
-
- def test_dml_does_not_auto_commit_before(self):
- self.cur1.execute("create table test(i)")
- self.cur1.execute("insert into test(i) values (5)")
- self.cur1.execute("create table test2(j)")
- self.cur2.execute("select i from test")
- res = self.cur2.fetchall()
- self.assertEqual(len(res), 0)
-
- def test_insert_starts_transaction(self):
- self.cur1.execute("create table test(i)")
- self.cur1.execute("insert into test(i) values (5)")
- self.cur2.execute("select i from test")
- res = self.cur2.fetchall()
- self.assertEqual(len(res), 0)
-
- def test_update_starts_transaction(self):
- self.cur1.execute("create table test(i)")
- self.cur1.execute("insert into test(i) values (5)")
- self.con1.commit()
- self.cur1.execute("update test set i=6")
- self.cur2.execute("select i from test")
- res = self.cur2.fetchone()[0]
- self.assertEqual(res, 5)
-
- def test_delete_starts_transaction(self):
- self.cur1.execute("create table test(i)")
- self.cur1.execute("insert into test(i) values (5)")
- self.con1.commit()
- self.cur1.execute("delete from test")
- self.cur2.execute("select i from test")
- res = self.cur2.fetchall()
- self.assertEqual(len(res), 1)
-
- def test_replace_starts_transaction(self):
- self.cur1.execute("create table test(i)")
- self.cur1.execute("insert into test(i) values (5)")
- self.con1.commit()
- self.cur1.execute("replace into test(i) values (6)")
- self.cur2.execute("select i from test")
- res = self.cur2.fetchall()
- self.assertEqual(len(res), 1)
- self.assertEqual(res[0][0], 5)
-
- def test_toggle_auto_commit(self):
- self.cur1.execute("create table test(i)")
- self.cur1.execute("insert into test(i) values (5)")
- self.con1.isolation_level = None
- self.assertEqual(self.con1.isolation_level, None)
- self.cur2.execute("select i from test")
- res = self.cur2.fetchall()
- self.assertEqual(len(res), 1)
-
- self.con1.isolation_level = "DEFERRED"
- self.assertEqual(self.con1.isolation_level , "DEFERRED")
- self.cur1.execute("insert into test(i) values (5)")
- self.cur2.execute("select i from test")
- res = self.cur2.fetchall()
- self.assertEqual(len(res), 1)
-
- def test_raise_timeout(self):
- self.cur1.execute("create table test(i)")
- self.cur1.execute("insert into test(i) values (5)")
- with self.assertRaises(sqlite.OperationalError):
- self.cur2.execute("insert into test(i) values (5)")
-
- def test_locking(self):
- """
- This tests the improved concurrency with pysqlite 2.3.4. You needed
- to roll back con2 before you could commit con1.
- """
- self.cur1.execute("create table test(i)")
- self.cur1.execute("insert into test(i) values (5)")
- with self.assertRaises(sqlite.OperationalError):
- self.cur2.execute("insert into test(i) values (5)")
- # NO self.con2.rollback() HERE!!!
- self.con1.commit()
-
- def test_rollback_cursor_consistency(self):
- """
- Checks if cursors on the connection are set into a "reset" state
- when a rollback is done on the connection.
- """
- con = sqlite.connect(":memory:")
- cur = con.cursor()
- cur.execute("create table test(x)")
- cur.execute("insert into test(x) values (5)")
- cur.execute("select 1 union select 2 union select 3")
-
- con.rollback()
- with self.assertRaises(sqlite.InterfaceError):
- cur.fetchall()
-
-class SpecialCommandTests(unittest.TestCase):
- def setUp(self):
- self.con = sqlite.connect(":memory:")
- self.cur = self.con.cursor()
-
- def test_drop_table(self):
- self.cur.execute("create table test(i)")
- self.cur.execute("insert into test(i) values (5)")
- self.cur.execute("drop table test")
-
- def test_pragma(self):
- self.cur.execute("create table test(i)")
- self.cur.execute("insert into test(i) values (5)")
- self.cur.execute("pragma count_changes=1")
-
- def tearDown(self):
- self.cur.close()
- self.con.close()
-
-class TransactionalDDL(unittest.TestCase):
- def setUp(self):
- self.con = sqlite.connect(":memory:")
-
- def test_ddl_does_not_autostart_transaction(self):
- # For backwards compatibility reasons, DDL statements should not
- # implicitly start a transaction.
- self.con.execute("create table test(i)")
- self.con.rollback()
- result = self.con.execute("select * from test").fetchall()
- self.assertEqual(result, [])
-
- def test_immediate_transactional_ddl(self):
- # You can achieve transactional DDL by issuing a BEGIN
- # statement manually.
- self.con.execute("begin immediate")
- self.con.execute("create table test(i)")
- self.con.rollback()
- with self.assertRaises(sqlite.OperationalError):
- self.con.execute("select * from test")
-
- def test_transactional_ddl(self):
- # You can achieve transactional DDL by issuing a BEGIN
- # statement manually.
- self.con.execute("begin")
- self.con.execute("create table test(i)")
- self.con.rollback()
- with self.assertRaises(sqlite.OperationalError):
- self.con.execute("select * from test")
-
- def tearDown(self):
- self.con.close()
-
-
-if __name__ == "__main__":
- unittest.main()
diff --git a/Lib/sqlite3/test/test_types.py b/Lib/sqlite3/test/test_types.py
deleted file mode 100644
index 0cfb72c..0000000
--- a/Lib/sqlite3/test/test_types.py
+++ /dev/null
@@ -1,535 +0,0 @@
-# pysqlite2/test/types.py: tests for type conversion and detection
-#
-# Copyright (C) 2005 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 datetime
-import unittest
-import sqlite3 as sqlite
-import sys
-try:
- import zlib
-except ImportError:
- zlib = None
-
-from test import support
-
-
-class SqliteTypeTests(unittest.TestCase):
- def setUp(self):
- self.con = sqlite.connect(":memory:")
- self.cur = self.con.cursor()
- self.cur.execute("create table test(i integer, s varchar, f number, b blob)")
-
- def tearDown(self):
- self.cur.close()
- self.con.close()
-
- def test_string(self):
- self.cur.execute("insert into test(s) values (?)", ("Österreich",))
- self.cur.execute("select s from test")
- row = self.cur.fetchone()
- self.assertEqual(row[0], "Österreich")
-
- def test_string_with_null_character(self):
- self.cur.execute("insert into test(s) values (?)", ("a\0b",))
- self.cur.execute("select s from test")
- row = self.cur.fetchone()
- self.assertEqual(row[0], "a\0b")
-
- def test_small_int(self):
- self.cur.execute("insert into test(i) values (?)", (42,))
- self.cur.execute("select i from test")
- row = self.cur.fetchone()
- self.assertEqual(row[0], 42)
-
- def test_large_int(self):
- num = 123456789123456789
- self.cur.execute("insert into test(i) values (?)", (num,))
- self.cur.execute("select i from test")
- row = self.cur.fetchone()
- self.assertEqual(row[0], num)
-
- def test_float(self):
- val = 3.14
- self.cur.execute("insert into test(f) values (?)", (val,))
- self.cur.execute("select f from test")
- row = self.cur.fetchone()
- self.assertEqual(row[0], val)
-
- def test_blob(self):
- sample = b"Guglhupf"
- val = memoryview(sample)
- self.cur.execute("insert into test(b) values (?)", (val,))
- self.cur.execute("select b from test")
- row = self.cur.fetchone()
- self.assertEqual(row[0], sample)
-
- def test_unicode_execute(self):
- self.cur.execute("select 'Österreich'")
- row = self.cur.fetchone()
- self.assertEqual(row[0], "Österreich")
-
- def test_too_large_int(self):
- for value in 2**63, -2**63-1, 2**64:
- with self.assertRaises(OverflowError):
- self.cur.execute("insert into test(i) values (?)", (value,))
- self.cur.execute("select i from test")
- row = self.cur.fetchone()
- self.assertIsNone(row)
-
- def test_string_with_surrogates(self):
- for value in 0xd8ff, 0xdcff:
- with self.assertRaises(UnicodeEncodeError):
- self.cur.execute("insert into test(s) values (?)", (chr(value),))
- self.cur.execute("select s from test")
- row = self.cur.fetchone()
- self.assertIsNone(row)
-
- @unittest.skipUnless(sys.maxsize > 2**32, 'requires 64bit platform')
- @support.bigmemtest(size=2**31, memuse=4, dry_run=False)
- def test_too_large_string(self, maxsize):
- with self.assertRaises(sqlite.InterfaceError):
- self.cur.execute("insert into test(s) values (?)", ('x'*(2**31-1),))
- with self.assertRaises(OverflowError):
- self.cur.execute("insert into test(s) values (?)", ('x'*(2**31),))
- self.cur.execute("select 1 from test")
- row = self.cur.fetchone()
- self.assertIsNone(row)
-
- @unittest.skipUnless(sys.maxsize > 2**32, 'requires 64bit platform')
- @support.bigmemtest(size=2**31, memuse=3, dry_run=False)
- def test_too_large_blob(self, maxsize):
- with self.assertRaises(sqlite.InterfaceError):
- self.cur.execute("insert into test(s) values (?)", (b'x'*(2**31-1),))
- with self.assertRaises(OverflowError):
- self.cur.execute("insert into test(s) values (?)", (b'x'*(2**31),))
- self.cur.execute("select 1 from test")
- row = self.cur.fetchone()
- self.assertIsNone(row)
-
-
-class DeclTypesTests(unittest.TestCase):
- class Foo:
- def __init__(self, _val):
- if isinstance(_val, bytes):
- # sqlite3 always calls __init__ with a bytes created from a
- # UTF-8 string when __conform__ was used to store the object.
- _val = _val.decode('utf-8')
- self.val = _val
-
- def __eq__(self, other):
- if not isinstance(other, DeclTypesTests.Foo):
- return NotImplemented
- return self.val == other.val
-
- def __conform__(self, protocol):
- if protocol is sqlite.PrepareProtocol:
- return self.val
- else:
- return None
-
- def __str__(self):
- return "<%s>" % self.val
-
- class BadConform:
- def __init__(self, exc):
- self.exc = exc
- def __conform__(self, protocol):
- raise self.exc
-
- def setUp(self):
- self.con = sqlite.connect(":memory:", detect_types=sqlite.PARSE_DECLTYPES)
- self.cur = self.con.cursor()
- self.cur.execute("""
- create table test(
- i int,
- s str,
- f float,
- b bool,
- u unicode,
- foo foo,
- bin blob,
- n1 number,
- n2 number(5),
- bad bad,
- cbin cblob)
- """)
-
- # override float, make them always return the same number
- sqlite.converters["FLOAT"] = lambda x: 47.2
-
- # and implement two custom ones
- sqlite.converters["BOOL"] = lambda x: bool(int(x))
- sqlite.converters["FOO"] = DeclTypesTests.Foo
- sqlite.converters["BAD"] = DeclTypesTests.BadConform
- sqlite.converters["WRONG"] = lambda x: "WRONG"
- sqlite.converters["NUMBER"] = float
- sqlite.converters["CBLOB"] = lambda x: b"blobish"
-
- def tearDown(self):
- del sqlite.converters["FLOAT"]
- del sqlite.converters["BOOL"]
- del sqlite.converters["FOO"]
- del sqlite.converters["BAD"]
- del sqlite.converters["WRONG"]
- del sqlite.converters["NUMBER"]
- del sqlite.converters["CBLOB"]
- self.cur.close()
- self.con.close()
-
- def test_string(self):
- # default
- self.cur.execute("insert into test(s) values (?)", ("foo",))
- self.cur.execute('select s as "s [WRONG]" from test')
- row = self.cur.fetchone()
- self.assertEqual(row[0], "foo")
-
- def test_small_int(self):
- # default
- self.cur.execute("insert into test(i) values (?)", (42,))
- self.cur.execute("select i from test")
- row = self.cur.fetchone()
- self.assertEqual(row[0], 42)
-
- def test_large_int(self):
- # default
- num = 123456789123456789
- self.cur.execute("insert into test(i) values (?)", (num,))
- self.cur.execute("select i from test")
- row = self.cur.fetchone()
- self.assertEqual(row[0], num)
-
- def test_float(self):
- # custom
- val = 3.14
- self.cur.execute("insert into test(f) values (?)", (val,))
- self.cur.execute("select f from test")
- row = self.cur.fetchone()
- self.assertEqual(row[0], 47.2)
-
- def test_bool(self):
- # custom
- self.cur.execute("insert into test(b) values (?)", (False,))
- self.cur.execute("select b from test")
- row = self.cur.fetchone()
- self.assertIs(row[0], False)
-
- self.cur.execute("delete from test")
- self.cur.execute("insert into test(b) values (?)", (True,))
- self.cur.execute("select b from test")
- row = self.cur.fetchone()
- self.assertIs(row[0], True)
-
- def test_unicode(self):
- # default
- val = "\xd6sterreich"
- self.cur.execute("insert into test(u) values (?)", (val,))
- self.cur.execute("select u from test")
- row = self.cur.fetchone()
- self.assertEqual(row[0], val)
-
- def test_foo(self):
- val = DeclTypesTests.Foo("bla")
- self.cur.execute("insert into test(foo) values (?)", (val,))
- self.cur.execute("select foo from test")
- row = self.cur.fetchone()
- self.assertEqual(row[0], val)
-
- def test_error_in_conform(self):
- val = DeclTypesTests.BadConform(TypeError)
- with self.assertRaises(sqlite.InterfaceError):
- self.cur.execute("insert into test(bad) values (?)", (val,))
- with self.assertRaises(sqlite.InterfaceError):
- self.cur.execute("insert into test(bad) values (:val)", {"val": val})
-
- val = DeclTypesTests.BadConform(KeyboardInterrupt)
- with self.assertRaises(KeyboardInterrupt):
- self.cur.execute("insert into test(bad) values (?)", (val,))
- with self.assertRaises(KeyboardInterrupt):
- self.cur.execute("insert into test(bad) values (:val)", {"val": val})
-
- def test_unsupported_seq(self):
- class Bar: pass
- val = Bar()
- with self.assertRaises(sqlite.InterfaceError):
- self.cur.execute("insert into test(f) values (?)", (val,))
-
- def test_unsupported_dict(self):
- class Bar: pass
- val = Bar()
- with self.assertRaises(sqlite.InterfaceError):
- self.cur.execute("insert into test(f) values (:val)", {"val": val})
-
- def test_blob(self):
- # default
- sample = b"Guglhupf"
- val = memoryview(sample)
- self.cur.execute("insert into test(bin) values (?)", (val,))
- self.cur.execute("select bin from test")
- row = self.cur.fetchone()
- self.assertEqual(row[0], sample)
-
- def test_number1(self):
- self.cur.execute("insert into test(n1) values (5)")
- value = self.cur.execute("select n1 from test").fetchone()[0]
- # if the converter is not used, it's an int instead of a float
- self.assertEqual(type(value), float)
-
- def test_number2(self):
- """Checks whether converter names are cut off at '(' characters"""
- self.cur.execute("insert into test(n2) values (5)")
- value = self.cur.execute("select n2 from test").fetchone()[0]
- # if the converter is not used, it's an int instead of a float
- self.assertEqual(type(value), float)
-
- def test_convert_zero_sized_blob(self):
- self.con.execute("insert into test(cbin) values (?)", (b"",))
- cur = self.con.execute("select cbin from test")
- # Zero-sized blobs with converters returns None. This differs from
- # blobs without a converter, where b"" is returned.
- self.assertIsNone(cur.fetchone()[0])
-
-
-class ColNamesTests(unittest.TestCase):
- def setUp(self):
- self.con = sqlite.connect(":memory:", detect_types=sqlite.PARSE_COLNAMES)
- self.cur = self.con.cursor()
- self.cur.execute("create table test(x foo)")
-
- sqlite.converters["FOO"] = lambda x: "[%s]" % x.decode("ascii")
- sqlite.converters["BAR"] = lambda x: "<%s>" % x.decode("ascii")
- sqlite.converters["EXC"] = lambda x: 5/0
- sqlite.converters["B1B1"] = lambda x: "MARKER"
-
- def tearDown(self):
- del sqlite.converters["FOO"]
- del sqlite.converters["BAR"]
- del sqlite.converters["EXC"]
- del sqlite.converters["B1B1"]
- self.cur.close()
- self.con.close()
-
- def test_decl_type_not_used(self):
- """
- Assures that the declared type is not used when PARSE_DECLTYPES
- is not set.
- """
- self.cur.execute("insert into test(x) values (?)", ("xxx",))
- self.cur.execute("select x from test")
- val = self.cur.fetchone()[0]
- self.assertEqual(val, "xxx")
-
- def test_none(self):
- self.cur.execute("insert into test(x) values (?)", (None,))
- self.cur.execute("select x from test")
- val = self.cur.fetchone()[0]
- self.assertEqual(val, None)
-
- def test_col_name(self):
- self.cur.execute("insert into test(x) values (?)", ("xxx",))
- self.cur.execute('select x as "x y [bar]" from test')
- val = self.cur.fetchone()[0]
- self.assertEqual(val, "<xxx>")
-
- # Check if the stripping of colnames works. Everything after the first
- # '[' (and the preceding space) should be stripped.
- self.assertEqual(self.cur.description[0][0], "x y")
-
- def test_case_in_converter_name(self):
- self.cur.execute("select 'other' as \"x [b1b1]\"")
- val = self.cur.fetchone()[0]
- self.assertEqual(val, "MARKER")
-
- def test_cursor_description_no_row(self):
- """
- cursor.description should at least provide the column name(s), even if
- no row returned.
- """
- self.cur.execute("select * from test where 0 = 1")
- self.assertEqual(self.cur.description[0][0], "x")
-
- def test_cursor_description_insert(self):
- self.cur.execute("insert into test values (1)")
- self.assertIsNone(self.cur.description)
-
-
-@unittest.skipIf(sqlite.sqlite_version_info < (3, 8, 3), "CTEs not supported")
-class CommonTableExpressionTests(unittest.TestCase):
-
- def setUp(self):
- self.con = sqlite.connect(":memory:")
- self.cur = self.con.cursor()
- self.cur.execute("create table test(x foo)")
-
- def tearDown(self):
- self.cur.close()
- self.con.close()
-
- def test_cursor_description_cte_simple(self):
- self.cur.execute("with one as (select 1) select * from one")
- self.assertIsNotNone(self.cur.description)
- self.assertEqual(self.cur.description[0][0], "1")
-
- def test_cursor_description_cte_multiple_columns(self):
- self.cur.execute("insert into test values(1)")
- self.cur.execute("insert into test values(2)")
- self.cur.execute("with testCTE as (select * from test) select * from testCTE")
- self.assertIsNotNone(self.cur.description)
- self.assertEqual(self.cur.description[0][0], "x")
-
- def test_cursor_description_cte(self):
- self.cur.execute("insert into test values (1)")
- self.cur.execute("with bar as (select * from test) select * from test where x = 1")
- self.assertIsNotNone(self.cur.description)
- self.assertEqual(self.cur.description[0][0], "x")
- self.cur.execute("with bar as (select * from test) select * from test where x = 2")
- self.assertIsNotNone(self.cur.description)
- self.assertEqual(self.cur.description[0][0], "x")
-
-
-class ObjectAdaptationTests(unittest.TestCase):
- def cast(obj):
- return float(obj)
- cast = staticmethod(cast)
-
- def setUp(self):
- self.con = sqlite.connect(":memory:")
- try:
- del sqlite.adapters[int]
- except:
- pass
- sqlite.register_adapter(int, ObjectAdaptationTests.cast)
- self.cur = self.con.cursor()
-
- def tearDown(self):
- del sqlite.adapters[(int, sqlite.PrepareProtocol)]
- self.cur.close()
- self.con.close()
-
- def test_caster_is_used(self):
- self.cur.execute("select ?", (4,))
- val = self.cur.fetchone()[0]
- self.assertEqual(type(val), float)
-
- def test_missing_adapter(self):
- with self.assertRaises(sqlite.ProgrammingError):
- sqlite.adapt(1.) # No float adapter registered
-
- def test_missing_protocol(self):
- with self.assertRaises(sqlite.ProgrammingError):
- sqlite.adapt(1, None)
-
- def test_defect_proto(self):
- class DefectProto():
- def __adapt__(self):
- return None
- with self.assertRaises(sqlite.ProgrammingError):
- sqlite.adapt(1., DefectProto)
-
- def test_defect_self_adapt(self):
- class DefectSelfAdapt(float):
- def __conform__(self, _):
- return None
- with self.assertRaises(sqlite.ProgrammingError):
- sqlite.adapt(DefectSelfAdapt(1.))
-
- def test_custom_proto(self):
- class CustomProto():
- def __adapt__(self):
- return "adapted"
- self.assertEqual(sqlite.adapt(1., CustomProto), "adapted")
-
- def test_adapt(self):
- val = 42
- self.assertEqual(float(val), sqlite.adapt(val))
-
- def test_adapt_alt(self):
- alt = "other"
- self.assertEqual(alt, sqlite.adapt(1., None, alt))
-
-
-@unittest.skipUnless(zlib, "requires zlib")
-class BinaryConverterTests(unittest.TestCase):
- def convert(s):
- return zlib.decompress(s)
- convert = staticmethod(convert)
-
- def setUp(self):
- self.con = sqlite.connect(":memory:", detect_types=sqlite.PARSE_COLNAMES)
- sqlite.register_converter("bin", BinaryConverterTests.convert)
-
- def tearDown(self):
- self.con.close()
-
- def test_binary_input_for_converter(self):
- testdata = b"abcdefg" * 10
- result = self.con.execute('select ? as "x [bin]"', (memoryview(zlib.compress(testdata)),)).fetchone()[0]
- self.assertEqual(testdata, result)
-
-class DateTimeTests(unittest.TestCase):
- def setUp(self):
- self.con = sqlite.connect(":memory:", detect_types=sqlite.PARSE_DECLTYPES)
- self.cur = self.con.cursor()
- self.cur.execute("create table test(d date, ts timestamp)")
-
- def tearDown(self):
- self.cur.close()
- self.con.close()
-
- def test_sqlite_date(self):
- d = sqlite.Date(2004, 2, 14)
- self.cur.execute("insert into test(d) values (?)", (d,))
- self.cur.execute("select d from test")
- d2 = self.cur.fetchone()[0]
- self.assertEqual(d, d2)
-
- def test_sqlite_timestamp(self):
- ts = sqlite.Timestamp(2004, 2, 14, 7, 15, 0)
- self.cur.execute("insert into test(ts) values (?)", (ts,))
- self.cur.execute("select ts from test")
- ts2 = self.cur.fetchone()[0]
- self.assertEqual(ts, ts2)
-
- def test_sql_timestamp(self):
- now = datetime.datetime.utcnow()
- self.cur.execute("insert into test(ts) values (current_timestamp)")
- self.cur.execute("select ts from test")
- ts = self.cur.fetchone()[0]
- self.assertEqual(type(ts), datetime.datetime)
- self.assertEqual(ts.year, now.year)
-
- def test_date_time_sub_seconds(self):
- ts = sqlite.Timestamp(2004, 2, 14, 7, 15, 0, 500000)
- self.cur.execute("insert into test(ts) values (?)", (ts,))
- self.cur.execute("select ts from test")
- ts2 = self.cur.fetchone()[0]
- self.assertEqual(ts, ts2)
-
- def test_date_time_sub_seconds_floating_point(self):
- ts = sqlite.Timestamp(2004, 2, 14, 7, 15, 0, 510241)
- self.cur.execute("insert into test(ts) values (?)", (ts,))
- self.cur.execute("select ts from test")
- ts2 = self.cur.fetchone()[0]
- self.assertEqual(ts, ts2)
-
-
-if __name__ == "__main__":
- unittest.main()
diff --git a/Lib/sqlite3/test/test_userfunctions.py b/Lib/sqlite3/test/test_userfunctions.py
deleted file mode 100644
index ad40847..0000000
--- a/Lib/sqlite3/test/test_userfunctions.py
+++ /dev/null
@@ -1,665 +0,0 @@
-# pysqlite2/test/userfunctions.py: tests for user-defined functions and
-# aggregates.
-#
-# Copyright (C) 2005-2007 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 functools
-import gc
-import io
-import sys
-import unittest
-import unittest.mock
-import sqlite3 as sqlite
-
-from test.support import bigmemtest
-
-
-def with_tracebacks(strings, traceback=True):
- """Convenience decorator for testing callback tracebacks."""
- if traceback:
- strings.append('Traceback')
-
- def decorator(func):
- @functools.wraps(func)
- def wrapper(self, *args, **kwargs):
- # First, run the test with traceback enabled.
- with check_tracebacks(self, strings):
- func(self, *args, **kwargs)
-
- # Then run the test with traceback disabled.
- func(self, *args, **kwargs)
- return wrapper
- return decorator
-
-@contextlib.contextmanager
-def check_tracebacks(self, strings):
- """Convenience context manager for testing callback tracebacks."""
- sqlite.enable_callback_tracebacks(True)
- try:
- buf = io.StringIO()
- with contextlib.redirect_stderr(buf):
- yield
- tb = buf.getvalue()
- for s in strings:
- self.assertIn(s, tb)
- finally:
- sqlite.enable_callback_tracebacks(False)
-
-def func_returntext():
- return "foo"
-def func_returntextwithnull():
- return "1\x002"
-def func_returnunicode():
- return "bar"
-def func_returnint():
- return 42
-def func_returnfloat():
- return 3.14
-def func_returnnull():
- return None
-def func_returnblob():
- return b"blob"
-def func_returnlonglong():
- return 1<<31
-def func_raiseexception():
- 5/0
-def func_memoryerror():
- raise 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
-
- def finalize(self):
- return 1
-
-class AggrNoFinalize:
- def __init__(self):
- pass
-
- def step(self, x):
- pass
-
-class AggrExceptionInInit:
- def __init__(self):
- 5/0
-
- def step(self, x):
- pass
-
- def finalize(self):
- pass
-
-class AggrExceptionInStep:
- def __init__(self):
- pass
-
- def step(self, x):
- 5/0
-
- def finalize(self):
- return 42
-
-class AggrExceptionInFinalize:
- def __init__(self):
- pass
-
- def step(self, x):
- pass
-
- def finalize(self):
- 5/0
-
-class AggrCheckType:
- def __init__(self):
- self.val = None
-
- def step(self, whichType, val):
- theType = {"str": str, "int": int, "float": float, "None": type(None),
- "blob": bytes}
- self.val = int(theType[whichType] is type(val))
-
- def finalize(self):
- return self.val
-
-class AggrCheckTypes:
- def __init__(self):
- self.val = 0
-
- def step(self, whichType, *vals):
- theType = {"str": str, "int": int, "float": float, "None": type(None),
- "blob": bytes}
- for val in vals:
- self.val += int(theType[whichType] is type(val))
-
- def finalize(self):
- return self.val
-
-class AggrSum:
- def __init__(self):
- self.val = 0.0
-
- def step(self, val):
- self.val += val
-
- def finalize(self):
- return self.val
-
-class AggrText:
- def __init__(self):
- self.txt = ""
- def step(self, txt):
- self.txt = self.txt + txt
- def finalize(self):
- return self.txt
-
-
-class FunctionTests(unittest.TestCase):
- def setUp(self):
- self.con = sqlite.connect(":memory:")
-
- self.con.create_function("returntext", 0, func_returntext)
- self.con.create_function("returntextwithnull", 0, func_returntextwithnull)
- self.con.create_function("returnunicode", 0, func_returnunicode)
- self.con.create_function("returnint", 0, func_returnint)
- self.con.create_function("returnfloat", 0, func_returnfloat)
- 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("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.execute("create table test(t text)")
-
- def tearDown(self):
- self.con.close()
-
- def test_func_error_on_create(self):
- with self.assertRaises(sqlite.OperationalError):
- self.con.create_function("bla", -100, lambda x: 2*x)
-
- def test_func_ref_count(self):
- def getfunc():
- def f():
- return 1
- return f
- f = getfunc()
- globals()["foo"] = f
- # self.con.create_function("reftest", 0, getfunc())
- self.con.create_function("reftest", 0, f)
- cur = self.con.cursor()
- cur.execute("select reftest()")
-
- def test_func_return_text(self):
- cur = self.con.cursor()
- cur.execute("select returntext()")
- val = cur.fetchone()[0]
- self.assertEqual(type(val), str)
- self.assertEqual(val, "foo")
-
- def test_func_return_text_with_null_char(self):
- cur = self.con.cursor()
- res = cur.execute("select returntextwithnull()").fetchone()[0]
- self.assertEqual(type(res), str)
- self.assertEqual(res, "1\x002")
-
- def test_func_return_unicode(self):
- cur = self.con.cursor()
- cur.execute("select returnunicode()")
- val = cur.fetchone()[0]
- self.assertEqual(type(val), str)
- self.assertEqual(val, "bar")
-
- def test_func_return_int(self):
- cur = self.con.cursor()
- cur.execute("select returnint()")
- val = cur.fetchone()[0]
- self.assertEqual(type(val), int)
- self.assertEqual(val, 42)
-
- def test_func_return_float(self):
- cur = self.con.cursor()
- cur.execute("select returnfloat()")
- val = cur.fetchone()[0]
- self.assertEqual(type(val), float)
- if val < 3.139 or val > 3.141:
- self.fail("wrong value")
-
- def test_func_return_null(self):
- cur = self.con.cursor()
- cur.execute("select returnnull()")
- val = cur.fetchone()[0]
- self.assertEqual(type(val), type(None))
- self.assertEqual(val, None)
-
- def test_func_return_blob(self):
- cur = self.con.cursor()
- cur.execute("select returnblob()")
- val = cur.fetchone()[0]
- self.assertEqual(type(val), bytes)
- self.assertEqual(val, b"blob")
-
- def test_func_return_long_long(self):
- cur = self.con.cursor()
- cur.execute("select returnlonglong()")
- val = cur.fetchone()[0]
- self.assertEqual(val, 1<<31)
-
- @with_tracebacks(['func_raiseexception', '5/0', 'ZeroDivisionError'])
- def test_func_exception(self):
- cur = self.con.cursor()
- with self.assertRaises(sqlite.OperationalError) as cm:
- cur.execute("select raiseexception()")
- cur.fetchone()
- self.assertEqual(str(cm.exception), 'user-defined function raised exception')
-
- @with_tracebacks(['func_memoryerror', 'MemoryError'])
- def test_func_memory_error(self):
- cur = self.con.cursor()
- with self.assertRaises(MemoryError):
- cur.execute("select memoryerror()")
- cur.fetchone()
-
- @with_tracebacks(['func_overflowerror', 'OverflowError'])
- def test_func_overflow_error(self):
- cur = self.con.cursor()
- with self.assertRaises(sqlite.DataError):
- 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))
- val = cur.fetchone()[0]
- self.assertEqual(val, 2)
-
- def test_empty_blob(self):
- cur = self.con.execute("select isblob(x'')")
- self.assertTrue(cur.fetchone()[0])
-
- # Regarding deterministic functions:
- #
- # Between 3.8.3 and 3.15.0, deterministic functions were only used to
- # optimize inner loops, so for those versions we can only test if the
- # sqlite machinery has factored out a call or not. From 3.15.0 and onward,
- # deterministic functions were permitted in WHERE clauses of partial
- # indices, which allows testing based on syntax, iso. the query optimizer.
- @unittest.skipIf(sqlite.sqlite_version_info < (3, 8, 3), "Requires SQLite 3.8.3 or higher")
- def test_func_non_deterministic(self):
- mock = unittest.mock.Mock(return_value=None)
- self.con.create_function("nondeterministic", 0, mock, deterministic=False)
- if sqlite.sqlite_version_info < (3, 15, 0):
- self.con.execute("select nondeterministic() = nondeterministic()")
- self.assertEqual(mock.call_count, 2)
- else:
- with self.assertRaises(sqlite.OperationalError):
- self.con.execute("create index t on test(t) where nondeterministic() is not null")
-
- @unittest.skipIf(sqlite.sqlite_version_info < (3, 8, 3), "Requires SQLite 3.8.3 or higher")
- def test_func_deterministic(self):
- mock = unittest.mock.Mock(return_value=None)
- self.con.create_function("deterministic", 0, mock, deterministic=True)
- if sqlite.sqlite_version_info < (3, 15, 0):
- self.con.execute("select deterministic() = deterministic()")
- self.assertEqual(mock.call_count, 1)
- else:
- try:
- self.con.execute("create index t on test(t) where deterministic() is not null")
- except sqlite.OperationalError:
- self.fail("Unexpected failure while creating partial index")
-
- @unittest.skipIf(sqlite.sqlite_version_info >= (3, 8, 3), "SQLite < 3.8.3 needed")
- def test_func_deterministic_not_supported(self):
- with self.assertRaises(sqlite.NotSupportedError):
- self.con.create_function("deterministic", 0, int, deterministic=True)
-
- def test_func_deterministic_keyword_only(self):
- with self.assertRaises(TypeError):
- self.con.create_function("deterministic", 0, int, True)
-
- def test_function_destructor_via_gc(self):
- # See bpo-44304: The destructor of the user function can
- # crash if is called without the GIL from the gc functions
- dest = sqlite.connect(':memory:')
- def md5sum(t):
- return
-
- dest.create_function("md5", 1, md5sum)
- x = dest("create table lang (name, first_appeared)")
- del md5sum, dest
-
- y = [x]
- y.append(y)
-
- del x,y
- gc.collect()
-
- def test_func_return_too_large_int(self):
- cur = self.con.cursor()
- for value in 2**63, -2**63-1, 2**64:
- self.con.create_function("largeint", 0, lambda value=value: value)
- with check_tracebacks(self, ['OverflowError']):
- with self.assertRaises(sqlite.DataError):
- cur.execute("select largeint()")
-
- def test_func_return_text_with_surrogates(self):
- cur = self.con.cursor()
- self.con.create_function("pychr", 1, chr)
- for value in 0xd8ff, 0xdcff:
- with check_tracebacks(self,
- ['UnicodeEncodeError', 'surrogates not allowed']):
- with self.assertRaises(sqlite.OperationalError):
- cur.execute("select pychr(?)", (value,))
-
- @unittest.skipUnless(sys.maxsize > 2**32, 'requires 64bit platform')
- @bigmemtest(size=2**31, memuse=3, dry_run=False)
- def test_func_return_too_large_text(self, size):
- cur = self.con.cursor()
- for size in 2**31-1, 2**31:
- self.con.create_function("largetext", 0, lambda size=size: "b" * size)
- with self.assertRaises(sqlite.DataError):
- cur.execute("select largetext()")
-
- @unittest.skipUnless(sys.maxsize > 2**32, 'requires 64bit platform')
- @bigmemtest(size=2**31, memuse=2, dry_run=False)
- def test_func_return_too_large_blob(self, size):
- cur = self.con.cursor()
- for size in 2**31-1, 2**31:
- self.con.create_function("largeblob", 0, lambda size=size: b"b" * size)
- with self.assertRaises(sqlite.DataError):
- cur.execute("select largeblob()")
-
-
-class AggregateTests(unittest.TestCase):
- def setUp(self):
- self.con = sqlite.connect(":memory:")
- cur = self.con.cursor()
- cur.execute("""
- create table test(
- t text,
- i integer,
- f float,
- n,
- b blob
- )
- """)
- cur.execute("insert into test(t, i, f, n, b) values (?, ?, ?, ?, ?)",
- ("foo", 5, 3.14, None, memoryview(b"blob"),))
-
- self.con.create_aggregate("nostep", 1, AggrNoStep)
- self.con.create_aggregate("nofinalize", 1, AggrNoFinalize)
- self.con.create_aggregate("excInit", 1, AggrExceptionInInit)
- self.con.create_aggregate("excStep", 1, AggrExceptionInStep)
- self.con.create_aggregate("excFinalize", 1, AggrExceptionInFinalize)
- self.con.create_aggregate("checkType", 2, AggrCheckType)
- self.con.create_aggregate("checkTypes", -1, AggrCheckTypes)
- self.con.create_aggregate("mysum", 1, AggrSum)
- self.con.create_aggregate("aggtxt", 1, AggrText)
-
- def tearDown(self):
- #self.cur.close()
- #self.con.close()
- pass
-
- def test_aggr_error_on_create(self):
- with self.assertRaises(sqlite.OperationalError):
- self.con.create_function("bla", -100, AggrSum)
-
- def test_aggr_no_step(self):
- cur = self.con.cursor()
- with self.assertRaises(AttributeError) as cm:
- cur.execute("select nostep(t) from test")
- self.assertEqual(str(cm.exception), "'AggrNoStep' object has no attribute 'step'")
-
- def test_aggr_no_finalize(self):
- cur = self.con.cursor()
- with self.assertRaises(sqlite.OperationalError) as cm:
- cur.execute("select nofinalize(t) from test")
- val = cur.fetchone()[0]
- self.assertEqual(str(cm.exception), "user-defined aggregate's 'finalize' method raised error")
-
- @with_tracebacks(['__init__', '5/0', 'ZeroDivisionError'])
- def test_aggr_exception_in_init(self):
- cur = self.con.cursor()
- with self.assertRaises(sqlite.OperationalError) as cm:
- cur.execute("select excInit(t) from test")
- val = cur.fetchone()[0]
- self.assertEqual(str(cm.exception), "user-defined aggregate's '__init__' method raised error")
-
- @with_tracebacks(['step', '5/0', 'ZeroDivisionError'])
- def test_aggr_exception_in_step(self):
- cur = self.con.cursor()
- with self.assertRaises(sqlite.OperationalError) as cm:
- cur.execute("select excStep(t) from test")
- val = cur.fetchone()[0]
- self.assertEqual(str(cm.exception), "user-defined aggregate's 'step' method raised error")
-
- @with_tracebacks(['finalize', '5/0', 'ZeroDivisionError'])
- def test_aggr_exception_in_finalize(self):
- cur = self.con.cursor()
- with self.assertRaises(sqlite.OperationalError) as cm:
- cur.execute("select excFinalize(t) from test")
- val = cur.fetchone()[0]
- self.assertEqual(str(cm.exception), "user-defined aggregate's 'finalize' method raised error")
-
- def test_aggr_check_param_str(self):
- cur = self.con.cursor()
- cur.execute("select checkTypes('str', ?, ?)", ("foo", str()))
- val = cur.fetchone()[0]
- self.assertEqual(val, 2)
-
- def test_aggr_check_param_int(self):
- cur = self.con.cursor()
- cur.execute("select checkType('int', ?)", (42,))
- val = cur.fetchone()[0]
- self.assertEqual(val, 1)
-
- def test_aggr_check_params_int(self):
- cur = self.con.cursor()
- cur.execute("select checkTypes('int', ?, ?)", (42, 24))
- val = cur.fetchone()[0]
- self.assertEqual(val, 2)
-
- def test_aggr_check_param_float(self):
- cur = self.con.cursor()
- cur.execute("select checkType('float', ?)", (3.14,))
- val = cur.fetchone()[0]
- self.assertEqual(val, 1)
-
- def test_aggr_check_param_none(self):
- cur = self.con.cursor()
- cur.execute("select checkType('None', ?)", (None,))
- val = cur.fetchone()[0]
- self.assertEqual(val, 1)
-
- def test_aggr_check_param_blob(self):
- cur = self.con.cursor()
- cur.execute("select checkType('blob', ?)", (memoryview(b"blob"),))
- val = cur.fetchone()[0]
- self.assertEqual(val, 1)
-
- def test_aggr_check_aggr_sum(self):
- cur = self.con.cursor()
- cur.execute("delete from test")
- cur.executemany("insert into test(i) values (?)", [(10,), (20,), (30,)])
- cur.execute("select mysum(i) from test")
- val = cur.fetchone()[0]
- self.assertEqual(val, 60)
-
- def test_aggr_no_match(self):
- cur = self.con.execute("select mysum(i) from (select 1 as i) where i == 0")
- val = cur.fetchone()[0]
- self.assertIsNone(val)
-
- def test_aggr_text(self):
- cur = self.con.cursor()
- for txt in ["foo", "1\x002"]:
- with self.subTest(txt=txt):
- cur.execute("select aggtxt(?) from test", (txt,))
- val = cur.fetchone()[0]
- self.assertEqual(val, txt)
-
-
-class AuthorizerTests(unittest.TestCase):
- @staticmethod
- def authorizer_cb(action, arg1, arg2, dbname, source):
- if action != sqlite.SQLITE_SELECT:
- return sqlite.SQLITE_DENY
- if arg2 == 'c2' or arg1 == 't2':
- return sqlite.SQLITE_DENY
- return sqlite.SQLITE_OK
-
- def setUp(self):
- self.con = sqlite.connect(":memory:")
- self.con.executescript("""
- create table t1 (c1, c2);
- create table t2 (c1, c2);
- insert into t1 (c1, c2) values (1, 2);
- insert into t2 (c1, c2) values (4, 5);
- """)
-
- # For our security test:
- self.con.execute("select c2 from t2")
-
- self.con.set_authorizer(self.authorizer_cb)
-
- def tearDown(self):
- pass
-
- def test_table_access(self):
- with self.assertRaises(sqlite.DatabaseError) as cm:
- self.con.execute("select * from t2")
- self.assertIn('prohibited', str(cm.exception))
-
- def test_column_access(self):
- with self.assertRaises(sqlite.DatabaseError) as cm:
- self.con.execute("select c2 from t1")
- self.assertIn('prohibited', str(cm.exception))
-
- def test_clear_authorizer(self):
- self.con.set_authorizer(None)
- self.con.execute("select * from t2")
- self.con.execute("select c2 from t1")
-
-
-class AuthorizerRaiseExceptionTests(AuthorizerTests):
- @staticmethod
- def authorizer_cb(action, arg1, arg2, dbname, source):
- if action != sqlite.SQLITE_SELECT:
- raise ValueError
- if arg2 == 'c2' or arg1 == 't2':
- raise ValueError
- return sqlite.SQLITE_OK
-
- @with_tracebacks(['authorizer_cb', 'ValueError'])
- def test_table_access(self):
- super().test_table_access()
-
- @with_tracebacks(['authorizer_cb', 'ValueError'])
- def test_column_access(self):
- super().test_table_access()
-
-class AuthorizerIllegalTypeTests(AuthorizerTests):
- @staticmethod
- def authorizer_cb(action, arg1, arg2, dbname, source):
- if action != sqlite.SQLITE_SELECT:
- return 0.0
- if arg2 == 'c2' or arg1 == 't2':
- return 0.0
- return sqlite.SQLITE_OK
-
-class AuthorizerLargeIntegerTests(AuthorizerTests):
- @staticmethod
- def authorizer_cb(action, arg1, arg2, dbname, source):
- if action != sqlite.SQLITE_SELECT:
- return 2**32
- if arg2 == 'c2' or arg1 == 't2':
- return 2**32
- return sqlite.SQLITE_OK
-
-
-if __name__ == "__main__":
- unittest.main()