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.assertIn(sqlite.threadsafety, {0, 1, 3},

"threadsafety is %d, should be 0, 1 or 3" %

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.assertIsSubclass(sqlite.Warning, Exception)

def test_error(self):

self.assertIsSubclass(sqlite.Error, Exception)

def test_interface_error(self):

self.assertIsSubclass(sqlite.InterfaceError, sqlite.Error)

def test_database_error(self):

self.assertIsSubclass(sqlite.DatabaseError, sqlite.Error)

def test_data_error(self):

self.assertIsSubclass(sqlite.DataError, sqlite.DatabaseError)

def test_operational_error(self):

self.assertIsSubclass(sqlite.OperationalError, sqlite.DatabaseError)

def test_integrity_error(self):

self.assertIsSubclass(sqlite.IntegrityError, sqlite.DatabaseError)

def test_internal_error(self):

self.assertIsSubclass(sqlite.InternalError, sqlite.DatabaseError)

def test_programming_error(self):

self.assertIsSubclass(sqlite.ProgrammingError, sqlite.DatabaseError)

def test_not_supported_error(self):

self.assertIsSubclass(sqlite.NotSupportedError, sqlite.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_LIMIT_WORKER_THREADS",

"SQLITE_LOCKED",

"SQLITE_MISMATCH",

"SQLITE_MISUSE",

"SQLITE_NOLFS",

"SQLITE_NOMEM",

"SQLITE_NOTADB",

"SQLITE_NOTFOUND",

"SQLITE_NOTICE",

"SQLITE_OK",

"SQLITE_PERM",

"SQLITE_PRAGMA",

"SQLITE_PROTOCOL",

"SQLITE_RANGE",

"SQLITE_READ",

"SQLITE_READONLY",

"SQLITE_RECURSIVE",

"SQLITE_REINDEX",

"SQLITE_ROW",

"SQLITE_SAVEPOINT",

"SQLITE_SCHEMA",

"SQLITE_SELECT",

"SQLITE_TOOBIG",

"SQLITE_TRANSACTION",

"SQLITE_UPDATE",

"SQLITE_WARNING",

# Run-time limit categories

"SQLITE_LIMIT_LENGTH",

"SQLITE_LIMIT_SQL_LENGTH",

"SQLITE_LIMIT_COLUMN",

"SQLITE_LIMIT_EXPR_DEPTH",

"SQLITE_LIMIT_COMPOUND_SELECT",

"SQLITE_LIMIT_VDBE_OP",

"SQLITE_LIMIT_FUNCTION_ARG",

"SQLITE_LIMIT_ATTACHED",

"SQLITE_LIMIT_LIKE_PATTERN_LENGTH",

"SQLITE_LIMIT_VARIABLE_NUMBER",

"SQLITE_LIMIT_TRIGGER_DEPTH",

]

consts += ["PARSE_DECLTYPES", "PARSE_COLNAMES"]

# Extended result codes

consts += [

"SQLITE_ABORT_ROLLBACK",

"SQLITE_AUTH_USER",

"SQLITE_BUSY_RECOVERY",

"SQLITE_BUSY_SNAPSHOT",

"SQLITE_CANTOPEN_CONVPATH",

"SQLITE_CANTOPEN_FULLPATH",

"SQLITE_CANTOPEN_ISDIR",

"SQLITE_CANTOPEN_NOTEMPDIR",

"SQLITE_CONSTRAINT_CHECK",

"SQLITE_CONSTRAINT_COMMITHOOK",

"SQLITE_CONSTRAINT_FOREIGNKEY",

"SQLITE_CONSTRAINT_FUNCTION",

"SQLITE_CONSTRAINT_NOTNULL",

"SQLITE_CONSTRAINT_PRIMARYKEY",

"SQLITE_CONSTRAINT_ROWID",

"SQLITE_CONSTRAINT_TRIGGER",

"SQLITE_CONSTRAINT_UNIQUE",

"SQLITE_CONSTRAINT_VTAB",

"SQLITE_CORRUPT_VTAB",

"SQLITE_IOERR_ACCESS",

"SQLITE_IOERR_AUTH",

"SQLITE_IOERR_BLOCKED",

"SQLITE_IOERR_CHECKRESERVEDLOCK",

"SQLITE_IOERR_CLOSE",

"SQLITE_IOERR_CONVPATH",

"SQLITE_IOERR_DELETE",

"SQLITE_IOERR_DELETE_NOENT",

"SQLITE_IOERR_DIR_CLOSE",

"SQLITE_IOERR_DIR_FSYNC",

"SQLITE_IOERR_FSTAT",

"SQLITE_IOERR_FSYNC",

"SQLITE_IOERR_GETTEMPPATH",

"SQLITE_IOERR_LOCK",

"SQLITE_IOERR_MMAP",

"SQLITE_IOERR_NOMEM",

"SQLITE_IOERR_RDLOCK",

"SQLITE_IOERR_READ",

"SQLITE_IOERR_SEEK",

"SQLITE_IOERR_SHMLOCK",

"SQLITE_IOERR_SHMMAP",

"SQLITE_IOERR_SHMOPEN",

"SQLITE_IOERR_SHMSIZE",

"SQLITE_IOERR_SHORT_READ",

"SQLITE_IOERR_TRUNCATE",

"SQLITE_IOERR_UNLOCK",

"SQLITE_IOERR_VNODE",

"SQLITE_IOERR_WRITE",

"SQLITE_LOCKED_SHAREDCACHE",

"SQLITE_NOTICE_RECOVER_ROLLBACK",

"SQLITE_NOTICE_RECOVER_WAL",

"SQLITE_OK_LOAD_PERMANENTLY",

"SQLITE_READONLY_CANTLOCK",

"SQLITE_READONLY_DBMOVED",

"SQLITE_READONLY_RECOVERY",

"SQLITE_READONLY_ROLLBACK",

"SQLITE_WARNING_AUTOINDEX",

]

if sqlite.sqlite_version_info >= (3, 21, 0):

consts += [

"SQLITE_IOERR_BEGIN_ATOMIC",

"SQLITE_IOERR_COMMIT_ATOMIC",

"SQLITE_IOERR_ROLLBACK_ATOMIC",

]

if sqlite.sqlite_version_info >= (3, 22, 0):

consts += [

"SQLITE_ERROR_MISSING_COLLSEQ",

"SQLITE_ERROR_RETRY",

"SQLITE_READONLY_CANTINIT",

"SQLITE_READONLY_DIRECTORY",

]

if sqlite.sqlite_version_info >= (3, 24, 0):

consts += ["SQLITE_CORRUPT_SEQUENCE", "SQLITE_LOCKED_VTAB"]

if sqlite.sqlite_version_info >= (3, 25, 0):

consts += ["SQLITE_CANTOPEN_DIRTYWAL", "SQLITE_ERROR_SNAPSHOT"]

if sqlite.sqlite_version_info >= (3, 31, 0):

consts += [

"SQLITE_CANTOPEN_SYMLINK",

"SQLITE_CONSTRAINT_PINNED",

"SQLITE_OK_SYMLINK",

]

if sqlite.sqlite_version_info >= (3, 32, 0):

consts += [

"SQLITE_BUSY_TIMEOUT",

"SQLITE_CORRUPT_INDEX",

"SQLITE_IOERR_DATA",

]

if sqlite.sqlite_version_info >= (3, 34, 0):

consts.append("SQLITE_IOERR_CORRUPTFS")

for const in consts:

with self.subTest(const=const):

self.assertHasAttr(sqlite, const)

def test_error_code_on_exception(self):

err_msg = "unable to open database file"

if sys.platform.startswith("win"):

err_code = sqlite.SQLITE_CANTOPEN_ISDIR

else:

err_code = sqlite.SQLITE_CANTOPEN

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, err_code)

