Function test_copy_in_text_pinned

tests/test_copy.py:526–555

(conn)

def test_copy_in_text_pinned(conn):
    cur = conn.cursor()
    cur.adapters.register_dumper(int, StrictIntDumper)

    cols = [
        "col1 serial primary key",
        "col2 int",
        "col3 int",
        "col4 double precision",
        "col5 double precision",
    ]
    ensure_table(cur, ",".join(cols))

    with cur.copy(
        "copy copy_in (col2,col3,col4,col5) from stdin (format text)"
    ) as copy:
        # pinned dumpers from set_types: type check & cast done on psycopg side
        # much faster, allows catching errors early without postgres involvement
        copy.set_types(["int4", "int4", "double precision", "double precision"])
        copy.write_row([1, 2, 3, 4.1])
        with pytest.raises(
            (e.DataError, TypeError)
        ):  # FIXME: should errors from dumpers be harmonized?
            copy.write_row([1.0, 2, 3, 4.1])
        with pytest.raises((e.DataError, TypeError)):
            copy.write_row([1, "2", 3, 4.1])

    cur.execute("select col2,col3,col4,col5 from copy_in order by 1")
    data = cur.fetchall()
    assert data == [(1, 2, 3, 4.1)]


def test_copy_in_allchars(conn):