Support multiple dotted sections in mssql schema names

Refined the logic used by the SQL Server dialect to interpret multi-part
schema names that contain many dots, to not actually lose any dots if the
name does not have bracking or quoting used, and additionally to support a
"dbname" token that has many parts including that it may have multiple,
independently-bracketed sections.

This fix addresses #5364 to some degree but probably does not
resolve it fully.

References: #5364
Fixes: #5366
Change-Id: I460cd74ce443efb35fb63b6864f00c6d81422688

2 files changed, 95 insertions, 49 deletions

diff --git a/test/dialect/mssql/test_compiler.py b/test/dialect/mssql/test_compiler.py
index b7a06c8e3..25af3240e 100644
--- a/test/dialect/mssql/test_compiler.py
+++ b/test/dialect/mssql/test_compiler.py
@@ -22,6 +22,7 @@ from sqlalchemy import union
 from sqlalchemy import UniqueConstraint
 from sqlalchemy import update
 from sqlalchemy.dialects import mssql
+from sqlalchemy.dialects.mssql import base as mssql_base
 from sqlalchemy.dialects.mssql import mxodbc
 from sqlalchemy.dialects.mssql.base import try_cast
 from sqlalchemy.sql import column
@@ -525,6 +526,42 @@ class CompileTest(fixtures.TestBase, AssertsCompiledSQL):
             checkpositional=("bar",),
         )
 
+    def test_schema_many_tokens_one(self):
+        metadata = MetaData()
+        tbl = Table(
+            "test",
+            metadata,
+            Column("id", Integer, primary_key=True),
+            schema="abc.def.efg.hij",
+        )
+
+        # for now, we don't really know what the above means, at least
+        # don't lose the dot
+        self.assert_compile(
+            select([tbl]),
+            "SELECT [abc.def.efg].hij.test.id FROM [abc.def.efg].hij.test",
+        )
+
+        dbname, owner = mssql_base._schema_elements("abc.def.efg.hij")
+        eq_(dbname, "abc.def.efg")
+        assert not isinstance(dbname, quoted_name)
+        eq_(owner, "hij")
+
+    def test_schema_many_tokens_two(self):
+        metadata = MetaData()
+        tbl = Table(
+            "test",
+            metadata,
+            Column("id", Integer, primary_key=True),
+            schema="[abc].[def].[efg].[hij]",
+        )
+
+        self.assert_compile(
+            select([tbl]),
+            "SELECT [abc].[def].[efg].hij.test.id "
+            "FROM [abc].[def].[efg].hij.test",
+        )
+
     def test_force_schema_quoted_name_w_dot_case_insensitive(self):
         metadata = MetaData()
         tbl = Table(
diff --git a/test/dialect/mssql/test_reflection.py b/test/dialect/mssql/test_reflection.py
index 5328513e4..176d3d2ec 100644
--- a/test/dialect/mssql/test_reflection.py
+++ b/test/dialect/mssql/test_reflection.py
@@ -484,56 +484,65 @@ class OwnerPlusDBTest(fixtures.TestBase):
         )
         eq_(mock_lambda.mock_calls, [mock.call("x", y="bar")])
 
-    def test_owner_database_pairs(self):
+    @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()
 
-        for identifier, expected_schema, expected_owner, use_stmt in [
-            ("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]",
+        schema, owner = base._owner_plus_db(dialect, identifier)
+
+        eq_(owner, expected_owner)
+        eq_(schema, expected_schema)
+
+        mock_connection = mock.Mock(
+            dialect=dialect,
+            exec_driver_sql=mock.Mock(
+                return_value=mock.Mock(
+                    scalar=mock.Mock(return_value="Some Database")
+                )
             ),
-        ]:
-            schema, owner = base._owner_plus_db(dialect, identifier)
-
-            eq_(owner, expected_owner)
-            eq_(schema, expected_schema)
-
-            mock_connection = mock.Mock(
-                dialect=dialect,
-                exec_driver_sql=mock.Mock(
-                    return_value=mock.Mock(
-                        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.exec_driver_sql("select db_name()"),
+                    mock.call.exec_driver_sql(use_stmt),
+                    mock.call.exec_driver_sql("use [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.exec_driver_sql("select db_name()"),
-                        mock.call.exec_driver_sql(use_stmt),
-                        mock.call.exec_driver_sql("use [Some  Database]"),
-                    ],
-                )
-                eq_(
-                    mock_connection.exec_driver_sql.return_value.mock_calls,
-                    [mock.call.scalar()],
-                )
-            eq_(mock_lambda.mock_calls, [mock.call("x", y="bar")])
+            eq_(
+                mock_connection.exec_driver_sql.return_value.mock_calls,
+                [mock.call.scalar()],
+            )
+        eq_(mock_lambda.mock_calls, [mock.call("x", y="bar")])