# -*- encoding: utf-8
from sqlalchemy import Column
from sqlalchemy import DDL
from sqlalchemy import event
from sqlalchemy import ForeignKey
from sqlalchemy import Index
from sqlalchemy import inspect
from sqlalchemy import Integer
from sqlalchemy import MetaData
from sqlalchemy import PrimaryKeyConstraint
from sqlalchemy import schema
from sqlalchemy import Table
from sqlalchemy import testing
from sqlalchemy import types
from sqlalchemy import util
from sqlalchemy.databases import mssql
from sqlalchemy.dialects.mssql import base
from sqlalchemy.dialects.mssql.information_schema import CoerceUnicode
from sqlalchemy.dialects.mssql.information_schema import tables
from sqlalchemy.engine.reflection import Inspector
from sqlalchemy.testing import AssertsCompiledSQL
from sqlalchemy.testing import ComparesTables
from sqlalchemy.testing import eq_
from sqlalchemy.testing import fixtures
from sqlalchemy.testing import in_
from sqlalchemy.testing import is_
from sqlalchemy.testing import mock


class ReflectionTest(fixtures.TestBase, ComparesTables, AssertsCompiledSQL):
    __only_on__ = "mssql"
    __backend__ = True

    @testing.provide_metadata
    def test_basic_reflection(self):
        meta = self.metadata

        users = Table(
            "engine_users",
            meta,
            Column("user_id", types.INT, primary_key=True),
            Column("user_name", types.VARCHAR(20), nullable=False),
            Column("test1", types.CHAR(5), nullable=False),
            Column("test2", types.Float(5), nullable=False),
            Column("test2.5", types.Float(), nullable=False),
            Column("test3", types.Text()),
            Column("test4", types.Numeric, nullable=False),
            Column("test4.5", types.Numeric(10, 2), nullable=False),
            Column("test5", types.DateTime),
            Column(
                "parent_user_id",
                types.Integer,
                ForeignKey("engine_users.user_id"),
            ),
            Column("test6", types.DateTime, nullable=False),
            Column("test7", types.Text()),
            Column("test8", types.LargeBinary()),
            Column("test_passivedefault2", types.Integer, server_default="5"),
            Column("test9", types.BINARY(100)),
            Column("test_numeric", types.Numeric()),
        )

        addresses = Table(
            "engine_email_addresses",
            meta,
            Column("address_id", types.Integer, primary_key=True),
            Column(
                "remote_user_id", types.Integer, ForeignKey(users.c.user_id)
            ),
            Column("email_address", types.String(20)),
        )
        meta.create_all()

        meta2 = MetaData()
        reflected_users = Table(
            "engine_users", meta2, autoload=True, autoload_with=testing.db
        )
        reflected_addresses = Table(
            "engine_email_addresses",
            meta2,
            autoload=True,
            autoload_with=testing.db,
        )
        self.assert_tables_equal(users, reflected_users)
        self.assert_tables_equal(addresses, reflected_addresses)

    @testing.provide_metadata
    def _test_specific_type(self, type_obj, ddl):
        metadata = self.metadata

        table = Table("type_test", metadata, Column("col1", type_obj))
        table.create()

        m2 = MetaData()
        table2 = Table("type_test", m2, autoload_with=testing.db)
        self.assert_compile(
            schema.CreateTable(table2),
            "CREATE TABLE type_test (col1 %s NULL)" % ddl,
        )

    def test_xml_type(self):
        self._test_specific_type(mssql.XML, "XML")

    def test_image_type(self):
        self._test_specific_type(mssql.IMAGE, "IMAGE")

    def test_money_type(self):
        self._test_specific_type(mssql.MONEY, "MONEY")

    def test_numeric_prec_scale(self):
        self._test_specific_type(mssql.NUMERIC(10, 2), "NUMERIC(10, 2)")

    def test_float(self):
        self._test_specific_type(mssql.FLOAT, "FLOAT(53)")

    def test_real(self):
        self._test_specific_type(mssql.REAL, "REAL")

    def test_float_as_real(self):
        # FLOAT(5) comes back as REAL
        self._test_specific_type(mssql.FLOAT(5), "REAL")

    @testing.provide_metadata
    def test_identity(self):
        metadata = self.metadata
        table = Table(
            "identity_test",
            metadata,
            Column(
                "col1",
                Integer,
                mssql_identity_start=2,
                mssql_identity_increment=3,
                primary_key=True,
            ),
        )
        table.create()

        meta2 = MetaData(testing.db)
        table2 = Table("identity_test", meta2, autoload=True)
        eq_(table2.c["col1"].dialect_options["mssql"]["identity_start"], 2)
        eq_(table2.c["col1"].dialect_options["mssql"]["identity_increment"], 3)

    @testing.emits_warning("Did not recognize")
    @testing.provide_metadata
    def test_skip_types(self):
        metadata = self.metadata
        testing.db.execute(
            """
            create table foo (id integer primary key, data xml)
        """
        )
        with mock.patch.object(
            testing.db.dialect, "ischema_names", {"int": mssql.INTEGER}
        ):
            t1 = Table("foo", metadata, autoload=True)
        assert isinstance(t1.c.id.type, Integer)
        assert isinstance(t1.c.data.type, types.NullType)

    @testing.provide_metadata
    def test_cross_schema_fk_pk_name_overlaps(self):
        # test for issue #4228
        metadata = self.metadata

        Table(
            "subject",
            metadata,
            Column("id", Integer),
            PrimaryKeyConstraint("id", name="subj_pk"),
            schema=testing.config.test_schema,
        )

        Table(
            "referrer",
            metadata,
            Column("id", Integer, primary_key=True),
            Column(
                "sid",
                ForeignKey(
                    "%s.subject.id" % testing.config.test_schema,
                    name="fk_subject",
                ),
            ),
            schema=testing.config.test_schema,
        )

        Table(
            "subject",
            metadata,
            Column("id", Integer),
            PrimaryKeyConstraint("id", name="subj_pk"),
            schema=testing.config.test_schema_2,
        )

        metadata.create_all()

        insp = inspect(testing.db)
        eq_(
            insp.get_foreign_keys("referrer", testing.config.test_schema),
            [
                {
                    "name": "fk_subject",
                    "constrained_columns": ["sid"],
                    "referred_schema": "test_schema",
                    "referred_table": "subject",
                    "referred_columns": ["id"],
                }
            ],
        )

    @testing.provide_metadata
    def test_table_name_that_is_greater_than_16_chars(self):
        metadata = self.metadata
        Table(
            "ABCDEFGHIJKLMNOPQRSTUVWXYZ",
            metadata,
            Column("id", Integer, primary_key=True),
            Column("foo", Integer),
            Index("foo_idx", "foo"),
        )
        metadata.create_all()

        t = Table(
            "ABCDEFGHIJKLMNOPQRSTUVWXYZ", MetaData(), autoload_with=testing.db
        )
        eq_(t.name, "ABCDEFGHIJKLMNOPQRSTUVWXYZ")

    @testing.provide_metadata
    def test_db_qualified_items(self):
        metadata = self.metadata
        Table("foo", metadata, Column("id", Integer, primary_key=True))
        Table(
            "bar",
            metadata,
            Column("id", Integer, primary_key=True),
            Column("foo_id", Integer, ForeignKey("foo.id", name="fkfoo")),
        )
        metadata.create_all()

        dbname = testing.db.scalar("select db_name()")
        owner = testing.db.scalar("SELECT user_name()")
        referred_schema = "%(dbname)s.%(owner)s" % {
            "dbname": dbname,
            "owner": owner,
        }

        inspector = inspect(testing.db)
        bar_via_db = inspector.get_foreign_keys("bar", schema=referred_schema)
        eq_(
            bar_via_db,
            [
                {
                    "referred_table": "foo",
                    "referred_columns": ["id"],
                    "referred_schema": referred_schema,
                    "name": "fkfoo",
                    "constrained_columns": ["foo_id"],
                }
            ],
        )

        assert testing.db.has_table("bar", schema=referred_schema)

        m2 = MetaData()
        Table(
            "bar",
            m2,
            schema=referred_schema,
            autoload=True,
            autoload_with=testing.db,
        )
        eq_(m2.tables["%s.foo" % referred_schema].schema, referred_schema)

    @testing.provide_metadata
    def test_indexes_cols(self):
        metadata = self.metadata

        t1 = Table("t", metadata, Column("x", Integer), Column("y", Integer))
        Index("foo", t1.c.x, t1.c.y)
        metadata.create_all()

        m2 = MetaData()
        t2 = Table("t", m2, autoload=True, autoload_with=testing.db)

        eq_(set(list(t2.indexes)[0].columns), set([t2.c["x"], t2.c.y]))

    @testing.provide_metadata
    def test_indexes_cols_with_commas(self):
        metadata = self.metadata

        t1 = Table(
            "t",
            metadata,
            Column("x, col", Integer, key="x"),
            Column("y", Integer),
        )
        Index("foo", t1.c.x, t1.c.y)
        metadata.create_all()

        m2 = MetaData()
        t2 = Table("t", m2, autoload=True, autoload_with=testing.db)

        eq_(set(list(t2.indexes)[0].columns), set([t2.c["x, col"], t2.c.y]))

    @testing.provide_metadata
    def test_indexes_cols_with_spaces(self):
        metadata = self.metadata

        t1 = Table(
            "t",
            metadata,
            Column("x col", Integer, key="x"),
            Column("y", Integer),
        )
        Index("foo", t1.c.x, t1.c.y)
        metadata.create_all()

        m2 = MetaData()
        t2 = Table("t", m2, autoload=True, autoload_with=testing.db)

        eq_(set(list(t2.indexes)[0].columns), set([t2.c["x col"], t2.c.y]))

    @testing.provide_metadata
    def test_max_ident_in_varchar_not_present(self):
        """test [ticket:3504].

        Here we are testing not just that the "max" token comes back
        as None, but also that these types accept "max" as the value
        of "length" on construction, which isn't a directly documented
        pattern however is likely in common use.

        """
        metadata = self.metadata

        Table(
            "t",
            metadata,
            Column("t1", types.String),
            Column("t2", types.Text("max")),
            Column("t3", types.Text("max")),
            Column("t4", types.LargeBinary("max")),
            Column("t5", types.VARBINARY("max")),
        )
        metadata.create_all()
        for col in inspect(testing.db).get_columns("t"):
            is_(col["type"].length, None)
            in_("max", str(col["type"].compile(dialect=testing.db.dialect)))


