Method test_bind_through_execute

test/orm/test_bind.py:415–492

(
        self, statement, expected_get_bind_args, expected_engine_name
    )

        ),
    )
    def test_bind_through_execute(
        self, statement, expected_get_bind_args, expected_engine_name
    ):
        users, Address, addresses, User = (
            self.tables.users,
            self.classes.Address,
            self.tables.addresses,
            self.classes.User,
        )

        self.mapper_registry.map_imperatively(
            User, users, properties={"addresses": relationship(Address)}
        )
        self.mapper_registry.map_imperatively(Address, addresses)

        e1 = engines.testing_engine()
        e2 = engines.testing_engine()
        e3 = engines.testing_engine()

        canary = mock.Mock()

        class GetBindSession(Session):
            def _connection_for_bind(self, bind, **kw):
                canary._connection_for_bind(bind, **kw)
                return mock.Mock()

            def get_bind(self, **kw):
                canary.get_bind(**kw)
                return Session.get_bind(self, **kw)

        sess = GetBindSession(e3)
        sess.bind_mapper(User, e1)
        sess.bind_mapper(Address, e2)

        lambda_args = dict(
            session=sess,
            User=User,
            Address=Address,
            e1=e1,
            e2=e2,
            e3=e3,
            addresses=addresses,
        )
        statement = testing.resolve_lambda(statement, **lambda_args)

        expected_get_bind_args = testing.resolve_lambda(
            expected_get_bind_args, **lambda_args
        )

        engine = {"e1": e1, "e2": e2, "e3": e3}[expected_engine_name]

        with (
            mock.patch(
                "sqlalchemy.orm.context._ORMCompileState."
                "orm_setup_cursor_result"
            ),
            mock.patch(
                "sqlalchemy.orm.context._ORMCompileState.orm_execute_statement"