self.assertStartsWith(e.sqlite_errorname, "SQLITE_CANTOPEN")

def test_extended_error_code_on_exception(self):

with memory_database() as con:

with con:

con.execute("create table t(t integer check(t > 0))")

errmsg = "constraint failed"

with self.assertRaisesRegex(sqlite.IntegrityError, errmsg) as cm:

con.execute("insert into t values(-1)")

exc = cm.exception

self.assertEqual(exc.sqlite_errorcode,

sqlite.SQLITE_CONSTRAINT_CHECK)

self.assertEqual(exc.sqlite_errorname, "SQLITE_CONSTRAINT_CHECK")

def test_disallow_instantiation(self):

with memory_database() as cx:

check_disallow_instantiation(self, type(cx("select 1")))

check_disallow_instantiation(self, sqlite.Blob)

def test_complete_statement(self):

self.assertFalse(sqlite.complete_statement("select t"))

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",))

cu.close()

def tearDown(self):

self.cx.close()

def test_commit(self):

self.cx.commit()

def test_commit_after_no_changes(self):

"""

self.cx.commit()

self.cx.commit()

def test_rollback(self):

self.cx.rollback()

def test_rollback_after_no_changes(self):

"""

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):

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.

with memory_database() as cx:

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)

cu.close()

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.assertHasAttr(self.cx, exc)

self.assertIs(getattr(sqlite, exc), getattr(self.cx, exc))

