(self, cls)
| 3255 | yield |
| 3256 | |
| 3257 | def _test_proxy(self, cls): |
| 3258 | with self._proxy_fixture(cls): |
| 3259 | rows = [] |
| 3260 | with self.engine.connect() as conn: |
| 3261 | r = conn.execute(select(self.table)) |
| 3262 | assert isinstance(r.cursor_strategy, cls) |
| 3263 | for i in range(5): |
| 3264 | rows.append(r.fetchone()) |
| 3265 | eq_(rows, [(i, "t_%d" % i) for i in range(1, 6)]) |
| 3266 | |
| 3267 | rows = r.fetchmany(3) |
| 3268 | eq_(rows, [(i, "t_%d" % i) for i in range(6, 9)]) |
| 3269 | |
| 3270 | rows = r.fetchall() |
| 3271 | eq_(rows, [(i, "t_%d" % i) for i in range(9, 12)]) |
| 3272 | |
| 3273 | r = conn.execute(select(self.table)) |
| 3274 | rows = r.fetchmany(None) |
| 3275 | eq_(rows[0], (1, "t_1")) |
| 3276 | # number of rows here could be one, or the whole thing |
| 3277 | assert len(rows) == 1 or len(rows) == 11 |
| 3278 | |
| 3279 | r = conn.execute(select(self.table).limit(1)) |
| 3280 | r.fetchone() |
| 3281 | eq_(r.fetchone(), None) |
| 3282 | |
| 3283 | r = conn.execute(select(self.table).limit(5)) |
| 3284 | rows = r.fetchmany(6) |
| 3285 | eq_(rows, [(i, "t_%d" % i) for i in range(1, 6)]) |
| 3286 | |
| 3287 | # result keeps going just fine with blank results... |
| 3288 | eq_(r.fetchmany(2), []) |
| 3289 | |
| 3290 | eq_(r.fetchmany(2), []) |
| 3291 | |
| 3292 | eq_(r.fetchall(), []) |
| 3293 | |
| 3294 | eq_(r.fetchone(), None) |
| 3295 | |
| 3296 | # until we close |
| 3297 | r.close() |
| 3298 | |
| 3299 | self._assert_result_closed(r) |
| 3300 | |
| 3301 | r = conn.execute(select(self.table).limit(5)) |
| 3302 | eq_(r.first(), (1, "t_1")) |
| 3303 | self._assert_result_closed(r) |
| 3304 | |
| 3305 | r = conn.execute(select(self.table).limit(5)) |
| 3306 | eq_(r.scalar(), 1) |
| 3307 | self._assert_result_closed(r) |
| 3308 | |
| 3309 | def _assert_result_closed(self, r): |
| 3310 | assert_raises_message( |
no test coverage detected