Method run_test

lib/sqlalchemy/testing/fixtures/base.py:265–382

(subject, trans_on_subject, execute_on_subject)

        t.create(eng)

        def run_test(subject, trans_on_subject, execute_on_subject):
            with subject.begin() as trans:
                if begin_nested:
                    if not config.requirements.savepoints.enabled:
                        config.skip_test("savepoints not enabled")
                    if execute_on_subject:
                        nested_trans = subject.begin_nested()
                    else:
                        nested_trans = trans.begin_nested()

                    with nested_trans:
                        if execute_on_subject:
                            subject.execute(t.insert(), {"data": 10})
                        else:
                            trans.execute(t.insert(), {"data": 10})

                        # for nested trans, we always commit/rollback on the
                        # "nested trans" object itself.
                        # only Session(future=False) will affect savepoint
                        # transaction for session.commit/rollback

                        if rollback:
                            nested_trans.rollback()
                        else:
                            nested_trans.commit()

                        if second_operation != "none":
                            with assertions.expect_raises_message(
                                sa.exc.InvalidRequestError,
                                "Can't operate on closed transaction "
                                "inside context "
                                "manager.  Please complete the context "
                                "manager "
                                "before emitting further commands.",
                            ):
                                if second_operation == "execute":
                                    if execute_on_subject:
                                        subject.execute(
                                            t.insert(), {"data": 12}
                                        )
                                    else:
                                        trans.execute(t.insert(), {"data": 12})
                                elif second_operation == "begin":
                                    if execute_on_subject:
                                        subject.begin_nested()
                                    else:
                                        trans.begin_nested()

                    # outside the nested trans block, but still inside the
                    # transaction block, we can run SQL, and it will be
                    # committed
                    if execute_on_subject:
                        subject.execute(t.insert(), {"data": 14})
                    else:
                        trans.execute(t.insert(), {"data": 14})

                else:
                    if execute_on_subject: