"""SQLite-specific tests."""

from sqlalchemy import and_
from sqlalchemy import CheckConstraint
from sqlalchemy import Column
from sqlalchemy import column
from sqlalchemy import create_engine
from sqlalchemy import event
from sqlalchemy import exc
from sqlalchemy import extract
from sqlalchemy import ForeignKey
from sqlalchemy import func
from sqlalchemy import Index
from sqlalchemy import literal
from sqlalchemy import MetaData
from sqlalchemy import PrimaryKeyConstraint
from sqlalchemy import schema
from sqlalchemy import select
from sqlalchemy import sql
from sqlalchemy import Table
from sqlalchemy import table
from sqlalchemy import testing
from sqlalchemy import text
from sqlalchemy import tuple_
from sqlalchemy import UniqueConstraint
from sqlalchemy.dialects.sqlite import base as sqlite
from sqlalchemy.dialects.sqlite import insert
from sqlalchemy.schema import CreateTable
from sqlalchemy.testing import assert_raises_message
from sqlalchemy.testing import AssertsCompiledSQL
from sqlalchemy.testing import eq_
from sqlalchemy.testing import eq_ignore_whitespace
from sqlalchemy.testing import expect_raises
from sqlalchemy.testing import fixtures
from sqlalchemy.testing import mock
from sqlalchemy.types import Boolean
from sqlalchemy.types import Integer
from sqlalchemy.types import String


def exec_sql(engine, sql, *args, **kwargs):
    # TODO: convert all tests to not use this
    with engine.begin() as conn:
        conn.exec_driver_sql(sql, *args, **kwargs)


