3 files changed, 91 insertions, 0 deletions
diff --git a/lib/sqlalchemy/testing/provision.py b/lib/sqlalchemy/testing/provision.py
index 678def327..9b5fa0255 100644
--- a/lib/sqlalchemy/testing/provision.py
+++ b/lib/sqlalchemy/testing/provision.py
@@ -313,3 +313,11 @@ def get_temp_table_name(cfg, eng, base_name):
     use. The mssql dialect's implementation will need a "#" prepended.
     """
     return base_name
+
+
+@register.init
+def set_default_schema_on_connection(cfg, dbapi_connection, schema_name):
+    raise NotImplementedError(
+        "backend does not implement a schema name set function: %s"
+        % (cfg.db.url,)
+    )
diff --git a/lib/sqlalchemy/testing/requirements.py b/lib/sqlalchemy/testing/requirements.py
index bd2d4eaf9..7b938c993 100644
--- a/lib/sqlalchemy/testing/requirements.py
+++ b/lib/sqlalchemy/testing/requirements.py
@@ -458,6 +458,13 @@ class SuiteRequirements(Requirements):
         return exclusions.closed()
 
     @property
+    def default_schema_name_switch(self):
+        """target dialect implements provisioning module including
+        set_default_schema_on_connection"""
+
+        return exclusions.closed()
+
+    @property
     def server_side_cursors(self):
         """Target dialect must support server side cursors."""
 
diff --git a/lib/sqlalchemy/testing/suite/test_dialect.py b/lib/sqlalchemy/testing/suite/test_dialect.py
index c5ede08c6..7f697b915 100644
--- a/lib/sqlalchemy/testing/suite/test_dialect.py
+++ b/lib/sqlalchemy/testing/suite/test_dialect.py
@@ -1,12 +1,15 @@
 #! coding: utf-8
 
+from sqlalchemy import event
 from .. import assert_raises
 from .. import config
+from .. import engines
 from .. import eq_
 from .. import fixtures
 from .. import ne_
 from .. import provide_metadata
 from ..config import requirements
+from ..provision import set_default_schema_on_connection
 from ..schema import Column
 from ..schema import Table
 from ... import exc
@@ -209,3 +212,76 @@ class EscapingTest(fixtures.TestBase):
                 ),
                 "some %% other value",
             )
+
+
+class WeCanSetDefaultSchemaWEventsTest(fixtures.TestBase):
+    __backend__ = True
+
+    __requires__ = ("default_schema_name_switch",)
+
+    def test_control_case(self):
+        default_schema_name = config.db.dialect.default_schema_name
+
+        eng = engines.testing_engine()
+        with eng.connect():
+            pass
+
+        eq_(eng.dialect.default_schema_name, default_schema_name)
+
+    def test_wont_work_wo_insert(self):
+        default_schema_name = config.db.dialect.default_schema_name
+
+        eng = engines.testing_engine()
+
+        @event.listens_for(eng, "connect")
+        def on_connect(dbapi_connection, connection_record):
+            set_default_schema_on_connection(
+                config, dbapi_connection, config.test_schema
+            )
+
+        with eng.connect() as conn:
+            what_it_should_be = eng.dialect._get_default_schema_name(conn)
+            eq_(what_it_should_be, config.test_schema)
+
+        eq_(eng.dialect.default_schema_name, default_schema_name)
+
+    def test_schema_change_on_connect(self):
+        eng = engines.testing_engine()
+
+        @event.listens_for(eng, "connect", insert=True)
+        def on_connect(dbapi_connection, connection_record):
+            set_default_schema_on_connection(
+                config, dbapi_connection, config.test_schema
+            )
+
+        with eng.connect() as conn:
+            what_it_should_be = eng.dialect._get_default_schema_name(conn)
+            eq_(what_it_should_be, config.test_schema)
+
+        eq_(eng.dialect.default_schema_name, config.test_schema)
+
+    def test_schema_change_works_w_transactions(self):
+        eng = engines.testing_engine()
+
+        @event.listens_for(eng, "connect", insert=True)
+        def on_connect(dbapi_connection, *arg):
+            set_default_schema_on_connection(
+                config, dbapi_connection, config.test_schema
+            )
+
+        with eng.connect() as conn:
+            trans = conn.begin()
+            what_it_should_be = eng.dialect._get_default_schema_name(conn)
+            eq_(what_it_should_be, config.test_schema)
+            trans.rollback()
+
+            what_it_should_be = eng.dialect._get_default_schema_name(conn)
+            eq_(what_it_should_be, config.test_schema)
+
+        eq_(eng.dialect.default_schema_name, config.test_schema)
+
+
+class FutureWeCanSetDefaultSchemaWEventsTest(
+    fixtures.FutureEngineMixin, WeCanSetDefaultSchemaWEventsTest
+):
+    pass