Add deprecation warnings to all deprecated APIs

A large change throughout the library has ensured that all objects, parameters,
and behaviors which have been noted as deprecated or legacy now emit
``DeprecationWarning`` warnings when invoked.   As the Python 3 interpreter now
defaults to displaying deprecation warnings, as well as that modern test suites
based on tools like tox and pytest tend to display deprecation warnings,
this change should make it easier to note what API features are obsolete.

See the notes added to the changelog and migration notes for further
details.

Fixes: #4393
Change-Id: If0ea11a1fc24f9a8029352eeadfc49a7a54c0a1b

1 files changed, 17 insertions, 359 deletions

diff --git a/test/engine/test_execute.py b/test/engine/test_execute.py
index 8613be5bc..061dae005 100644
--- a/test/engine/test_execute.py
+++ b/test/engine/test_execute.py
@@ -22,7 +22,6 @@ from sqlalchemy import util
 from sqlalchemy import VARCHAR
 from sqlalchemy.engine import default
 from sqlalchemy.engine.base import Engine
-from sqlalchemy.interfaces import ConnectionProxy
 from sqlalchemy.sql import column
 from sqlalchemy.sql import literal
 from sqlalchemy.testing import assert_raises
@@ -372,8 +371,7 @@ class ExecuteTest(fixtures.TestBase):
         def _go(conn):
             assert_raises_message(
                 tsa.exc.StatementError,
-                r"\(test.engine.test_execute.SomeException\) "
-                r"nope \[SQL\: u?'SELECT 1 ",
+                r"\(.*.SomeException\) " r"nope \[SQL\: u?'SELECT 1 ",
                 conn.execute,
                 select([1]).where(column("foo") == literal("bar", MyType())),
             )
@@ -613,7 +611,7 @@ class ExecuteTest(fixtures.TestBase):
         eng = engines.testing_engine(
             options={"execution_options": {"foo": "bar"}}
         )
-        with eng.contextual_connect() as conn:
+        with eng.connect() as conn:
             eq_(conn._execution_options["foo"], "bar")
             eq_(
                 conn.execution_options(bat="hoho")._execution_options["foo"],
@@ -628,7 +626,7 @@ class ExecuteTest(fixtures.TestBase):
                 "hoho",
             )
             eng.update_execution_options(foo="hoho")
-            conn = eng.contextual_connect()
+            conn = eng.connect()
             eq_(conn._execution_options["foo"], "hoho")
 
     @testing.requires.ad_hoc_engines
@@ -787,32 +785,6 @@ class ConvenienceExecuteTest(fixtures.TablesTest):
         )
         self._assert_no_data()
 
-    def test_transaction_tlocal_engine_ctx_commit(self):
-        fn = self._trans_fn()
-        engine = engines.testing_engine(
-            options=dict(strategy="threadlocal", pool=testing.db.pool)
-        )
-        ctx = engine.begin()
-        testing.run_as_contextmanager(ctx, fn, 5, value=8)
-        self._assert_fn(5, value=8)
-
-    def test_transaction_tlocal_engine_ctx_rollback(self):
-        fn = self._trans_rollback_fn()
-        engine = engines.testing_engine(
-            options=dict(strategy="threadlocal", pool=testing.db.pool)
-        )
-        ctx = engine.begin()
-        assert_raises_message(
-            Exception,
-            "breakage",
-            testing.run_as_contextmanager,
-            ctx,
-            fn,
-            5,
-            value=8,
-        )
-        self._assert_no_data()
-
     def test_transaction_connection_ctx_commit(self):
         fn = self._trans_fn(True)
         with testing.db.connect() as conn:
@@ -1495,11 +1467,16 @@ class EngineEventsTest(fixtures.TestBase):
         ):
             cursor_stmts.append((str(statement), parameters, None))
 
