| 548 | |
| 549 | @pytest.mark.parametrize("row_factory", ["tuple_row", "namedtuple_row"]) |
| 550 | async def test_steal_cursor(aconn, row_factory): |
| 551 | cur1 = aconn.cursor() |
| 552 | await cur1.execute("declare test cursor for select generate_series(1, 6) as s") |
| 553 | |
| 554 | cur2 = aconn.cursor("test", row_factory=getattr(rows, row_factory)) |
| 555 | # can call fetch without execute |
| 556 | rec = await cur2.fetchone() |
| 557 | assert rec == (1,) |
| 558 | if row_factory == "namedtuple_row": |
| 559 | assert rec.s == 1 |
| 560 | assert await cur2.fetchmany(3) == [(2,), (3,), (4,)] |
| 561 | assert await cur2.fetchall() == [(5,), (6,)] |
| 562 | await cur2.close() |
| 563 | |
| 564 | |
| 565 | async def test_stolen_cursor_close(aconn): |