class InfoCoerceUnicodeTest(fixtures.TestBase, AssertsCompiledSQL):
    def test_info_unicode_coercion(self):

        dialect = mssql.dialect()
        value = CoerceUnicode().bind_processor(dialect)("a string")
        assert isinstance(value, util.text_type)

    def test_info_unicode_cast_no_2000(self):
        dialect = mssql.dialect()
        dialect.server_version_info = base.MS_2000_VERSION
        stmt = tables.c.table_name == "somename"
        self.assert_compile(
            stmt,
            "[INFORMATION_SCHEMA].[TABLES].[TABLE_NAME] = :table_name_1",
            dialect=dialect,
        )

    def test_info_unicode_cast(self):
        dialect = mssql.dialect()
        dialect.server_version_info = base.MS_2005_VERSION
        stmt = tables.c.table_name == "somename"
        self.assert_compile(
            stmt,
            "[INFORMATION_SCHEMA].[TABLES].[TABLE_NAME] = "
            "CAST(:table_name_1 AS NVARCHAR(max))",
            dialect=dialect,
        )


class ReflectHugeViewTest(fixtures.TestBase):
    __only_on__ = "mssql"
    __backend__ = True

    # crashes on freetds 0.91, not worth it
    __skip_if__ = (lambda: testing.requires.mssql_freetds.enabled,)

    def setup(self):
        self.col_num = 150

        self.metadata = MetaData(testing.db)
        t = Table(
            "base_table",
            self.metadata,
            *[
                Column("long_named_column_number_%d" % i, Integer)
                for i in range(self.col_num)
            ]
        )
        self.view_str = (
            view_str
        ) = "CREATE VIEW huge_named_view AS SELECT %s FROM base_table" % (
            ",".join(
                "long_named_column_number_%d" % i for i in range(self.col_num)
            )
        )
        assert len(view_str) > 4000

        event.listen(t, "after_create", DDL(view_str))
        event.listen(t, "before_drop", DDL("DROP VIEW huge_named_view"))

        self.metadata.create_all()

    def teardown(self):
        self.metadata.drop_all()

    def test_inspect_view_definition(self):
        inspector = Inspector.from_engine(testing.db)
        view_def = inspector.get_view_definition("huge_named_view")
        eq_(view_def, self.view_str)


