Method test_execution_options

test/orm/test_session.py:772–864

(
        self,
        session_type: testing.Variation,
        merge: testing.Variation,
        method: testing.Variation,
        add_statement_options: testing.Variation,
    )

    )
    @testing.variation("add_statement_options", [True, False])
    def test_execution_options(
        self,
        session_type: testing.Variation,
        merge: testing.Variation,
        method: testing.Variation,
        add_statement_options: testing.Variation,
    ):
        users, User = self.tables.users, self.classes.User
        self.mapper_registry.map_imperatively(User, users)

        session_execution_options = {
            "populate_existing": True,
            "autoflush": False,
            "opt1": "z",
            "opt5": "q",
        }

        expected_opts = session_execution_options

        if add_statement_options:
            statement_options = {"opt2": "w", "opt4": "y", "opt5": "w"}
            expected_opts = {**expected_opts, **statement_options}
        else:
            statement_options = {}

        if merge:
            query_opts = {
                "compiled_cache": {},
                "opt1": "q",
                "opt2": "p",
                "opt3": "r",
                "populate_existing": False,
            }
            expected_opts = {**expected_opts, **query_opts}
        else:
            query_opts = {}

        if session_type.plain:
            sess = Session(
                testing.db, execution_options=session_execution_options
            )
        elif session_type.sessionmaker:
            maker = sessionmaker(
                testing.db, execution_options=session_execution_options
            )
            sess = maker()
        else:
            session_type.fail()

        gather_options = {}

        @event.listens_for(sess, "do_orm_execute")
        def check(ctx: ORMExecuteState) -> None:
            assert not gather_options
            gather_options.update(ctx.execution_options)

        if method.scalar:
            statement = select(User).limit(1)