Method test_initialize_connect_calls

test/engine/test_execute.py:3834–3901

test for :ticket:`5497`, on_connect not called twice

(self)

    @testing.only_on("sqlite+pysqlite")
    def test_initialize_connect_calls(self):
        """test for :ticket:`5497`, on_connect not called twice"""

        m1 = Mock()
        cls_ = testing.db.dialect.__class__

        class SomeDialect(cls_):
            def initialize(self, connection):
                super().initialize(connection)
                m1.initialize(connection)

            def on_connect(self):
                oc = super().on_connect()

                def my_on_connect(conn):
                    if oc:
                        oc(conn)
                    m1.on_connect(conn)

                return my_on_connect

        u1 = Mock(
            username=None,
            password=None,
            host=None,
            port=None,
            query={},
            database=None,
            _instantiate_plugins=lambda kw: (u1, [], kw),
            _get_entrypoint=Mock(
                return_value=Mock(get_dialect_cls=lambda u: SomeDialect)
            ),
        )
        eng = create_engine(u1, poolclass=QueuePool)
        # make sure other dialects aren't getting pulled in here
        eq_(eng.name, "sqlite")
        c = eng.connect()
        dbapi_conn_one = c.connection.dbapi_connection
        c.close()

        eq_(
            m1.mock_calls,
            [call.on_connect(dbapi_conn_one), call.initialize(mock.ANY)],
        )

        c = eng.connect()

        eq_(
            m1.mock_calls,
            [call.on_connect(dbapi_conn_one), call.initialize(mock.ANY)],
        )

        c2 = eng.connect()
        dbapi_conn_two = c2.connection.dbapi_connection

        is_not(dbapi_conn_one, dbapi_conn_two)

        eq_(