+        with testing.expect_deprecated(
+            "The 'threadlocal' engine strategy is deprecated"
+        ):
+            tl_engine = engines.testing_engine(
+                options=dict(implicit_returning=False, strategy="threadlocal")
+            )
+
         for engine in [
             engines.testing_engine(options=dict(implicit_returning=False)),
-            engines.testing_engine(
-                options=dict(implicit_returning=False, strategy="threadlocal")
-            ),
+            tl_engine,
             engines.testing_engine(
                 options=dict(implicit_returning=False)
             ).connect(),
@@ -1999,63 +1976,6 @@ class HandleErrorTest(fixtures.TestBase):
         Engine.dispatch._clear()
         Engine._has_events = False
 
-    def test_legacy_dbapi_error(self):
-        engine = engines.testing_engine()
-        canary = Mock()
-
-        event.listen(engine, "dbapi_error", canary)
-
-        with engine.connect() as conn:
-            try:
-                conn.execute("SELECT FOO FROM I_DONT_EXIST")
-                assert False
-            except tsa.exc.DBAPIError as e:
-                eq_(canary.mock_calls[0][1][5], e.orig)
-                eq_(canary.mock_calls[0][1][2], "SELECT FOO FROM I_DONT_EXIST")
-
-    def test_legacy_dbapi_error_no_ad_hoc_context(self):
-        engine = engines.testing_engine()
-
-        listener = Mock(return_value=None)
-        event.listen(engine, "dbapi_error", listener)
-
-        nope = SomeException("nope")
-
-        class MyType(TypeDecorator):
-            impl = Integer
-
-            def process_bind_param(self, value, dialect):
-                raise nope
-
-        with engine.connect() as conn:
-            assert_raises_message(
-                tsa.exc.StatementError,
-                r"\(test.engine.test_execute.SomeException\) "
-                r"nope \[SQL\: u?'SELECT 1 ",
-                conn.execute,
-                select([1]).where(column("foo") == literal("bar", MyType())),
-            )
-        # no legacy event
-        eq_(listener.mock_calls, [])
-
-    def test_legacy_dbapi_error_non_dbapi_error(self):
-        engine = engines.testing_engine()
-
-        listener = Mock(return_value=None)
-        event.listen(engine, "dbapi_error", listener)
-
-        nope = TypeError("I'm not a DBAPI error")
-        with engine.connect() as c:
-            c.connection.cursor = Mock(
-                return_value=Mock(execute=Mock(side_effect=nope))
-            )
-
-            assert_raises_message(
-                TypeError, "I'm not a DBAPI error", c.execute, "select "
-            )
-        # no legacy event
-        eq_(listener.mock_calls, [])
-
     def test_handle_error(self):
         engine = engines.testing_engine()
         canary = Mock(return_value=None)
@@ -2249,8 +2169,7 @@ class HandleErrorTest(fixtures.TestBase):
         with engine.connect() as conn:
             assert_raises_message(
                 tsa.exc.StatementError,
-                r"\(test.engine.test_execute.SomeException\) "
-                r"nope \[SQL\: u?'SELECT 1 ",
+                r"\(.*.SomeException\) " r"nope \[SQL\: u?'SELECT 1 ",
                 conn.execute,
                 select([1]).where(column("foo") == literal("bar", MyType())),
             )
@@ -2571,27 +2490,15 @@ class HandleInvalidatedOnConnectTest(fixtures.TestBase):
         except tsa.exc.DBAPIError:
             assert conn.invalidated
 
-    def _test_dont_touch_non_dbapi_exception_on_connect(self, connect_fn):
+    def test_dont_touch_non_dbapi_exception_on_connect(self):
         dbapi = self.dbapi
         dbapi.connect = Mock(side_effect=TypeError("I'm not a DBAPI error"))
 
         e = create_engine("sqlite://", module=dbapi)
         e.dialect.is_disconnect = is_disconnect = Mock()
-        assert_raises_message(
-            TypeError, "I'm not a DBAPI error", connect_fn, e
-        )
+        assert_raises_message(TypeError, "I'm not a DBAPI error", e.connect)
         eq_(is_disconnect.call_count, 0)
 
-    def test_dont_touch_non_dbapi_exception_on_connect(self):
-        self._test_dont_touch_non_dbapi_exception_on_connect(
-            lambda engine: engine.connect()
-        )
-
-    def test_dont_touch_non_dbapi_exception_on_contextual_connect(self):
-        self._test_dont_touch_non_dbapi_exception_on_connect(
-            lambda engine: engine.contextual_connect()
-        )
-
     def test_ensure_dialect_does_is_disconnect_no_conn(self):
         """test that is_disconnect() doesn't choke if no connection,
         cursor given."""
@@ -2601,275 +2508,26 @@ class HandleInvalidatedOnConnectTest(fixtures.TestBase):
             dbapi.OperationalError("test"), None, None
         )
 