class SQLTest(fixtures.TestBase, AssertsCompiledSQL):
    """Tests SQLite-dialect specific compilation."""

    __dialect__ = sqlite.dialect()

    def test_extract(self):
        t = sql.table("t", sql.column("col1"))
        mapping = {
            "month": "%m",
            "day": "%d",
            "year": "%Y",
            "second": "%S",
            "hour": "%H",
            "doy": "%j",
            "minute": "%M",
            "epoch": "%s",
            "dow": "%w",
            "week": "%W",
        }
        for field, subst in mapping.items():
            self.assert_compile(
                select(extract(field, t.c.col1)),
                "SELECT CAST(STRFTIME('%s', t.col1) AS "
                "INTEGER) AS anon_1 FROM t" % subst,
            )

    def test_plain_stringify_returning(self):
        t = Table(
            "t",
            MetaData(),
            Column("myid", Integer, primary_key=True),
            Column("name", String, server_default="some str"),
            Column("description", String, default=func.lower("hi")),
        )
        stmt = t.insert().values().return_defaults()
        eq_ignore_whitespace(
            str(stmt.compile(dialect=sqlite.SQLiteDialect())),
            "INSERT INTO t (description) VALUES (lower(?)) "
            "RETURNING myid, name, description",
        )

    def test_true_false(self):
        self.assert_compile(sql.false(), "0")
        self.assert_compile(sql.true(), "1")

    def test_is_distinct_from(self):
        self.assert_compile(
            sql.column("x").is_distinct_from(None), "x IS NOT NULL"
        )

        self.assert_compile(
            sql.column("x").is_not_distinct_from(False), "x IS 0"
        )

    def test_localtime(self):
        self.assert_compile(
            func.localtimestamp(), "DATETIME(CURRENT_TIMESTAMP, 'localtime')"
        )

    def test_constraints_with_schemas(self):
        metadata = MetaData()
        Table(
            "t1",
            metadata,
            Column("id", Integer, primary_key=True),
            schema="master",
        )
        t2 = Table(
            "t2",
            metadata,
            Column("id", Integer, primary_key=True),
            Column("t1_id", Integer, ForeignKey("master.t1.id")),
            schema="master",
        )
        t3 = Table(
            "t3",
            metadata,
            Column("id", Integer, primary_key=True),
            Column("t1_id", Integer, ForeignKey("master.t1.id")),
            schema="alternate",
        )
        t4 = Table(
            "t4",
            metadata,
            Column("id", Integer, primary_key=True),
            Column("t1_id", Integer, ForeignKey("master.t1.id")),
        )

        # schema->schema, generate REFERENCES with no schema name
        self.assert_compile(
            schema.CreateTable(t2),
            "CREATE TABLE master.t2 ("
            "id INTEGER NOT NULL, "
            "t1_id INTEGER, "
            "PRIMARY KEY (id), "
            "FOREIGN KEY(t1_id) REFERENCES t1 (id)"
            ")",
        )

        # schema->different schema, don't generate REFERENCES
        self.assert_compile(
            schema.CreateTable(t3),
            "CREATE TABLE alternate.t3 ("
            "id INTEGER NOT NULL, "
            "t1_id INTEGER, "
            "PRIMARY KEY (id)"
            ")",
        )

        # same for local schema
        self.assert_compile(
            schema.CreateTable(t4),
            "CREATE TABLE t4 ("
            "id INTEGER NOT NULL, "
            "t1_id INTEGER, "
            "PRIMARY KEY (id)"
            ")",
        )

    @testing.combinations(
        (
            Boolean(create_constraint=True),
            sql.false(),
            "BOOLEAN DEFAULT 0, CHECK (x IN (0, 1))",
        ),
        (
            String(),
            func.sqlite_version(),
            "VARCHAR DEFAULT (sqlite_version())",
        ),
        (Integer(), func.abs(-5) + 17, "INTEGER DEFAULT (abs(-5) + 17)"),
        (
            # test #12425
            String(),
            func.now(),
            "VARCHAR DEFAULT CURRENT_TIMESTAMP",
        ),
        (
            # test #12425
            String(),
            func.datetime(func.now(), "localtime"),
            "VARCHAR DEFAULT (datetime(CURRENT_TIMESTAMP, 'localtime'))",
        ),
        (
            # test #12425
            String(),
            text("datetime(CURRENT_TIMESTAMP, 'localtime')"),
            "VARCHAR DEFAULT (datetime(CURRENT_TIMESTAMP, 'localtime'))",
        ),
        (
            # default with leading spaces that should not be
            # parenthesized
            String,
            text("  'some default'"),
            "VARCHAR DEFAULT   'some default'",
        ),
        (String, text("'some default'"), "VARCHAR DEFAULT 'some default'"),
        argnames="datatype,default,expected",
    )
    def test_column_defaults_ddl(self, datatype, default, expected):
        t = Table(
            "t",
            MetaData(),
            Column(
                "x",
                datatype,
                server_default=default,
            ),
        )

        self.assert_compile(
            CreateTable(t),
            f"CREATE TABLE t (x {expected})",
        )

    def test_create_partial_index(self):
        m = MetaData()
        tbl = Table("testtbl", m, Column("data", Integer))
        idx = Index(
            "test_idx1",
            tbl.c.data,
            sqlite_where=and_(tbl.c.data > 5, tbl.c.data < 10),
        )

        # test quoting and all that

        idx2 = Index(
            "test_idx2",
            tbl.c.data,
            sqlite_where=and_(tbl.c.data > "a", tbl.c.data < "b's"),
        )
        self.assert_compile(
            schema.CreateIndex(idx),
            "CREATE INDEX test_idx1 ON testtbl (data) "
            "WHERE data > 5 AND data < 10",
            dialect=sqlite.dialect(),
        )
        self.assert_compile(
            schema.CreateIndex(idx2),
            "CREATE INDEX test_idx2 ON testtbl (data) "
            "WHERE data > 'a' AND data < 'b''s'",
            dialect=sqlite.dialect(),
        )

    def test_no_autoinc_on_composite_pk(self):
        m = MetaData()
        t = Table(
            "t",
            m,
            Column("x", Integer, primary_key=True, autoincrement=True),
            Column("y", Integer, primary_key=True),
        )
        assert_raises_message(
            exc.CompileError,
            "SQLite does not support autoincrement for composite",
            CreateTable(t).compile,
            dialect=sqlite.dialect(),
        )

    def test_in_tuple(self):
        compiled = (
            tuple_(column("q"), column("p"))
            .in_([(1, 2), (3, 4)])
            .compile(dialect=sqlite.dialect())
        )
        eq_(str(compiled), "(q, p) IN (__[POSTCOMPILE_param_1])")
        eq_(
            compiled._literal_execute_expanding_parameter(
                "param_1",
                compiled.binds["param_1"],
                compiled.binds["param_1"].value,
            ),
            (
                [
                    ("param_1_1_1", 1),
                    ("param_1_1_2", 2),
                    ("param_1_2_1", 3),
                    ("param_1_2_2", 4),
                ],
                "VALUES (?, ?), (?, ?)",
            ),
        )

    def test_create_table_without_rowid(self):
        m = MetaData()
        tbl = Table(
            "atable", m, Column("id", Integer), sqlite_with_rowid=False
        )
        self.assert_compile(
            schema.CreateTable(tbl),
            "CREATE TABLE atable (id INTEGER) WITHOUT ROWID",
        )

    def test_create_table_strict(self):
        m = MetaData()
        table = Table("atable", m, Column("id", Integer), sqlite_strict=True)
        self.assert_compile(
            schema.CreateTable(table),
            "CREATE TABLE atable (id INTEGER) STRICT",
        )

    def test_create_table_without_rowid_strict(self):
        m = MetaData()
        table = Table(
            "atable",
            m,
            Column("id", Integer),
            sqlite_with_rowid=False,
            sqlite_strict=True,
        )
        self.assert_compile(
            schema.CreateTable(table),
            "CREATE TABLE atable (id INTEGER) WITHOUT ROWID, STRICT",
        )