def test_interrupt_on_closed_db(self):

self.cx.close()

with self.assertRaises(sqlite.ProgrammingError):

self.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)

def test_connection_limits(self):

category = sqlite.SQLITE_LIMIT_SQL_LENGTH

saved_limit = self.cx.getlimit(category)

try:

new_limit = 10

prev_limit = self.cx.setlimit(category, new_limit)

self.assertEqual(saved_limit, prev_limit)

self.assertEqual(self.cx.getlimit(category), new_limit)

msg = "query string is too large"

self.assertRaisesRegex(sqlite.DataError, msg,

self.cx.execute, "select 1 as '16'")

finally: # restore saved limit

self.cx.setlimit(category, saved_limit)

def test_connection_bad_limit_category(self):

msg = "'category' is out of bounds"

cat = 1111

self.assertRaisesRegex(sqlite.ProgrammingError, msg,

self.cx.getlimit, cat)

self.assertRaisesRegex(sqlite.ProgrammingError, msg,

self.cx.setlimit, cat, 0)

def test_connection_init_bad_isolation_level(self):

msg = (

"isolation_level string must be '', 'DEFERRED', 'IMMEDIATE', or "

"'EXCLUSIVE'"

)

levels = (

"BOGUS",

" ",

"DEFERRE",

"IMMEDIAT",

"EXCLUSIV",

"DEFERREDS",

"IMMEDIATES",

"EXCLUSIVES",

)

for level in levels:

with self.subTest(level=level):

with self.assertRaisesRegex(ValueError, msg):

memory_database(isolation_level=level)

with memory_database() as cx:

with self.assertRaisesRegex(ValueError, msg):

cx.isolation_level = level

# Check that the default level is not changed

self.assertEqual(cx.isolation_level, "")

def test_connection_init_good_isolation_levels(self):

for level in ("", "DEFERRED", "IMMEDIATE", "EXCLUSIVE", None):

with self.subTest(level=level):

with memory_database(isolation_level=level) as cx:

self.assertEqual(cx.isolation_level, level)

with memory_database() as cx:

self.assertEqual(cx.isolation_level, "")

cx.isolation_level = level

self.assertEqual(cx.isolation_level, level)

def test_connection_reinit(self):

with memory_database() as cx:

cx.text_factory = bytes

cx.row_factory = sqlite.Row

cu = cx.cursor()

cu.execute("create table foo (bar)")

cu.executemany("insert into foo (bar) values (?)",

((str(v),) for v in range(4)))

cu.execute("select bar from foo")

rows = [r for r in cu.fetchmany(2)]

self.assertTrue(all(isinstance(r, sqlite.Row) for r in rows))

self.assertEqual([r[0] for r in rows], [b"0", b"1"])

cx.__init__(":memory:")

cx.execute("create table foo (bar)")

cx.executemany("insert into foo (bar) values (?)",

((v,) for v in ("a", "b", "c", "d")))

# This uses the old database, old row factory, but new text factory

rows = [r for r in cu.fetchall()]

self.assertTrue(all(isinstance(r, sqlite.Row) for r in rows))

self.assertEqual([r[0] for r in rows], ["2", "3"])

cu.close()

def test_connection_bad_reinit(self):

cx = sqlite.connect(":memory:")

with cx:

cx.execute("create table t(t)")

with temp_dir() as db:

self.assertRaisesRegex(sqlite.OperationalError,

"unable to open database file",

cx.__init__, db)

self.assertRaisesRegex(sqlite.ProgrammingError,

"Base Connection.__init__ not called",

cx.executemany, "insert into t values(?)",

((v,) for v in range(3)))

def test_connection_config(self):

op = sqlite.SQLITE_DBCONFIG_ENABLE_FKEY

with memory_database() as cx:

with self.assertRaisesRegex(ValueError, "unknown"):

cx.getconfig(-1)

# Toggle and verify.

old = cx.getconfig(op)

new = not old

cx.setconfig(op, new)

self.assertEqual(cx.getconfig(op), new)

cx.setconfig(op) # defaults to True

self.assertTrue(cx.getconfig(op))

# Check that foreign key support was actually enabled.

with cx:

cx.executescript("""

""")

with self.assertRaisesRegex(sqlite.IntegrityError, "constraint"):

cx.execute("insert into u values(0)")

@unittest.expectedFailure # TODO: RUSTPYTHON; deprecation warning not emitted for positional args

def test_connect_positional_arguments(self):