-    def _test_invalidate_on_connect(self, connect_fn):
+    def test_invalidate_on_connect(self):
         """test that is_disconnect() is called during connect.
 
         interpretation of connection failures are not supported by
         every backend.
 
         """
-
         dbapi = self.dbapi
         dbapi.connect = Mock(
             side_effect=self.ProgrammingError(
                 "Cannot operate on a closed database."
             )
         )
+        e = create_engine("sqlite://", module=dbapi)
         try:
-            connect_fn(create_engine("sqlite://", module=dbapi))
+            e.connect()
             assert False
         except tsa.exc.DBAPIError as de:
             assert de.connection_invalidated
 
-    def test_invalidate_on_connect(self):
-        """test that is_disconnect() is called during connect.
-
-        interpretation of connection failures are not supported by
-        every backend.
-
-        """
-        self._test_invalidate_on_connect(lambda engine: engine.connect())
-
-    def test_invalidate_on_contextual_connect(self):
-        """test that is_disconnect() is called during connect.
-
-        interpretation of connection failures are not supported by
-        every backend.
-
-        """
-        self._test_invalidate_on_connect(
-            lambda engine: engine.contextual_connect()
-        )
-
-
-class ProxyConnectionTest(fixtures.TestBase):
-
-    """These are the same tests as EngineEventsTest, except using
-    the deprecated ConnectionProxy interface.
-
-    """
-
-    __requires__ = ("ad_hoc_engines",)
-    __prefer_requires__ = ("two_phase_transactions",)
-
-    @testing.uses_deprecated(r".*Use event.listen")
-    @testing.fails_on("firebird", "Data type unknown")
-    def test_proxy(self):
-
-        stmts = []
-        cursor_stmts = []
-
-        class MyProxy(ConnectionProxy):
-            def execute(
-                self, conn, execute, clauseelement, *multiparams, **params
-            ):
-                stmts.append((str(clauseelement), params, multiparams))
-                return execute(clauseelement, *multiparams, **params)
-
-            def cursor_execute(
-                self,
-                execute,
-                cursor,
-                statement,
-                parameters,
-                context,
-                executemany,
-            ):
-                cursor_stmts.append((str(statement), parameters, None))
-                return execute(cursor, statement, parameters, context)
-
-        def assert_stmts(expected, received):
-            for stmt, params, posn in expected:
-                if not received:
-                    assert False, "Nothing available for stmt: %s" % stmt
-                while received:
-                    teststmt, testparams, testmultiparams = received.pop(0)
-                    teststmt = (
-                        re.compile(r"[\n\t ]+", re.M)
-                        .sub(" ", teststmt)
-                        .strip()
-                    )
-                    if teststmt.startswith(stmt) and (
-                        testparams == params or testparams == posn
-                    ):
-                        break
-
-        for engine in (
-            engines.testing_engine(
-                options=dict(implicit_returning=False, proxy=MyProxy())
-            ),
-            engines.testing_engine(
-                options=dict(
-                    implicit_returning=False,
-                    proxy=MyProxy(),
-                    strategy="threadlocal",
-                )
-            ),
-        ):
-            m = MetaData(engine)
-            t1 = Table(
-                "t1",
-                m,
-                Column("c1", Integer, primary_key=True),
-                Column(
-                    "c2",
-                    String(50),
-                    default=func.lower("Foo"),
-                    primary_key=True,
-                ),
-            )
-            m.create_all()
-            try:
-                t1.insert().execute(c1=5, c2="some data")
-                t1.insert().execute(c1=6)
-                eq_(
-                    engine.execute("select * from t1").fetchall(),
-                    [(5, "some data"), (6, "foo")],
-                )
-            finally:
-                m.drop_all()
-            engine.dispose()
-            compiled = [
-                ("CREATE TABLE t1", {}, None),
-                (
-                    "INSERT INTO t1 (c1, c2)",
-                    {"c2": "some data", "c1": 5},
-                    None,
-                ),
-                ("INSERT INTO t1 (c1, c2)", {"c1": 6}, None),
-                ("select * from t1", {}, None),
-                ("DROP TABLE t1", {}, None),
-            ]
-
-            cursor = [
-                ("CREATE TABLE t1", {}, ()),
-                (
-                    "INSERT INTO t1 (c1, c2)",
-                    {"c2": "some data", "c1": 5},
-                    (5, "some data"),
-                ),
-                ("SELECT lower", {"lower_1": "Foo"}, ("Foo",)),
-                (
-                    "INSERT INTO t1 (c1, c2)",
-                    {"c2": "foo", "c1": 6},
-                    (6, "foo"),
-                ),
-                ("select * from t1", {}, ()),
-                ("DROP TABLE t1", {}, ()),
-            ]
-
-            assert_stmts(compiled, stmts)
-            assert_stmts(cursor, cursor_stmts)
-
-    @testing.uses_deprecated(r".*Use event.listen")
-    def test_options(self):
-        canary = []
-
-        class TrackProxy(ConnectionProxy):
-            def __getattribute__(self, key):
-                fn = object.__getattribute__(self, key)
-
-                def go(*arg, **kw):
-                    canary.append(fn.__name__)
-                    return fn(*arg, **kw)
-
-                return go
-
-        engine = engines.testing_engine(options={"proxy": TrackProxy()})
-        conn = engine.connect()
-        c2 = conn.execution_options(foo="bar")
-        eq_(c2._execution_options, {"foo": "bar"})
-        c2.execute(select([1]))
-        c3 = c2.execution_options(bar="bat")
-        eq_(c3._execution_options, {"foo": "bar", "bar": "bat"})
-        eq_(canary, ["execute", "cursor_execute"])
-
-    @testing.uses_deprecated(r".*Use event.listen")
-    def test_transactional(self):
-        canary = []
-
-        class TrackProxy(ConnectionProxy):
-            def __getattribute__(self, key):
-                fn = object.__getattribute__(self, key)
-
-                def go(*arg, **kw):
-                    canary.append(fn.__name__)
-                    return fn(*arg, **kw)
-
-                return go
-
-        engine = engines.testing_engine(options={"proxy": TrackProxy()})
-        conn = engine.connect()
-        trans = conn.begin()
-        conn.execute(select([1]))
-        trans.rollback()
-        trans = conn.begin()
-        conn.execute(select([1]))
-        trans.commit()
-
-        eq_(
-            canary,
-            [
-                "begin",
-                "execute",
-                "cursor_execute",
-                "rollback",
-                "begin",
-                "execute",
-                "cursor_execute",
-                "commit",
-            ],
-        )
-
-    @testing.uses_deprecated(r".*Use event.listen")
-    @testing.requires.savepoints
-    @testing.requires.two_phase_transactions
-    def test_transactional_advanced(self):
-        canary = []
-
-        class TrackProxy(ConnectionProxy):
-            def __getattribute__(self, key):
-                fn = object.__getattribute__(self, key)
-
-                def go(*arg, **kw):
-                    canary.append(fn.__name__)
-                    return fn(*arg, **kw)
-
-                return go
-
-        engine = engines.testing_engine(options={"proxy": TrackProxy()})
-        conn = engine.connect()
-
-        trans = conn.begin()
-        trans2 = conn.begin_nested()
-        conn.execute(select([1]))
-        trans2.rollback()
-        trans2 = conn.begin_nested()
-        conn.execute(select([1]))
-        trans2.commit()
-        trans.rollback()
-
-        trans = conn.begin_twophase()
-        conn.execute(select([1]))
-        trans.prepare()
-        trans.commit()
-
-        canary = [t for t in canary if t not in ("cursor_execute", "execute")]
-        eq_(
-            canary,
-            [
-                "begin",
-                "savepoint",
-                "rollback_savepoint",
-                "savepoint",
-                "release_savepoint",
-                "rollback",
-                "begin_twophase",
-                "prepare_twophase",
-                "commit_twophase",
-            ],
-        )
-
 
 class DialectEventTest(fixtures.TestBase):
     @contextmanager