Method _proxy_fixture

test/sql/test_resultset.py:3230–3255

(self, cls)

    @contextmanager
    def _proxy_fixture(self, cls):
        self.table = self.tables.test

        class ExcCtx(default.DefaultExecutionContext):
            def post_exec(self):
                if cls is _cursor.CursorFetchStrategy:
                    pass
                elif cls is _cursor.BufferedRowCursorFetchStrategy:
                    self.cursor_fetch_strategy = cls(
                        self.cursor, self.execution_options
                    )
                elif cls is _cursor.FullyBufferedCursorFetchStrategy:
                    self.cursor_fetch_strategy = cls(
                        self.cursor,
                        self.cursor.description,
                        self.cursor.fetchall(),
                    )
                else:
                    assert False

        self.patcher = patch.object(
            self.engine.dialect, "execution_ctx_cls", ExcCtx
        )

        with self.patcher:
            yield

    def _test_proxy(self, cls):
        with self._proxy_fixture(cls):