regex = (

r"Passing more than 1 positional argument to sqlite3.connect\(\)"

" is deprecated. Parameters 'timeout', 'detect_types', "

"'isolation_level', 'check_same_thread', 'factory', "

"'cached_statements' and 'uri' will become keyword-only "

"parameters in Python 3.15."

)

with self.assertWarnsRegex(DeprecationWarning, regex) as cm:

cx = sqlite.connect(":memory:", 1.0)

cx.close()

self.assertEqual(cm.filename, __file__)

@unittest.expectedFailure # TODO: RUSTPYTHON; ResourceWarning not emitted

def test_connection_resource_warning(self):

with self.assertWarns(ResourceWarning):

cx = sqlite.connect(":memory:")

del cx

gc_collect()

@unittest.expectedFailure # TODO: RUSTPYTHON; Connection signature inspection not working

def test_connection_signature(self):

from inspect import signature

sig = signature(self.cx)

def setUp(self):

self.cx = sqlite.Connection.__new__(sqlite.Connection)

@unittest.skip("TODO: RUSTPYTHON")

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",

def test_serialize_deserialize(self):

with memory_database() as cx:

with cx:

cx.execute("create table t(t)")

data = cx.serialize()

# Remove test table, verify that it was removed.

with cx:

cx.execute("drop table t")

regex = "no such table"

with self.assertRaisesRegex(sqlite.OperationalError, regex):

cx.execute("select t from t")

# Deserialize and verify that test table is restored.

cx.deserialize(data)

cx.execute("select t from t")

def test_deserialize_wrong_args(self):

dataset = (

(BufferError, memoryview(b"blob")[::2]),

(TypeError, []),

(TypeError, 1),

(TypeError, None),

)

for exc, arg in dataset:

with self.subTest(exc=exc, arg=arg):

with memory_database() as cx:

self.assertRaises(exc, cx.deserialize, arg)

def test_deserialize_corrupt_database(self):

with memory_database() as cx:

regex = "file is not a database"

with self.assertRaisesRegex(sqlite.DatabaseError, regex):

cx.deserialize(b"\0\1\3")

# SQLite does not generate an error until you try to query the

# deserialized database.

_sql = "create table test(id integer)"

def test_open_with_bytes_path(self):

path = os.fsencode(TESTFN)

self.addCleanup(unlink, path)

self.assertFalse(os.path.exists(path))

with contextlib.closing(sqlite.connect(path)) as cx:

self.assertTrue(os.path.exists(path))

cx.execute(self._sql)

def test_open_with_path_like_object(self):

""" Checks that we can successfully connect to a database using an object that

path = FakePath(TESTFN)

self.addCleanup(unlink, path)

self.assertFalse(os.path.exists(path))

with contextlib.closing(sqlite.connect(path)) as cx:

self.assertTrue(os.path.exists(path))

cx.execute(self._sql)

def get_undecodable_path(self):

path = TESTFN_UNDECODABLE

if not path:

self.skipTest("only works if there are undecodable paths")

try:

open(path, 'wb').close()

except OSError:

self.skipTest(f"can't create file with undecodable path {path!r}")

unlink(path)

return path

@unittest.skipIf(sys.platform == "win32", "skipped on Windows")

def test_open_with_undecodable_path(self):

path = self.get_undecodable_path()

self.addCleanup(unlink, path)

with contextlib.closing(sqlite.connect(path)) as cx:

self.assertTrue(os.path.exists(path))

cx.execute(self._sql)

def test_open_uri(self):

path = TESTFN

self.addCleanup(unlink, path)

uri = "file:" + urllib.parse.quote(os.fsencode(path))

self.assertFalse(os.path.exists(path))

with contextlib.closing(sqlite.connect(uri, uri=True)) as cx:

self.assertTrue(os.path.exists(path))

cx.execute(self._sql)

def test_open_unquoted_uri(self):

path = TESTFN

self.addCleanup(unlink, path)

uri = "file:" + path

self.assertFalse(os.path.exists(path))

with contextlib.closing(sqlite.connect(uri, uri=True)) as cx:

self.assertTrue(os.path.exists(path))

cx.execute(self._sql)

def test_open_uri_readonly(self):

path = TESTFN

self.addCleanup(unlink, path)

uri = "file:" + urllib.parse.quote(os.fsencode(path)) + "?mode=ro"

self.assertFalse(os.path.exists(path))

# Cannot create new DB

with self.assertRaises(sqlite.OperationalError):

sqlite.connect(uri, uri=True)

self.assertFalse(os.path.exists(path))

sqlite.connect(path).close()

self.assertTrue(os.path.exists(path))

# Cannot modify new DB

with contextlib.closing(sqlite.connect(uri, uri=True)) as cx:

with self.assertRaises(sqlite.OperationalError):

cx.execute(self._sql)

@unittest.skipIf(sys.platform == "win32", "skipped on Windows")

def test_open_undecodable_uri(self):

path = self.get_undecodable_path()

self.addCleanup(unlink, path)

uri = "file:" + urllib.parse.quote(path)

with contextlib.closing(sqlite.connect(uri, uri=True)) as cx:

self.assertTrue(os.path.exists(path))

cx.execute(self._sql)

@unittest.expectedFailure # TODO: RUSTPYTHON

def test_factory_database_arg(self):

def factory(database, *args, **kwargs):

nonlocal database_arg

database_arg = database

return sqlite.Connection(":memory:", *args, **kwargs)

for database in (TESTFN, os.fsencode(TESTFN),

FakePath(TESTFN), FakePath(os.fsencode(TESTFN))):

database_arg = None

sqlite.connect(database, factory=factory).close()

self.assertEqual(database_arg, database)

def test_database_keyword(self):

with contextlib.closing(sqlite.connect(database=":memory:")) as cx:

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_multiple_statements(self):

msg = "You can only execute one statement at a time"

dataset = (

"select 1; select 2",

"select 1; // c++ comments are not allowed",

"select 1; *not a comment",

"select 1; -*not a comment",

"select 1; /* */ a",

"select 1; /**/a",

"select 1; -",

"select 1; /",

"select 1; -\n- select 2",

"""select 1;

