Deprecate connection branching

The :meth:`.Connection.connect` method is deprecated as is the concept of
"connection branching", which copies a :class:`.Connection` into a new one
that has a no-op ".close()" method.  This pattern is oriented around the
"connectionless execution" concept which is also being removed in 2.0.

As part of this change we begin to move the internals away from
"connectionless execution" overall.   Remove the "connectionless
execution" concept from the reflection internals and replace with
explicit patterns at the Inspector level.

Fixes: #5131
Change-Id: Id23d28a9889212ac5ae7329b85136157815d3e6f

9 files changed, 199 insertions, 126 deletions

diff --git a/test/dialect/mssql/test_reflection.py b/test/dialect/mssql/test_reflection.py
index 794588dce..120092e66 100644
--- a/test/dialect/mssql/test_reflection.py
+++ b/test/dialect/mssql/test_reflection.py
@@ -17,7 +17,6 @@ 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_
@@ -415,7 +414,7 @@ class ReflectHugeViewTest(fixtures.TestBase):
         self.metadata.drop_all()
 
     def test_inspect_view_definition(self):
-        inspector = Inspector.from_engine(testing.db)
+        inspector = inspect(testing.db)
         view_def = inspector.get_view_definition("huge_named_view")
         eq_(view_def, self.view_str)
 
diff --git a/test/dialect/postgresql/test_reflection.py b/test/dialect/postgresql/test_reflection.py
index b410ca748..830a54eef 100644
--- a/test/dialect/postgresql/test_reflection.py
+++ b/test/dialect/postgresql/test_reflection.py
@@ -25,7 +25,6 @@ from sqlalchemy.dialects.postgresql import ExcludeConstraint
 from sqlalchemy.dialects.postgresql import INTEGER
 from sqlalchemy.dialects.postgresql import INTERVAL
 from sqlalchemy.dialects.postgresql import TSRANGE
-from sqlalchemy.engine import reflection
 from sqlalchemy.sql.schema import CheckConstraint
 from sqlalchemy.testing import fixtures
 from sqlalchemy.testing import mock
@@ -1199,7 +1198,7 @@ class ReflectionTest(fixtures.TestBase):
             metadata=self.metadata,
         )
         enum_type.create(conn)
-        inspector = reflection.Inspector.from_engine(conn.engine)
+        inspector = inspect(conn)
         eq_(
             inspector.get_enums("test_schema"),
             [
@@ -1218,7 +1217,7 @@ class ReflectionTest(fixtures.TestBase):
             "cat", "dog", "rat", name="pet", metadata=self.metadata
         )
         enum_type.create(testing.db)
-        inspector = reflection.Inspector.from_engine(testing.db)
+        inspector = inspect(testing.db)
         eq_(
             inspector.get_enums(),
             [
@@ -1356,7 +1355,7 @@ class ReflectionTest(fixtures.TestBase):
         )
         enum_type.create(testing.db)
         schema_enum_type.create(testing.db)
-        inspector = reflection.Inspector.from_engine(testing.db)
+        inspector = inspect(testing.db)
 
         eq_(
             inspector.get_enums(),
@@ -1392,7 +1391,7 @@ class ReflectionTest(fixtures.TestBase):
     def test_inspect_enum_empty(self):
         enum_type = postgresql.ENUM(name="empty", metadata=self.metadata)
         enum_type.create(testing.db)
-        inspector = reflection.Inspector.from_engine(testing.db)
+        inspector = inspect(testing.db)
 
         eq_(
             inspector.get_enums(),
diff --git a/test/dialect/test_sqlite.py b/test/dialect/test_sqlite.py
index da349c1f8..7675b8aa6 100644
--- a/test/dialect/test_sqlite.py
+++ b/test/dialect/test_sqlite.py
@@ -35,7 +35,6 @@ from sqlalchemy import UniqueConstraint
 from sqlalchemy import util
 from sqlalchemy.dialects.sqlite import base as sqlite
 from sqlalchemy.dialects.sqlite import pysqlite as pysqlite_dialect
-from sqlalchemy.engine.reflection import Inspector
 from sqlalchemy.engine.url import make_url
 from sqlalchemy.schema import CreateTable
 from sqlalchemy.schema import FetchedValue
@@ -1963,7 +1962,7 @@ class ConstraintReflectionTest(fixtures.TestBase):
 
     def test_foreign_key_name_is_none(self):
         # and not "0"
-        inspector = Inspector(testing.db)
+        inspector = inspect(testing.db)
         fks = inspector.get_foreign_keys("b")
         eq_(
             fks,
@@ -1988,7 +1987,7 @@ class ConstraintReflectionTest(fixtures.TestBase):
         )
 
     def test_foreign_key_name_is_not_none(self):
-        inspector = Inspector(testing.db)
+        inspector = inspect(testing.db)
         fks = inspector.get_foreign_keys("c")
         eq_(
             fks,
@@ -2013,7 +2012,7 @@ class ConstraintReflectionTest(fixtures.TestBase):
         )
 
     def test_foreign_key_implicit_parent(self):
-        inspector = Inspector(testing.db)
+        inspector = inspect(testing.db)
         fks = inspector.get_foreign_keys("implicit_referrer")
         eq_(
             fks,
@@ -2030,7 +2029,7 @@ class ConstraintReflectionTest(fixtures.TestBase):
         )
 
     def test_foreign_key_composite_implicit_parent(self):
-        inspector = Inspector(testing.db)
+        inspector = inspect(testing.db)
         fks = inspector.get_foreign_keys("implicit_referrer_comp")
         eq_(
             fks,
@@ -2049,7 +2048,7 @@ class ConstraintReflectionTest(fixtures.TestBase):
     def test_foreign_key_implicit_missing_parent(self):
         # test when the FK refers to a non-existent table and column names
         # aren't given.   only sqlite allows this case to exist
-        inspector = Inspector(testing.db)
+        inspector = inspect(testing.db)
         fks = inspector.get_foreign_keys("implicit_referrer_comp_fake")
         # the referred table doesn't exist but the operation does not fail
         eq_(
@@ -2079,7 +2078,7 @@ class ConstraintReflectionTest(fixtures.TestBase):
         )
 
     def test_unnamed_inline_foreign_key(self):
-        inspector = Inspector(testing.db)
+        inspector = inspect(testing.db)
         fks = inspector.get_foreign_keys("e")
         eq_(
             fks,
@@ -2096,7 +2095,7 @@ class ConstraintReflectionTest(fixtures.TestBase):
         )
 
     def test_unnamed_inline_foreign_key_quoted(self):
-        inspector = Inspector(testing.db)
+        inspector = inspect(testing.db)
         fks = inspector.get_foreign_keys("e1")
         eq_(
             fks,
@@ -2127,7 +2126,7 @@ class ConstraintReflectionTest(fixtures.TestBase):
         )
 
     def test_foreign_key_composite_broken_casing(self):
-        inspector = Inspector(testing.db)
+        inspector = inspect(testing.db)
         fks = inspector.get_foreign_keys("j")
         eq_(
             fks,
@@ -2158,7 +2157,7 @@ class ConstraintReflectionTest(fixtures.TestBase):
         )
 
     def test_foreign_key_ondelete_onupdate(self):
-        inspector = Inspector(testing.db)
+        inspector = inspect(testing.db)
         fks = inspector.get_foreign_keys("onud_test")
         eq_(
             fks,
@@ -2221,7 +2220,7 @@ class ConstraintReflectionTest(fixtures.TestBase):
             )
 
     def test_dont_reflect_autoindex(self):
-        inspector = Inspector(testing.db)
+        inspector = inspect(testing.db)
         eq_(inspector.get_indexes("o"), [])
         eq_(
             inspector.get_indexes("o", include_auto_indexes=True),
@@ -2237,7 +2236,7 @@ class ConstraintReflectionTest(fixtures.TestBase):
     def test_create_index_with_schema(self):
         """Test creation of index with explicit schema"""
 
-        inspector = Inspector(testing.db)
+        inspector = inspect(testing.db)
         eq_(
             inspector.get_indexes("l", schema="main"),
             [
@@ -2250,35 +2249,35 @@ class ConstraintReflectionTest(fixtures.TestBase):
         )
 
     def test_unique_constraint_named(self):
-        inspector = Inspector(testing.db)
+        inspector = inspect(testing.db)
         eq_(
             inspector.get_unique_constraints("f"),
             [{"column_names": ["x"], "name": "foo_fx"}],
         )
 
     def test_unique_constraint_named_broken_casing(self):
-        inspector = Inspector(testing.db)
+        inspector = inspect(testing.db)
         eq_(
             inspector.get_unique_constraints("h"),
             [{"column_names": ["x"], "name": "foo_hx"}],
         )
 
     def test_unique_constraint_named_broken_temp(self):
-        inspector = Inspector(testing.db)
+        inspector = inspect(testing.db)
         eq_(
             inspector.get_unique_constraints("g"),
             [{"column_names": ["x"], "name": "foo_gx"}],
         )
 
     def test_unique_constraint_unnamed_inline(self):
-        inspector = Inspector(testing.db)
+        inspector = inspect(testing.db)
         eq_(
             inspector.get_unique_constraints("d"),
             [{"column_names": ["x"], "name": None}],
         )
 
     def test_unique_constraint_unnamed_inline_quoted(self):
-        inspector = Inspector(testing.db)
+        inspector = inspect(testing.db)
         eq_(
             inspector.get_unique_constraints("d1"),
             [{"column_names": ["some ( STUPID n,ame"], "name": None}],
@@ -2293,42 +2292,42 @@ class ConstraintReflectionTest(fixtures.TestBase):
         )
 
     def test_unique_constraint_unnamed_normal(self):
-        inspector = Inspector(testing.db)
+        inspector = inspect(testing.db)
         eq_(
             inspector.get_unique_constraints("m"),
             [{"column_names": ["x"], "name": None}],
         )
 
     def test_unique_constraint_unnamed_normal_temporary(self):
-        inspector = Inspector(testing.db)
+        inspector = inspect(testing.db)
         eq_(
             inspector.get_unique_constraints("n"),
             [{"column_names": ["x"], "name": None}],
         )
 
     def test_primary_key_constraint_named(self):
-        inspector = Inspector(testing.db)
+        inspector = inspect(testing.db)
         eq_(
             inspector.get_pk_constraint("p"),
             {"constrained_columns": ["id"], "name": "pk_name"},
         )
 
     def test_primary_key_constraint_unnamed(self):
-        inspector = Inspector(testing.db)
+        inspector = inspect(testing.db)
         eq_(
             inspector.get_pk_constraint("q"),
             {"constrained_columns": ["id"], "name": None},
         )
 
     def test_primary_key_constraint_no_pk(self):
-        inspector = Inspector(testing.db)
+        inspector = inspect(testing.db)
         eq_(
             inspector.get_pk_constraint("d"),
             {"constrained_columns": [], "name": None},
         )
 
     def test_check_constraint(self):
-        inspector = Inspector(testing.db)
+        inspector = inspect(testing.db)
         eq_(
             inspector.get_check_constraints("cp"),
             [
diff --git a/test/engine/test_bind.py b/test/engine/test_bind.py
index 39acfed2c..956874846 100644
--- a/test/engine/test_bind.py
+++ b/test/engine/test_bind.py
@@ -26,14 +26,6 @@ class BindTest(fixtures.TestBase):
             assert not conn.closed
         assert conn.closed
 
-    def test_bind_close_conn(self):
-        e = testing.db
-        conn = e.connect()
-        with conn.connect() as c2:
-            assert not c2.closed
-        assert not conn.closed
-        assert c2.closed
-
     def test_create_drop_explicit(self):
         metadata = MetaData()
         table = Table("test_table", metadata, Column("foo", Integer))
diff --git a/test/engine/test_deprecations.py b/test/engine/test_deprecations.py
index 6fbf1be5b..884838351 100644
--- a/test/engine/test_deprecations.py
+++ b/test/engine/test_deprecations.py
@@ -13,6 +13,7 @@ from sqlalchemy import String
 from sqlalchemy import testing
 from sqlalchemy import text
 from sqlalchemy import TypeDecorator
+from sqlalchemy.engine import reflection
 from sqlalchemy.engine.base import Engine
 from sqlalchemy.engine.mock import MockConnection
 from sqlalchemy.testing import assert_raises
@@ -20,6 +21,7 @@ from sqlalchemy.testing import assert_raises_message
 from sqlalchemy.testing import engines
 from sqlalchemy.testing import eq_
 from sqlalchemy.testing import fixtures
+from sqlalchemy.testing import is_
 from sqlalchemy.testing import is_false
 from sqlalchemy.testing import is_true
 from sqlalchemy.testing.mock import Mock
@@ -31,6 +33,52 @@ class SomeException(Exception):
     pass
 
 
+class ConnectionlessDeprecationTest(fixtures.TestBase):
+    """test various things associated with "connectionless" executions."""
+
+    def test_inspector_constructor_engine(self):
+        with testing.expect_deprecated(
+            r"The __init__\(\) method on Inspector is deprecated and will "
+            r"be removed in a future release."
+        ):
+            i1 = reflection.Inspector(testing.db)
+
+        is_(i1.bind, testing.db)
+
+    def test_inspector_constructor_connection(self):
+        with testing.db.connect() as conn:
+            with testing.expect_deprecated(
+                r"The __init__\(\) method on Inspector is deprecated and "
+                r"will be removed in a future release."
+            ):
+                i1 = reflection.Inspector(conn)
+
+            is_(i1.bind, conn)
+            is_(i1.engine, testing.db)
+
+    def test_inspector_from_engine(self):
+        with testing.expect_deprecated(
+            r"The from_engine\(\) method on Inspector is deprecated and will "
+            r"be removed in a future release."
+        ):
+            i1 = reflection.Inspector.from_engine(testing.db)
+
+        is_(i1.bind, testing.db)
+
+    def test_bind_close_conn(self):
+        e = testing.db
+        conn = e.connect()
+
+        with testing.expect_deprecated(
+            r"The .close\(\) method on a so-called 'branched' "
+            "connection is deprecated"
+        ):
+            with conn.connect() as c2:
+                assert not c2.closed
+        assert not conn.closed
+        assert c2.closed
+
+
 class CreateEngineTest(fixtures.TestBase):
     def test_strategy_keyword_mock(self):
         def executor(x, y):
@@ -512,3 +560,17 @@ class DeprecatedReflectionTest(fixtures.TablesTest):
         ):
             table_names = testing.db.table_names()
         is_true(set(table_names).issuperset(metadata.tables))
+
+
+class ExecutionOptionsTest(fixtures.TestBase):
+    def test_branched_connection_execution_options(self):
+        engine = engines.testing_engine("sqlite://")
+
+        conn = engine.connect()
+        c2 = conn.execution_options(foo="bar")
+
+        with testing.expect_deprecated_20(
+            r"The Connection.connect\(\) function/method is considered "
+        ):
+            c2_branch = c2.connect()
+        eq_(c2_branch._execution_options, {"foo": "bar"})
diff --git a/test/engine/test_execute.py b/test/engine/test_execute.py
index ac9f034fe..5acd14177 100644
--- a/test/engine/test_execute.py
+++ b/test/engine/test_execute.py
@@ -1262,14 +1262,6 @@ class ExecutionOptionsTest(fixtures.TestBase):
         eq_(c1._execution_options, {"foo": "bar"})
         eq_(c2._execution_options, {"foo": "bar", "bat": "hoho"})
 
-    def test_branched_connection_execution_options(self):
-        engine = testing_engine("sqlite://")
-
-        conn = engine.connect()
-        c2 = conn.execution_options(foo="bar")
-        c2_branch = c2.connect()
-        eq_(c2_branch._execution_options, {"foo": "bar"})
-
     def test_get_engine_execution_options(self):
         engine = testing_engine("sqlite://")
         engine.dialect = Mock()
diff --git a/test/engine/test_reflection.py b/test/engine/test_reflection.py
index c137488ec..301614061 100644
--- a/test/engine/test_reflection.py
+++ b/test/engine/test_reflection.py
@@ -34,9 +34,6 @@ from sqlalchemy.testing.schema import Table
 from sqlalchemy.util import ue
 
 
-metadata, users = None, None
-
-
 class ReflectionTest(fixtures.TestBase, ComparesTables):
     __backend__ = True
 
@@ -906,15 +903,15 @@ class ReflectionTest(fixtures.TestBase, ComparesTables):
             test_needs_fk=True,
         )
 
-        meta.create_all()
-        meta2 = MetaData(testing.db)
+        meta.create_all(testing.db)
+        meta2 = MetaData()
         a2 = Table(
             "addresses",
             meta2,
             Column("user_id", sa.Integer, sa.ForeignKey("users.id")),
-            autoload=True,
+            autoload_with=testing.db,
         )
-        u2 = Table("users", meta2, autoload=True)
+        u2 = Table("users", meta2, autoload_with=testing.db)
         s = sa.select([a2]).subquery()
 
         assert s.c.user_id is not None
@@ -926,19 +923,19 @@ class ReflectionTest(fixtures.TestBase, ComparesTables):
         assert list(a2.c.user_id.foreign_keys)[0].parent is a2.c.user_id
         assert u2.join(a2).onclause.compare(u2.c.id == a2.c.user_id)
 
-        meta2 = MetaData(testing.db)
+        meta2 = MetaData()
         u2 = Table(
             "users",
             meta2,
             Column("id", sa.Integer, primary_key=True),
-            autoload=True,
+            autoload_with=testing.db,
         )
         a2 = Table(
             "addresses",
             meta2,
             Column("id", sa.Integer, primary_key=True),
             Column("user_id", sa.Integer, sa.ForeignKey("users.id")),
-            autoload=True,
+            autoload_with=testing.db,
         )
         s = sa.select([a2]).subquery()
 
@@ -1012,29 +1009,28 @@ class ReflectionTest(fixtures.TestBase, ComparesTables):
             for attr in test_attrs:
                 eq_(getattr(fk, attr), getattr(ref, attr))
 
+    @testing.provide_metadata
     def test_pks_not_uniques(self):
         """test that primary key reflection not tripped up by unique
         indexes"""
 
-        testing.db.execute(
-            """
-        CREATE TABLE book (
-            id INTEGER NOT NULL,
-            title VARCHAR(100) NOT NULL,
-            series INTEGER,
-            series_id INTEGER,
-            UNIQUE(series, series_id),
-            PRIMARY KEY(id)
-        )"""
-        )
-        try:
-            metadata = MetaData(bind=testing.db)
-            book = Table("book", metadata, autoload=True)
-            assert book.primary_key.contains_column(book.c.id)
-            assert not book.primary_key.contains_column(book.c.series)
-            assert len(book.primary_key) == 1
-        finally:
-            testing.db.execute("drop table book")
+        with testing.db.begin() as conn:
+            conn.execute(
+                """
+                CREATE TABLE book (
+                    id INTEGER NOT NULL,
+                    title VARCHAR(100) NOT NULL,
+                    series INTEGER,
+                    series_id INTEGER,
+                    UNIQUE(series, series_id),
+                    PRIMARY KEY(id)
+                )"""
+            )
+
+        book = Table("book", self.metadata, autoload_with=testing.db)
+        assert book.primary_key.contains_column(book.c.id)
+        assert not book.primary_key.contains_column(book.c.series)
+        eq_(len(book.primary_key), 1)
 
     def test_fk_error(self):
         metadata = MetaData(testing.db)
@@ -1054,30 +1050,28 @@ class ReflectionTest(fixtures.TestBase, ComparesTables):
             metadata.create_all,
         )
 
+    @testing.provide_metadata
     def test_composite_pks(self):
         """test reflection of a composite primary key"""
 
-        testing.db.execute(
-            """
-        CREATE TABLE book (
-            id INTEGER NOT NULL,
-            isbn VARCHAR(50) NOT NULL,
-            title VARCHAR(100) NOT NULL,
-            series INTEGER NOT NULL,
-            series_id INTEGER NOT NULL,
-            UNIQUE(series, series_id),
-            PRIMARY KEY(id, isbn)
-        )"""
-        )
-        try:
-            metadata = MetaData(bind=testing.db)
-            book = Table("book", metadata, autoload=True)
-            assert book.primary_key.contains_column(book.c.id)
-            assert book.primary_key.contains_column(book.c.isbn)
-            assert not book.primary_key.contains_column(book.c.series)
-            assert len(book.primary_key) == 2
-        finally:
-            testing.db.execute("drop table book")
+        with testing.db.begin() as conn:
+            conn.execute(
+                """
+                CREATE TABLE book (
+                    id INTEGER NOT NULL,
+                    isbn VARCHAR(50) NOT NULL,
+                    title VARCHAR(100) NOT NULL,
+                    series INTEGER NOT NULL,
+                    series_id INTEGER NOT NULL,
+                    UNIQUE(series, series_id),
+                    PRIMARY KEY(id, isbn)
+                )"""
+            )
+        book = Table("book", self.metadata, autoload_with=testing.db)
+        assert book.primary_key.contains_column(book.c.id)
+        assert book.primary_key.contains_column(book.c.isbn)
+        assert not book.primary_key.contains_column(book.c.series)
+        eq_(len(book.primary_key), 2)
 
     @testing.exclude("mysql", "<", (4, 1, 1), "innodb funkiness")
     @testing.provide_metadata
@@ -1747,16 +1741,17 @@ class SchemaTest(fixtures.TestBase):
                 "dialect %s doesn't have a has_schema method"
                 % testing.db.dialect.name
             )
-        eq_(
-            testing.db.dialect.has_schema(
-                testing.db, testing.config.test_schema
-            ),
-            True,
-        )
-        eq_(
-            testing.db.dialect.has_schema(testing.db, "sa_fake_schema_123"),
-            False,
-        )
+        with testing.db.connect() as conn:
+            eq_(
+                testing.db.dialect.has_schema(
+                    conn, testing.config.test_schema
+                ),
+                True,
+            )
+            eq_(
+                testing.db.dialect.has_schema(conn, "sa_fake_schema_123"),
+                False,
+            )
 
     @testing.requires.schemas
     @testing.requires.cross_schema_fk_reflection
@@ -2018,24 +2013,29 @@ def createIndexes(con, schema=None):
 
 @testing.requires.views
 def _create_views(con, schema=None):
-    for table_name in ("users", "email_addresses"):
-        fullname = table_name
-        if schema:
-            fullname = "%s.%s" % (schema, table_name)
-        view_name = fullname + "_v"
-        query = "CREATE VIEW %s AS SELECT * FROM %s" % (view_name, fullname)
-        con.execute(sa.sql.text(query))
+    with testing.db.connect() as conn:
+        for table_name in ("users", "email_addresses"):
+            fullname = table_name
+            if schema:
+                fullname = "%s.%s" % (schema, table_name)
+            view_name = fullname + "_v"
+            query = "CREATE VIEW %s AS SELECT * FROM %s" % (
+                view_name,
+                fullname,
+            )
+            conn.execute(sa.sql.text(query))
 
 
 @testing.requires.views
 def _drop_views(con, schema=None):
-    for table_name in ("email_addresses", "users"):
-        fullname = table_name
-        if schema:
-            fullname = "%s.%s" % (schema, table_name)
-        view_name = fullname + "_v"
-        query = "DROP VIEW %s" % view_name
-        con.execute(sa.sql.text(query))
+    with testing.db.connect() as conn:
+        for table_name in ("email_addresses", "users"):
+            fullname = table_name
+            if schema:
+                fullname = "%s.%s" % (schema, table_name)
+            view_name = fullname + "_v"
+            query = "DROP VIEW %s" % view_name
+            conn.execute(sa.sql.text(query))
 
 
 class ReverseCasingReflectTest(fixtures.TestBase, AssertsCompiledSQL):
diff --git a/test/sql/test_defaults.py b/test/sql/test_defaults.py
index b31b070d8..2fded335b 100644
--- a/test/sql/test_defaults.py
+++ b/test/sql/test_defaults.py
@@ -163,12 +163,7 @@ class DefaultTest(fixtures.TestBase):
 
         def mydefault_using_connection(ctx):
             conn = ctx.connection
-            try:
-                return conn.execute(sa.select([sa.text("12")])).scalar()
-            finally:
-                # ensure a "close()" on this connection does nothing,
-                # since its a "branched" connection
-                conn.close()
+            return conn.execute(sa.select([sa.text("12")])).scalar()
 
         use_function_defaults = testing.against("postgresql", "mssql")
         is_oracle = testing.against("oracle")
diff --git a/test/sql/test_deprecations.py b/test/sql/test_deprecations.py
index b2b1f470b..b058cbe1b 100644
--- a/test/sql/test_deprecations.py
+++ b/test/sql/test_deprecations.py
@@ -1494,3 +1494,38 @@ class PositionalTextTest(fixtures.TablesTest):
             "Could not locate column in row for column 'text1.b'",
             lambda: row[text1.c.b],
         )
+
+
+class DefaultTest(fixtures.TestBase):
+    __backend__ = True
+
+    @testing.provide_metadata
+    def test_close_on_branched(self):
+        metadata = self.metadata
+
+        def mydefault_using_connection(ctx):
+            conn = ctx.connection
+            try:
+                return conn.execute(select([text("12")])).scalar()
+            finally:
+                # ensure a "close()" on this connection does nothing,
+                # since its a "branched" connection
+                conn.close()
+
+        table = Table(
+            "foo",
+            metadata,
+            Column("x", Integer),
+            Column("y", Integer, default=mydefault_using_connection),
+        )
+
+        metadata.create_all(testing.db)
+        with testing.db.connect() as conn:
+            with testing.expect_deprecated(
+                r"The .close\(\) method on a so-called 'branched' "
+                r"connection is deprecated as of 1.4, as are "
+                r"'branched' connections overall"
+            ):
+                conn.execute(table.insert().values(x=5))
+
+            eq_(conn.execute(select([table])).first(), (5, 12))