Function work

tests/test_cursor_raw.py:85–105

()

    row_factory = getattr(rows, row_factory)

    def work():
        with conn_cls.connect(dsn) as conn:
            with conn.transaction(force_rollback=True):
                with conn.cursor(binary=fmt_out, row_factory=row_factory) as cur:
                    cur.execute(faker.drop_stmt)
                    cur.execute(faker.create_stmt)
                    with faker.find_insert_problem(conn):
                        cur.executemany(faker.insert_stmt, faker.records)
                    cur.execute(ph(cur, faker.select_stmt))

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

    n = []
    gc.collect()