(conn, format, buffer)
| 601 | [(pq.Format.TEXT, "sample_text"), (pq.Format.BINARY, "sample_binary")], |
| 602 | ) |
| 603 | def test_file_writer(conn, format, buffer): |
| 604 | file = BytesIO() |
| 605 | conn.execute("set client_encoding to utf8") |
| 606 | cur = conn.cursor() |
| 607 | with Copy(cur, binary=format, writer=FileWriter(file)) as copy: |
| 608 | for record in sample_records: |
| 609 | copy.write_row(record) |
| 610 | |
| 611 | file.seek(0) |
| 612 | want = globals()[buffer] |
| 613 | got = file.read() |
| 614 | assert got == want |
| 615 | |
| 616 | |
| 617 | @pytest.mark.slow |