class OnConflictDDLTest(fixtures.TestBase, AssertsCompiledSQL):
    __dialect__ = sqlite.dialect()

    def test_on_conflict_clause_column_not_null(self):
        c = Column(
            "test", Integer, nullable=False, sqlite_on_conflict_not_null="FAIL"
        )

        self.assert_compile(
            schema.CreateColumn(c),
            "test INTEGER NOT NULL ON CONFLICT FAIL",
            dialect=sqlite.dialect(),
        )

    def test_on_conflict_clause_column_many_clause(self):
        meta = MetaData()
        t = Table(
            "n",
            meta,
            Column(
                "test",
                Integer,
                nullable=False,
                primary_key=True,
                sqlite_on_conflict_not_null="FAIL",
                sqlite_on_conflict_primary_key="IGNORE",
            ),
        )

        self.assert_compile(
            CreateTable(t),
            "CREATE TABLE n ("
            "test INTEGER NOT NULL ON CONFLICT FAIL, "
            "PRIMARY KEY (test) ON CONFLICT IGNORE)",
            dialect=sqlite.dialect(),
        )

    def test_on_conflict_clause_unique_constraint_from_column(self):
        meta = MetaData()
        t = Table(
            "n",
            meta,
            Column(
                "x", String(30), unique=True, sqlite_on_conflict_unique="FAIL"
            ),
        )

        self.assert_compile(
            CreateTable(t),
            "CREATE TABLE n (x VARCHAR(30), UNIQUE (x) ON CONFLICT FAIL)",
            dialect=sqlite.dialect(),
        )

    def test_on_conflict_clause_unique_constraint(self):
        meta = MetaData()
        t = Table(
            "n",
            meta,
            Column("id", Integer),
            Column("x", String(30)),
            UniqueConstraint("id", "x", sqlite_on_conflict="FAIL"),
        )

        self.assert_compile(
            CreateTable(t),
            "CREATE TABLE n (id INTEGER, x VARCHAR(30), "
            "UNIQUE (id, x) ON CONFLICT FAIL)",
            dialect=sqlite.dialect(),
        )

    def test_on_conflict_clause_primary_key(self):
        meta = MetaData()
        t = Table(
            "n",
            meta,
            Column(
                "id",
                Integer,
                primary_key=True,
                sqlite_on_conflict_primary_key="FAIL",
            ),
            sqlite_autoincrement=True,
        )

        self.assert_compile(
            CreateTable(t),
            "CREATE TABLE n (id INTEGER NOT NULL "
            "PRIMARY KEY ON CONFLICT FAIL AUTOINCREMENT)",
            dialect=sqlite.dialect(),
        )

    def test_on_conflict_clause_primary_key_constraint_from_column(self):
        meta = MetaData()
        t = Table(
            "n",
            meta,
            Column(
                "x",
                String(30),
                sqlite_on_conflict_primary_key="FAIL",
                primary_key=True,
            ),
        )

        self.assert_compile(
            CreateTable(t),
            "CREATE TABLE n (x VARCHAR(30) NOT NULL, "
            "PRIMARY KEY (x) ON CONFLICT FAIL)",
            dialect=sqlite.dialect(),
        )

    def test_on_conflict_clause_check_constraint(self):
        meta = MetaData()
        t = Table(
            "n",
            meta,
            Column("id", Integer),
            Column("x", Integer),
            CheckConstraint("id > x", sqlite_on_conflict="FAIL"),
        )

        self.assert_compile(
            CreateTable(t),
            "CREATE TABLE n (id INTEGER, x INTEGER, "
            "CHECK (id > x) ON CONFLICT FAIL)",
            dialect=sqlite.dialect(),
        )

    def test_on_conflict_clause_check_constraint_from_column(self):
        meta = MetaData()
        t = Table(
            "n",
            meta,
            Column(
                "x",
                Integer,
                CheckConstraint("x > 1", sqlite_on_conflict="FAIL"),
            ),
        )

        assert_raises_message(
            exc.CompileError,
            "SQLite does not support on conflict "
            "clause for column check constraint",
            CreateTable(t).compile,
            dialect=sqlite.dialect(),
        )

    def test_on_conflict_clause_primary_key_constraint(self):
        meta = MetaData()
        t = Table(
            "n",
            meta,
            Column("id", Integer),
            Column("x", String(30)),
            PrimaryKeyConstraint("id", "x", sqlite_on_conflict="FAIL"),
        )

        self.assert_compile(
            CreateTable(t),
            "CREATE TABLE n ("
            "id INTEGER NOT NULL, "
            "x VARCHAR(30) NOT NULL, "
            "PRIMARY KEY (id, x) ON CONFLICT FAIL)",
            dialect=sqlite.dialect(),
        )