class OwnerPlusDBTest(fixtures.TestBase):
    def test_default_schema_name_not_interpreted_as_tokenized(self):
        dialect = mssql.dialect()
        dialect.server_version_info = base.MS_2014_VERSION

        mock_connection = mock.Mock(scalar=lambda sql: "Jonah.The.Whale")
        schema_name = dialect._get_default_schema_name(mock_connection)
        eq_(schema_name, "Jonah.The.Whale")
        eq_(
            base._owner_plus_db(dialect, schema_name),
            (None, "Jonah.The.Whale"),
        )

    def test_owner_database_pairs_dont_use_for_same_db(self):
        dialect = mssql.dialect()

        identifier = "my_db.some_schema"
        schema, owner = base._owner_plus_db(dialect, identifier)

        mock_connection = mock.Mock(
            dialect=dialect, scalar=mock.Mock(return_value="my_db")
        )
        mock_lambda = mock.Mock()
        base._switch_db(schema, mock_connection, mock_lambda, "x", y="bar")
        eq_(mock_connection.mock_calls, [mock.call.scalar("select db_name()")])
        eq_(mock_lambda.mock_calls, [mock.call("x", y="bar")])

    def test_owner_database_pairs_switch_for_different_db(self):
        dialect = mssql.dialect()

        identifier = "my_other_db.some_schema"
        schema, owner = base._owner_plus_db(dialect, identifier)

        mock_connection = mock.Mock(
            dialect=dialect, scalar=mock.Mock(return_value="my_db")
        )
        mock_lambda = mock.Mock()
        base._switch_db(schema, mock_connection, mock_lambda, "x", y="bar")
        eq_(
            mock_connection.mock_calls,
            [
                mock.call.scalar("select db_name()"),
                mock.call.execute("use my_other_db"),
                mock.call.execute("use my_db"),
            ],
        )
        eq_(mock_lambda.mock_calls, [mock.call("x", y="bar")])

    @testing.combinations(
        ("foo", None, "foo", "use foo"),
        ("foo.bar", "foo", "bar", "use foo"),
        ("Foo.Bar", "Foo", "Bar", "use [Foo]"),
        ("[Foo.Bar]", None, "Foo.Bar", "use [Foo.Bar]"),
        ("[Foo.Bar].[bat]", "Foo.Bar", "bat", "use [Foo.Bar]"),
        (
            "[foo].]do something; select [foo",
            "foo",
            "do something; select foo",
            "use foo",
        ),
        (
            "something; select [foo].bar",
            "something; select foo",
            "bar",
            "use [something; select foo]",
        ),
        (
            "[abc].[def].[efg].[hij]",
            "[abc].[def].[efg]",
            "hij",
            "use [abc].[def].[efg]",
        ),
        ("abc.def.efg.hij", "abc.def.efg", "hij", "use [abc.def.efg]"),
    )
    def test_owner_database_pairs(
        self, identifier, expected_schema, expected_owner, use_stmt
    ):
        dialect = mssql.dialect()

        schema, owner = base._owner_plus_db(dialect, identifier)

        eq_(owner, expected_owner)
        eq_(schema, expected_schema)

        mock_connection = mock.Mock(
            dialect=dialect,
            scalar=mock.Mock(return_value="Some  Database"),
        )
        mock_lambda = mock.Mock()
        base._switch_db(schema, mock_connection, mock_lambda, "x", y="bar")
        if schema is None:
            eq_(mock_connection.mock_calls, [])
        else:
            eq_(
                mock_connection.mock_calls,
                [
                    mock.call.scalar("select db_name()"),
                    mock.call.execute(use_stmt),
                    mock.call.execute("use [Some  Database]"),
                ],
            )
        eq_(mock_lambda.mock_calls, [mock.call("x", y="bar")])