From 59f2d0c96152128d8e02e43dcda045593153c86e Mon Sep 17 00:00:00 2001
From: Mike Barry <michael.barry@gmo.com>
Date: Wed, 26 Oct 2022 13:40:32 -0400
Subject: use only object_id() function for temp tables

Fixed issue with :meth:`.Inspector.has_table` when used against a temporary
table for the SQL Server dialect would fail an invalid object name error on
some Azure variants, due to an unnecessary information schema query that is
not supported on those server versions. Pull request courtesy Mike Barry.

the patch also fills out test support for has_table()
against temp tables, temp views, adding to the has_table() support just
added for views in #8700.

Fixes: #8714
Closes: #8716
Pull-request: https://github.com/sqlalchemy/sqlalchemy/pull/8716
Pull-request-sha: e2ac7a52e2b09a349a703ba1e1a2911f4d3c0912

Change-Id: Ia73e4e9e977a2d6b7e100abd2f81a8c8777dc9bb
---
 lib/sqlalchemy/testing/requirements.py          |  5 ++
 lib/sqlalchemy/testing/suite/test_reflection.py | 77 ++++++++++++++++++++-----
 2 files changed, 69 insertions(+), 13 deletions(-)

(limited to 'lib/sqlalchemy/testing')

diff --git a/lib/sqlalchemy/testing/requirements.py b/lib/sqlalchemy/testing/requirements.py
index 1d6f1103d..8ceb2af2f 100644
--- a/lib/sqlalchemy/testing/requirements.py
+++ b/lib/sqlalchemy/testing/requirements.py
@@ -709,6 +709,11 @@ class SuiteRequirements(Requirements):
         """target dialect supports listing of temporary table names"""
         return exclusions.closed()
 
+    @property
+    def has_temp_table(self):
+        """target dialect supports checking a single temp table name"""
+        return exclusions.closed()
+
     @property
     def temporary_tables(self):
         """target database supports temporary tables"""
diff --git a/lib/sqlalchemy/testing/suite/test_reflection.py b/lib/sqlalchemy/testing/suite/test_reflection.py
index dd7d38c5a..68d1c13fa 100644
--- a/lib/sqlalchemy/testing/suite/test_reflection.py
+++ b/lib/sqlalchemy/testing/suite/test_reflection.py
@@ -45,7 +45,23 @@ from ...testing import mock
 metadata, users = None, None
 
 
-class HasTableTest(fixtures.TablesTest):
+class OneConnectionTablesTest(fixtures.TablesTest):
+    @classmethod
+    def setup_bind(cls):
+        # TODO: when temp tables are subject to server reset,
+        # this will also have to disable that server reset from
+        # happening
+        if config.requirements.independent_connections.enabled:
+            from sqlalchemy import pool
+
+            return engines.testing_engine(
+                options=dict(poolclass=pool.StaticPool, scope="class"),
+            )
+        else:
+            return config.db
+
+
+class HasTableTest(OneConnectionTablesTest):
     __backend__ = True
 
     @classmethod
@@ -67,6 +83,8 @@ class HasTableTest(fixtures.TablesTest):
 
         if testing.requires.view_reflection:
             cls.define_views(metadata)
+        if testing.requires.has_temp_table.enabled:
+            cls.define_temp_tables(metadata)
 
     @classmethod
     def define_views(cls, metadata):
@@ -87,6 +105,37 @@ class HasTableTest(fixtures.TablesTest):
                 DDL("DROP VIEW %s.vv" % (config.test_schema)),
             )
 
+    @classmethod
+    def temp_table_name(cls):
+        return get_temp_table_name(
+            config, config.db, f"user_tmp_{config.ident}"
+        )
+
+    @classmethod
+    def define_temp_tables(cls, metadata):
+        kw = temp_table_keyword_args(config, config.db)
+        table_name = cls.temp_table_name()
+        user_tmp = Table(
+            table_name,
+            metadata,
+            Column("id", sa.INT, primary_key=True),
+            Column("name", sa.VARCHAR(50)),
+            **kw,
+        )
+        if (
+            testing.requires.view_reflection.enabled
+            and testing.requires.temporary_views.enabled
+        ):
+            event.listen(
+                user_tmp,
+                "after_create",
+                DDL(
+                    "create temporary view user_tmp_v as "
+                    "select * from user_tmp_%s" % config.ident
+                ),
+            )
+            event.listen(user_tmp, "before_drop", DDL("drop view user_tmp_v"))
+
     def test_has_table(self):
         with config.db.begin() as conn:
             is_true(config.db.dialect.has_table(conn, "test_table"))
@@ -130,6 +179,19 @@ class HasTableTest(fixtures.TablesTest):
         insp = inspect(connection)
         is_true(insp.has_table("vv"))
 
+    @testing.requires.has_temp_table
+    def test_has_table_temp_table(self, connection):
+        insp = inspect(connection)
+        temp_table_name = self.temp_table_name()
+        is_true(insp.has_table(temp_table_name))
+
+    @testing.requires.has_temp_table
+    @testing.requires.view_reflection
+    @testing.requires.temporary_views
+    def test_has_table_temp_view(self, connection):
+        insp = inspect(connection)
+        is_true(insp.has_table("user_tmp_v"))
+
     @testing.requires.views
     @testing.requires.schemas
     def test_has_table_view_schema(self, connection):
@@ -393,22 +455,11 @@ def _multi_combination(fn):
     return schema(scope(kind(filter_names(fn))))
 
 
-class ComponentReflectionTest(ComparesTables, fixtures.TablesTest):
+class ComponentReflectionTest(ComparesTables, OneConnectionTablesTest):
     run_inserts = run_deletes = None
 
     __backend__ = True
 
-    @classmethod
-    def setup_bind(cls):
-        if config.requirements.independent_connections.enabled:
-            from sqlalchemy import pool
-
-            return engines.testing_engine(
-                options=dict(poolclass=pool.StaticPool, scope="class"),
-            )
-        else:
-            return config.db
-
     @classmethod
     def define_tables(cls, metadata):
         cls.define_reflected_tables(metadata, None)
-- 
cgit v1.2.1