class RegexpTest(fixtures.TestBase, testing.AssertsCompiledSQL):
    __dialect__ = "sqlite"

    def setup_test(self):
        self.table = table(
            "mytable", column("myid", String), column("name", String)
        )

    def _only_on_py38_w_sqlite_39():
        """in python 3.9 and above you can actually do::

            @(testing.requires.python38 + testing.only_on("sqlite > 3.9"))
            def test_determinsitic_parameter(self): ...

        that'll be cool.  until then...

        """
        return testing.requires.python38 + testing.only_on("sqlite >= 3.9")

    @_only_on_py38_w_sqlite_39()
    def test_determinsitic_parameter(self):
        """for #9379, make sure that "deterministic=True" is used when we are
        on python 3.8 with modern SQLite version.

        For the case where we are not on py3.8 or not on modern sqlite version,
        the rest of the test suite confirms that connection still passes.

        """
        e = create_engine("sqlite://")

        @event.listens_for(e, "do_connect", retval=True)
        def _mock_connect(dialect, conn_rec, cargs, cparams):
            conn = e.dialect.loaded_dbapi.connect(":memory:")
            return mock.Mock(wraps=conn)

        c = e.connect()
        eq_(
            c.connection.driver_connection.create_function.mock_calls,
            [
                mock.call("regexp", 2, mock.ANY, deterministic=True),
                mock.call("floor", 1, mock.ANY, deterministic=True),
            ],
        )

    def test_regexp_match(self):
        self.assert_compile(
            self.table.c.myid.regexp_match("pattern"),
            "mytable.myid REGEXP ?",
            checkpositional=("pattern",),
        )

    def test_regexp_match_column(self):
        self.assert_compile(
            self.table.c.myid.regexp_match(self.table.c.name),
            "mytable.myid REGEXP mytable.name",
            checkparams={},
        )

    def test_regexp_match_str(self):
        self.assert_compile(
            literal("string").regexp_match(self.table.c.name),
            "? REGEXP mytable.name",
            checkpositional=("string",),
        )

    def test_regexp_match_flags(self):
        self.assert_compile(
            self.table.c.myid.regexp_match("pattern", flags="ig"),
            "mytable.myid REGEXP ?",
            checkpositional=("pattern",),
        )

    def test_not_regexp_match(self):
        self.assert_compile(
            ~self.table.c.myid.regexp_match("pattern"),
            "mytable.myid NOT REGEXP ?",
            checkpositional=("pattern",),
        )

    def test_not_regexp_match_flags(self):
        self.assert_compile(
            ~self.table.c.myid.regexp_match("pattern", flags="ig"),
            "mytable.myid NOT REGEXP ?",
            checkpositional=("pattern",),
        )

    def test_not_regexp_match_column(self):
        self.assert_compile(
            ~self.table.c.myid.regexp_match(self.table.c.name),
            "mytable.myid NOT REGEXP mytable.name",
            checkparams={},
        )

    def test_not_regexp_match_str(self):
        self.assert_compile(
            ~literal("string").regexp_match(self.table.c.name),
            "? NOT REGEXP mytable.name",
            checkpositional=("string",),
        )

    def test_regexp_replace(self):
        assert_raises_message(
            exc.CompileError,
            "sqlite dialect does not support regular expression replacements",
            self.table.c.myid.regexp_replace("pattern", "rep").compile,
            dialect=sqlite.dialect(),
        )


