| 32 | |
| 33 | @pytest.mark.parametrize("format", pq.Format) |
| 34 | def test_copy_out_read(conn, format): |
| 35 | if format == pq.Format.TEXT: |
| 36 | want = [row + b"\n" for row in sample_text.splitlines()] |
| 37 | else: |
| 38 | want = sample_binary_rows |
| 39 | |
| 40 | cur = conn.cursor() |
| 41 | with cur.copy(f"copy ({sample_values}) to stdout (format {format.name})") as copy: |
| 42 | for row in want: |
| 43 | got = copy.read() |
| 44 | assert got == row |
| 45 | assert conn.info.transaction_status == pq.TransactionStatus.ACTIVE |
| 46 | |
| 47 | assert copy.read() == b"" |
| 48 | assert copy.read() == b"" |
| 49 | |
| 50 | assert copy.read() == b"" |
| 51 | assert conn.info.transaction_status == pq.TransactionStatus.INTRANS |
| 52 | |
| 53 | |
| 54 | @pytest.mark.parametrize("format", pq.Format) |