""",

)

for query in dataset:

with self.subTest(query=query):

with self.assertRaisesRegex(sqlite.ProgrammingError, msg):

self.cu.execute(query)

def test_execute_with_appended_comments(self):

dataset = (

"select 1; -- foo bar",

"select 1; --",

"select 1; /*", # Unclosed comments ending in \0 are skipped.

"""

""",

)

for query in dataset:

with self.subTest(query=query):

self.cu.execute(query)

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(sqlite.ProgrammingError) 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_named_param_and_sequence(self):

dataset = (

("select :a", (1,)),

("select :a, ?, ?", (1, 2, 3)),

("select ?, :b, ?", (1, 2, 3)),

("select ?, ?, :c", (1, 2, 3)),

("select :a, :b, ?", (1, 2, 3)),

)

msg = "Binding.*is a named parameter"

for query, params in dataset:

with self.subTest(query=query, params=params):

with self.assertRaisesRegex(sqlite.ProgrammingError, msg) as cm:

self.cu.execute(query, params)

def test_execute_indexed_nameless_params(self):

# See gh-117995: "'?1' is considered a named placeholder"

for query, params, expected in (

("select ?1, ?2", (1, 2), (1, 2)),

("select ?2, ?1", (1, 2), (2, 1)),

):

with self.subTest(query=query, params=params):

with warnings.catch_warnings():

warnings.simplefilter("error", DeprecationWarning)

cu = self.cu.execute(query, params)

actual, = cu.fetchall()

self.assertEqual(actual, expected)

def test_execute_too_many_params(self):

category = sqlite.SQLITE_LIMIT_VARIABLE_NUMBER

msg = "too many SQL variables"

with cx_limit(self.cx, category=category, limit=1):

self.cu.execute("select * from test where id=?", (1,))

with self.assertRaisesRegex(sqlite.OperationalError, msg):

self.cu.execute("select * from test where id!=? and id!=?",

(1, 2))

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):

"""

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)

@unittest.skipIf(sqlite.sqlite_version_info < (3, 35, 0),

def test_rowcount_update_returning(self):

# gh-93421: rowcount is updated correctly for UPDATE...RETURNING queries

self.cu.execute("update test set name='bar' where name='foo' returning 1")

self.assertEqual(self.cu.fetchone()[0], 1)

self.assertEqual(self.cu.rowcount, 1)

def test_rowcount_prefixed_with_comment(self):

# gh-79579: rowcount is updated even if query is prefixed with comments

self.cu.execute("""

""")

self.assertEqual(self.cu.rowcount, 2)

self.cu.execute("""

""")

self.assertEqual(self.cu.rowcount, 1)

self.cu.execute("/* bar */ update test set name='bar' where name='foo'")

self.assertEqual(self.cu.rowcount, 3)

def test_rowcount_vaccuum(self):

data = ((1,), (2,), (3,))

self.cu.executemany("insert into test(income) values(?)", data)

self.assertEqual(self.cu.rowcount, 3)

self.cx.commit()

self.cu.execute("vacuum")

self.assertEqual(self.cu.rowcount, -1)

def test_total_changes(self):

self.cu.execute("insert into test(name) values ('foo')")

self.cu.execute("insert into test(name) values ('foo')")