test_dbapi.py

# 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 os
import sqlite3 as sqlite
import subprocess
import sys
import threading
import unittest
import urllib.parse

from test.support import (
    SHORT_TIMEOUT, check_disallow_instantiation,# requires_subprocess,
    #is_emscripten, is_wasi
# TODO: RUSTPYTHON
)
from test.support import threading_helper
# TODO: RUSTPYTHON
#from _testcapi import INT_MAX, ULLONG_MAX
from os import SEEK_SET, SEEK_CUR, SEEK_END
from test.support.os_helper import TESTFN, TESTFN_UNDECODABLE, unlink, temp_dir, FakePath


# Helper for temporary memory databases
def memory_database(*args, **kwargs):
    cx = sqlite.connect(":memory:", *args, **kwargs)
    return contextlib.closing(cx)


# Temporarily limit a database connection parameter
@contextlib.contextmanager
def cx_limit(cx, category=sqlite.SQLITE_LIMIT_SQL_LENGTH, limit=128):
    try:
        _prev = cx.setlimit(category, limit)
        yield limit
    finally:
        cx.setlimit(category, _prev)


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_deprecated_version(self):
        msg = "deprecated and will be removed in Python 3.14"
        for attr in "version", "version_info":
            with self.subTest(attr=attr):
                with self.assertWarnsRegex(DeprecationWarning, msg) as cm:
                    getattr(sqlite, attr)
                self.assertEqual(cm.filename,  __file__)
                with self.assertWarnsRegex(DeprecationWarning, msg) as cm:
                    getattr(sqlite.dbapi2, attr)
                self.assertEqual(cm.filename,  __file__)

    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.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_RANGE",
            "SQLITE_READ",
            "SQLITE_READONLY",
            "SQLITE_REINDEX",
            "SQLITE_ROW",
            "SQLITE_SAVEPOINT",
            "SQLITE_SCHEMA",
            "SQLITE_SELECT",
            "SQLITE_TOOBIG",
            "SQLITE_TRANSACTION",
            "SQLITE_UPDATE",
            # 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",
        ]
        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")
        if sqlite.sqlite_version_info >= (3, 8, 7):
            consts.append("SQLITE_LIMIT_WORKER_THREADS")
        consts += ["PARSE_DECLTYPES", "PARSE_COLNAMES"]
        # Extended result codes
        consts += [
            "SQLITE_ABORT_ROLLBACK",
            "SQLITE_BUSY_RECOVERY",
            "SQLITE_CANTOPEN_FULLPATH",
            "SQLITE_CANTOPEN_ISDIR",
            "SQLITE_CANTOPEN_NOTEMPDIR",
            "SQLITE_CORRUPT_VTAB",
            "SQLITE_IOERR_ACCESS",
            "SQLITE_IOERR_BLOCKED",
            "SQLITE_IOERR_CHECKRESERVEDLOCK",
            "SQLITE_IOERR_CLOSE",
            "SQLITE_IOERR_DELETE",
            "SQLITE_IOERR_DELETE_NOENT",
            "SQLITE_IOERR_DIR_CLOSE",
            "SQLITE_IOERR_DIR_FSYNC",
            "SQLITE_IOERR_FSTAT",
            "SQLITE_IOERR_FSYNC",
            "SQLITE_IOERR_LOCK",
            "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_WRITE",
            "SQLITE_LOCKED_SHAREDCACHE",
            "SQLITE_READONLY_CANTLOCK",
            "SQLITE_READONLY_RECOVERY",
        ]
        if sqlite.sqlite_version_info >= (3, 7, 16):
            consts += [
                "SQLITE_CONSTRAINT_CHECK",
                "SQLITE_CONSTRAINT_COMMITHOOK",
                "SQLITE_CONSTRAINT_FOREIGNKEY",
                "SQLITE_CONSTRAINT_FUNCTION",
                "SQLITE_CONSTRAINT_NOTNULL",
                "SQLITE_CONSTRAINT_PRIMARYKEY",
                "SQLITE_CONSTRAINT_TRIGGER",
                "SQLITE_CONSTRAINT_UNIQUE",
                "SQLITE_CONSTRAINT_VTAB",
                "SQLITE_READONLY_ROLLBACK",
            ]
        if sqlite.sqlite_version_info >= (3, 7, 17):
            consts += [
                "SQLITE_IOERR_MMAP",
                "SQLITE_NOTICE_RECOVER_ROLLBACK",
                "SQLITE_NOTICE_RECOVER_WAL",
            ]
        if sqlite.sqlite_version_info >= (3, 8, 0):
            consts += [
                "SQLITE_BUSY_SNAPSHOT",
                "SQLITE_IOERR_GETTEMPPATH",
                "SQLITE_WARNING_AUTOINDEX",
            ]
        if sqlite.sqlite_version_info >= (3, 8, 1):
            consts += ["SQLITE_CANTOPEN_CONVPATH", "SQLITE_IOERR_CONVPATH"]
        if sqlite.sqlite_version_info >= (3, 8, 2):
            consts.append("SQLITE_CONSTRAINT_ROWID")
        if sqlite.sqlite_version_info >= (3, 8, 3):
            consts.append("SQLITE_READONLY_DBMOVED")
        if sqlite.sqlite_version_info >= (3, 8, 7):
            consts.append("SQLITE_AUTH_USER")
        if sqlite.sqlite_version_info >= (3, 9, 0):
            consts.append("SQLITE_IOERR_VNODE")
        if sqlite.sqlite_version_info >= (3, 10, 0):
            consts.append("SQLITE_IOERR_AUTH")
        if sqlite.sqlite_version_info >= (3, 14, 1):
            consts.append("SQLITE_OK_LOAD_PERMANENTLY")
        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.assertTrue(hasattr(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.assertTrue(e.sqlite_errorname.startswith("SQLITE_CANTOPEN"))

    @unittest.skipIf(sqlite.sqlite_version_info <= (3, 7, 16),
                     "Requires SQLite 3.7.16 or newer")
    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):
        cx = sqlite.connect(":memory:")
        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"))
        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):
            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

    # TODO: RUSTPYTHON
    @unittest.expectedFailure
    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

    # TODO: RUSTPYTHON
    @unittest.expectedFailure
    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)

    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)

    # TODO: RUSTPYTHON
    # @unittest.expectedFailure
    @unittest.skip("TODO: RUSTPYTHON deadlock")
    def test_connection_reinit(self):
        db = ":memory:"
        cx = sqlite.connect(db)
        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__(db)
        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"])

    # TODO: RUSTPYTHON
    @unittest.expectedFailure
    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)))


