Function test_leak

tests/test_cursor_raw_async.py:76–110

(aconn_cls, dsn, faker, fmt, fmt_out, fetch, row_factory, gc)

@pytest.mark.parametrize("fetch", ["one", "many", "all", "iter"])
@pytest.mark.parametrize("row_factory", ["tuple_row", "dict_row", "namedtuple_row"])
async def test_leak(aconn_cls, dsn, faker, fmt, fmt_out, fetch, row_factory, gc):
    faker.format = fmt
    faker.choose_schema(ncols=5)
    faker.make_records(10)
    row_factory = getattr(rows, row_factory)

    async def work():
        async with await aconn_cls.connect(dsn) as aconn:
            async with aconn.transaction(force_rollback=True):
                async with aconn.cursor(binary=fmt_out, row_factory=row_factory) as cur:
                    await cur.execute(faker.drop_stmt)
                    await cur.execute(faker.create_stmt)
                    async with faker.find_insert_problem_async(aconn):
                        await cur.executemany(faker.insert_stmt, faker.records)
                    await cur.execute(ph(cur, faker.select_stmt))

                    if fetch == "one":
                        while (await cur.fetchone()) is not None:
                            pass
                    elif fetch == "many":
                        while await cur.fetchmany(3):
                            pass
                    elif fetch == "all":
                        await cur.fetchall()
                    elif fetch == "iter":
                        async for rec in cur:
                            pass

    n = []
    gc.collect()
    for i in range(3):
        await work()
        gc.collect()
        n.append(gc.count())
    assert n[0] == n[1] == n[2], f"objects leaked: {n[1] - n[0]}, {n[2] - n[1]}"