Method _test_proxy

test/sql/test_resultset.py:3170–3220

(self, cls)

            yield

    def _test_proxy(self, cls):
        with self._proxy_fixture(cls):
            rows = []
            with self.engine.connect() as conn:
                r = conn.execute(select(self.table))
                assert isinstance(r.cursor_strategy, cls)
                for i in range(5):
                    rows.append(r.fetchone())
                eq_(rows, [(i, "t_%d" % i) for i in range(1, 6)])

                rows = r.fetchmany(3)
                eq_(rows, [(i, "t_%d" % i) for i in range(6, 9)])

                rows = r.fetchall()
                eq_(rows, [(i, "t_%d" % i) for i in range(9, 12)])

                r = conn.execute(select(self.table))
                rows = r.fetchmany(None)
                eq_(rows[0], (1, "t_1"))
                # number of rows here could be one, or the whole thing
                assert len(rows) == 1 or len(rows) == 11

                r = conn.execute(select(self.table).limit(1))
                r.fetchone()
                eq_(r.fetchone(), None)

                r = conn.execute(select(self.table).limit(5))
                rows = r.fetchmany(6)
                eq_(rows, [(i, "t_%d" % i) for i in range(1, 6)])

                # result keeps going just fine with blank results...
                eq_(r.fetchmany(2), [])

                eq_(r.fetchmany(2), [])

                eq_(r.fetchall(), [])

                eq_(r.fetchone(), None)

                # until we close
                r.close()

                self._assert_result_closed(r)

                r = conn.execute(select(self.table).limit(5))
                eq_(r.first(), (1, "t_1"))
                self._assert_result_closed(r)

                r = conn.execute(select(self.table).limit(5))
                eq_(r.scalar(), 1)
                self._assert_result_closed(r)

    def _assert_result_closed(self, r):
        assert_raises_message(