@unittest.skip("TODO: RUSTPYTHON")
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)


@unittest.skipUnless(hasattr(sqlite.Connection, "serialize"),
                     "Needs SQLite serialize API")
class SerializeTests(unittest.TestCase):
    def test_serialize_deserialize(self):
        with memory_database() as cx:
            with cx:
                cx.execute("create table t(t)")
            data = cx.serialize()
            self.assertEqual(len(data), 8192)

            # 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.
                cx.execute("create table fail(f)")


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__(). """
        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)

    @unittest.skipIf(sys.platform == "win32", "skipped on Windows")
    @unittest.skipIf(sys.platform == "darwin", "skipped on macOS")
    # TODO: RUSTPYTHON
    # @unittest.skipIf(is_emscripten or is_wasi, "not supported on Emscripten/WASI")
    @unittest.skipUnless(TESTFN_UNDECODABLE, "only works if there are undecodable paths")
    def test_open_with_undecodable_path(self):
        path = TESTFN_UNDECODABLE
        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_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")
    @unittest.skipIf(sys.platform == "darwin", "skipped on macOS")
    # TODO: RUSTPYTHON
    # @unittest.skipIf(is_emscripten or is_wasi, "not supported on Emscripten/WASI")
    @unittest.skipUnless(TESTFN_UNDECODABLE, "only works if there are undecodable paths")
    def test_open_undecodable_uri(self):
        path = TESTFN_UNDECODABLE
        self.addCleanup(unlink, path)
        uri = "file:" + urllib.parse.quote(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)

    # TODO: RUSTPYTHON
    @unittest.expectedFailure
    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:
            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_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;
               -- comment
               select 2
            """,
        )
        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.
            """
            select 5+4;

            /*
            foo
            */
            """,
        )
        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_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):
        """
        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)

    @unittest.skipIf(sqlite.sqlite_version_info < (3, 35, 0),
                     "Requires SQLite 3.35.0 or newer")
    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("""
            -- foo
            insert into test(name) values ('foo'), ('foo')
        """)
        self.assertEqual(self.cu.rowcount, 2)
        self.cu.execute("""
            /* -- messy *r /* /* ** *- *--
            */
            /* one more */ insert into test(name) values ('messy')
        """)
        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')")
        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