| 120 | |
| 121 | @pytest.mark.parametrize("format", pq.Format) |
| 122 | def test_copy_in_records(conn, format): |
| 123 | cur = conn.cursor() |
| 124 | ensure_table(cur, sample_tabledef) |
| 125 | |
| 126 | with cur.copy(f"copy copy_in from stdin {copyopt(format)}") as copy: |
| 127 | row: tuple[Any, ...] |
| 128 | for row in sample_records: |
| 129 | if format == pq.Format.BINARY: |
| 130 | row = tuple((Int4(i) if isinstance(i, int) else i for i in row)) |
| 131 | copy.write_row(row) |
| 132 | |
| 133 | cur.execute("select * from copy_in order by 1") |
| 134 | data = cur.fetchall() |
| 135 | assert data == sample_records |
| 136 | |
| 137 | |
| 138 | @pytest.mark.parametrize("format", pq.Format) |