Function test_copy_in_buffers_pg_error

tests/test_copy.py:231–238

(conn)

def test_copy_in_buffers_pg_error(conn):
    cur = conn.cursor()
    ensure_table(cur, sample_tabledef)
    with pytest.raises(e.UniqueViolation):
        with cur.copy("copy copy_in from stdin (format text)") as copy:
            copy.write(sample_text)
            copy.write(sample_text)
    assert conn.info.transaction_status == pq.TransactionStatus.INERROR


def test_copy_bad_result(conn):