# -*- encoding: utf-8
from sqlalchemy import and_
from sqlalchemy import Column
from sqlalchemy import DDL
from sqlalchemy import desc
from sqlalchemy import event
from sqlalchemy import ForeignKey
from sqlalchemy import func
from sqlalchemy import Identity
from sqlalchemy import Integer
from sqlalchemy import literal
from sqlalchemy import or_
from sqlalchemy import PrimaryKeyConstraint
from sqlalchemy import select
from sqlalchemy import String
from sqlalchemy import Table
from sqlalchemy import testing
from sqlalchemy.dialects.mssql import base as mssql
from sqlalchemy.testing import AssertsCompiledSQL
from sqlalchemy.testing import config
from sqlalchemy.testing import engines
from sqlalchemy.testing import eq_
from sqlalchemy.testing import fixtures
from sqlalchemy.testing.assertsql import CursorSQL
from sqlalchemy.testing.assertsql import DialectSQL


class IdentityInsertTest(fixtures.TablesTest, AssertsCompiledSQL):
    __only_on__ = "mssql"
    __dialect__ = mssql.MSDialect()
    __backend__ = True

    @classmethod
    def define_tables(cls, metadata):
        Table(
            "cattable",
            metadata,
            Column("id", Integer),
            Column("description", String(50)),
            PrimaryKeyConstraint("id", name="PK_cattable"),
        )

    def test_compiled(self):
        cattable = self.tables.cattable
        self.assert_compile(
            cattable.insert().values(id=9, description="Python"),
            "INSERT INTO cattable (id, description) "
            "VALUES (:id, :description)",
        )

    def test_execute(self, connection):
        conn = connection
        cattable = self.tables.cattable
        conn.execute(cattable.insert().values(id=9, description="Python"))

        cats = conn.execute(cattable.select().order_by(cattable.c.id))
        eq_([(9, "Python")], list(cats))

        result = conn.execute(cattable.insert().values(description="PHP"))
        eq_(result.inserted_primary_key, (10,))
        lastcat = conn.execute(cattable.select().order_by(desc(cattable.c.id)))
        eq_((10, "PHP"), lastcat.first())

    def test_executemany(self, connection):
        conn = connection
        cattable = self.tables.cattable
        conn.execute(
            cattable.insert(),
            [
                {"id": 89, "description": "Python"},
                {"id": 8, "description": "Ruby"},
                {"id": 3, "description": "Perl"},
                {"id": 1, "description": "Java"},
            ],
        )
        cats = conn.execute(cattable.select().order_by(cattable.c.id))
        eq_(
            [(1, "Java"), (3, "Perl"), (8, "Ruby"), (89, "Python")],
            list(cats),
        )
        conn.execute(
            cattable.insert(),
            [{"description": "PHP"}, {"description": "Smalltalk"}],
        )
        lastcats = conn.execute(
            cattable.select().order_by(desc(cattable.c.id)).limit(2)
        )
        eq_([(91, "Smalltalk"), (90, "PHP")], list(lastcats))

    def test_insert_plain_param(self, connection):
        conn = connection
        cattable = self.tables.cattable
        conn.execute(cattable.insert(), dict(id=5))
        eq_(conn.scalar(select(cattable.c.id)), 5)

    def test_insert_values_key_plain(self, connection):
        conn = connection
        cattable = self.tables.cattable
        conn.execute(cattable.insert().values(id=5))
        eq_(conn.scalar(select(cattable.c.id)), 5)

    def test_insert_values_key_expression(self, connection):
        conn = connection
        cattable = self.tables.cattable
        conn.execute(cattable.insert().values(id=literal(5)))
        eq_(conn.scalar(select(cattable.c.id)), 5)

    def test_insert_values_col_plain(self, connection):
        conn = connection
        cattable = self.tables.cattable
        conn.execute(cattable.insert().values({cattable.c.id: 5}))
        eq_(conn.scalar(select(cattable.c.id)), 5)

    def test_insert_values_col_expression(self, connection):
        conn = connection
        cattable = self.tables.cattable
        conn.execute(cattable.insert().values({cattable.c.id: literal(5)}))
        eq_(conn.scalar(select(cattable.c.id)), 5)

    @testing.requires.schemas
    def test_insert_using_schema_translate(self, connection, metadata):

        t = Table(
            "t",
            metadata,
            Column("id", Integer),
            Column("description", String(50)),
            PrimaryKeyConstraint("id", name="PK_cattable"),
            schema=None,
        )
        conn = connection.execution_options(
            schema_translate_map={None: config.test_schema}
        )
        metadata.create_all(conn)

        conn.execute(t.insert().values({"id": 1, "description": "descrip"}))

        eq_(conn.execute(select(t)).first(), (1, "descrip"))


class QueryTest(testing.AssertsExecutionResults, fixtures.TestBase):
    __only_on__ = "mssql"
    __backend__ = True

    def test_fetchid_trigger(self, metadata, connection):
        # TODO: investigate test hang on mssql when connection fixture is used
        """
        Verify identity return value on inserting to a trigger table.

        MSSQL's OUTPUT INSERTED clause does not work for the
        case of a table having an identity (autoincrement)
        primary key column, and which also has a trigger configured