| 574 | |
| 575 | |
| 576 | def test_copy_in_format(conn): |
| 577 | file = BytesIO() |
| 578 | conn.execute("set client_encoding to utf8") |
| 579 | cur = conn.cursor() |
| 580 | with Copy(cur, writer=FileWriter(file)) as copy: |
| 581 | for i in range(1, 256): |
| 582 | copy.write_row((i, chr(i))) |
| 583 | |
| 584 | file.seek(0) |
| 585 | rows = file.read().split(b"\n") |
| 586 | assert not rows[-1] |
| 587 | del rows[-1] |
| 588 | |
| 589 | for i, row in enumerate(rows, start=1): |
| 590 | fields = row.split(b"\t") |
| 591 | assert len(fields) == 2 |
| 592 | assert int(fields[0].decode()) == i |
| 593 | if i in special_chars: |
| 594 | assert fields[1].decode() == f"\\{special_chars[i]}" |
| 595 | else: |
| 596 | assert fields[1].decode() == chr(i) |
| 597 | |
| 598 | |
| 599 | @pytest.mark.parametrize( |