class OnConflictCompileTest(AssertsCompiledSQL, fixtures.TestBase):
    __dialect__ = "sqlite"

    @testing.combinations(
        (
            lambda users, stmt: stmt.on_conflict_do_nothing(
                index_elements=["id"], index_where=text("name = 'hi'")
            ),
            "ON CONFLICT (id) WHERE name = 'hi' DO NOTHING",
        ),
        (
            lambda users, stmt: stmt.on_conflict_do_nothing(
                index_elements=["id"], index_where="name = 'hi'"
            ),
            exc.ArgumentError,
        ),
        (
            lambda users, stmt: stmt.on_conflict_do_nothing(
                index_elements=[users.c.id], index_where=users.c.name == "hi"
            ),
            "ON CONFLICT (id) WHERE name = __[POSTCOMPILE_name_1] DO NOTHING",
        ),
        (
            lambda users, stmt: stmt.on_conflict_do_update(
                index_elements=[users.c.id],
                set_={users.c.name: "there"},
                where=users.c.name == "hi",
            ),
            "ON CONFLICT (id) DO UPDATE SET name = ? " "WHERE users.name = ?",
        ),
        (
            lambda users, stmt: stmt.on_conflict_do_update(
                index_elements=[users.c.id],
                set_={users.c.name: "there"},
                where=text("name = 'hi'"),
            ),
            "ON CONFLICT (id) DO UPDATE SET name = ? " "WHERE name = 'hi'",
        ),
        (
            lambda users, stmt: stmt.on_conflict_do_update(
                index_elements=[users.c.id],
                set_={users.c.name: "there"},
                where="name = 'hi'",
            ),
            exc.ArgumentError,
        ),
        argnames="case,expected",
    )
    def test_assorted_arg_coercion(self, users, case, expected):
        stmt = insert(users)

        if isinstance(expected, type) and issubclass(expected, Exception):
            with expect_raises(expected):
                testing.resolve_lambda(case, stmt=stmt, users=users),
        else:
            self.assert_compile(
                testing.resolve_lambda(case, stmt=stmt, users=users),
                f"INSERT INTO users (id, name) VALUES (?, ?) {expected}",
            )

    @testing.combinations("control", "excluded", "dict", argnames="scenario")
    def test_set_excluded(self, scenario, users, users_w_key):
        """test #8014, sending all of .excluded to set"""

        if scenario == "control":

            stmt = insert(users)
            self.assert_compile(
                stmt.on_conflict_do_update(set_=stmt.excluded),
                "INSERT INTO users (id, name) VALUES (?, ?) ON CONFLICT  "
                "DO UPDATE SET id = excluded.id, name = excluded.name",
            )
        else:

            stmt = insert(users_w_key)

            if scenario == "excluded":
                self.assert_compile(
                    stmt.on_conflict_do_update(set_=stmt.excluded),
                    "INSERT INTO users_w_key (id, name) VALUES (?, ?) "
                    "ON CONFLICT  "
                    "DO UPDATE SET id = excluded.id, name = excluded.name",
                )
            else:
                self.assert_compile(
                    stmt.on_conflict_do_update(
                        set_={
                            "id": stmt.excluded.id,
                            "name_keyed": stmt.excluded.name_keyed,
                        }
                    ),
                    "INSERT INTO users_w_key (id, name) VALUES (?, ?) "
                    "ON CONFLICT  "
                    "DO UPDATE SET id = excluded.id, name = excluded.name",
                )

    def test_dont_consume_set_collection(self, users):
        stmt = insert(users).values(
            [
                {
                    "name": "spongebob",
                },
                {
                    "name": "sandy",
                },
            ]
        )
        stmt = stmt.on_conflict_do_update(
            index_elements=[users.c.name], set_=dict(name=stmt.excluded.name)
        )
        self.assert_compile(
            stmt,
            "INSERT INTO users (name) VALUES (?), (?) "
            "ON CONFLICT (name) DO UPDATE SET name = excluded.name",
        )
        stmt = stmt.returning(users)
        self.assert_compile(
            stmt,
            "INSERT INTO users (name) VALUES (?), (?) "
            "ON CONFLICT (name) DO UPDATE SET name = excluded.name "
            "RETURNING id, name",
        )

    def test_on_conflict_do_update_exotic_targets_six(self, users_xtra):
        users = users_xtra

        unique_partial_index = schema.Index(
            "idx_unique_partial_name",
            users_xtra.c.name,
            users_xtra.c.lets_index_this,
            unique=True,
            sqlite_where=users_xtra.c.lets_index_this == "unique_name",
        )

        i = insert(users)
        i = i.on_conflict_do_update(
            index_elements=unique_partial_index.columns,
            index_where=unique_partial_index.dialect_options["sqlite"][
                "where"
            ],
            set_=dict(
                name=i.excluded.name, login_email=i.excluded.login_email
            ),
        )

        # this test illustrates that the index_where clause can't use
        # bound parameters, where we see below a literal_execute parameter is
        # used (will be sent as literal to the DBAPI).  SQLite otherwise
        # fails here with "(sqlite3.OperationalError) ON CONFLICT clause does
        # not match any PRIMARY KEY or UNIQUE constraint" if sent as a real
        # bind parameter.
        self.assert_compile(
            i,
            "INSERT INTO users_xtra (id, name, login_email, lets_index_this) "
            "VALUES (?, ?, ?, ?) ON CONFLICT (name, lets_index_this) "
            "WHERE lets_index_this = __[POSTCOMPILE_lets_index_this_1] "
            "DO UPDATE "
            "SET name = excluded.name, login_email = excluded.login_email",
        )

    @testing.fixture
    def users(self):
        metadata = MetaData()
        return Table(
            "users",
            metadata,
            Column("id", Integer, primary_key=True),
            Column("name", String(50)),
        )

    @testing.fixture
    def users_w_key(self):
        metadata = MetaData()
        return Table(
            "users_w_key",
            metadata,
            Column("id", Integer, primary_key=True),
            Column("name", String(50), key="name_keyed"),
        )

    @testing.fixture
    def users_xtra(self):
        metadata = MetaData()
        return Table(
            "users_xtra",
            metadata,
            Column("id", Integer, primary_key=True),
            Column("name", String(50)),
            Column("login_email", String(50)),
            Column("lets_index_this", String(50)),
        )