From: Daniele Varrazzo Date: Mon, 19 Sep 2022 00:54:44 +0000 (+0100) Subject: test: verify that dumpers returning None are used correctly in queries X-Git-Tag: 3.2.0~20^2~5 X-Git-Url: http://git.ipfire.org/?a=commitdiff_plain;h=0bdeeaba7fbb46a6afd37a3ebd0cc402babd9106;p=thirdparty%2Fpsycopg.git test: verify that dumpers returning None are used correctly in queries --- diff --git a/tests/test_adapt.py b/tests/test_adapt.py index 31668d20e..688be1a12 100644 --- a/tests/test_adapt.py +++ b/tests/test_adapt.py @@ -1,16 +1,18 @@ import datetime as dt from types import ModuleType -from typing import Any, List +from typing import Any, List, Optional import pytest import psycopg from psycopg import pq, sql, postgres from psycopg import errors as e +from psycopg.abc import Buffer from psycopg.adapt import Transformer, PyFormat, Dumper, Loader from psycopg._cmodule import _psycopg from psycopg.postgres import types as builtins from psycopg.types.array import ListDumper, ListBinaryDumper +from psycopg.types.string import StrDumper, StrBinaryDumper @pytest.mark.parametrize( @@ -47,6 +49,20 @@ def test_quote(data, result): assert dumper.quote(data) == result +@pytest.mark.parametrize( + "data, result", + [ + ("hello", b"'hello'"), + ("", b"NULL"), + ], +) +def test_quote_none(data, result, global_adapters): + psycopg.adapters.register_dumper(str, StrNoneDumper) + t = Transformer() + dumper = t.get_dumper(data, PyFormat.TEXT) + assert dumper.quote(data) == result + + def test_register_dumper_by_class(conn): dumper = make_dumper("x") assert conn.adapters.get_dumper(MyStr, PyFormat.TEXT) is not dumper @@ -121,9 +137,6 @@ def test_dump_subclass(conn): def test_subclass_dumper(conn): - # This might be a C fast object: make sure that the Python code is called - from psycopg.types.string import StrDumper - class MyStrDumper(StrDumper): def dump(self, obj): return (obj * 2).encode() @@ -348,14 +361,28 @@ def test_last_dumper_registered_ctx(conn): assert cur.execute("select %s", ["hello"]).fetchone()[0] == "hellob" -@pytest.mark.parametrize("fmt_in", [PyFormat.TEXT, PyFormat.BINARY]) +@pytest.mark.parametrize("fmt_in", PyFormat) def test_none_type_argument(conn, fmt_in): cur = conn.cursor() cur.execute("create table none_args (id serial primary key, num integer)") - cur.execute("insert into none_args (num) values (%s) returning id", (None,)) + cur.execute( + f"insert into none_args (num) values (%{fmt_in.value}) returning id", (None,) + ) assert cur.fetchone()[0] +@pytest.mark.parametrize("fmt_in", [PyFormat.TEXT, PyFormat.BINARY]) +def test_dump_to_none(conn, fmt_in): + cur = conn.cursor() + dumper = StrNoneDumper if fmt_in == PyFormat.TEXT else StrNoneBinaryDumper + cur.adapters.register_dumper(str, dumper) + cur.execute("create table none_args (id serial primary key, data text)") + for s in ["foo", ""]: + cur.execute("insert into none_args (data) values (%s)", (s,)) + cur.execute("select data from none_args order by id") + assert cur.fetchall() == [("foo",), (None,)] + + @pytest.mark.crdb("skip", reason="test in crdb test suite") @pytest.mark.parametrize("fmt_in", PyFormat) def test_return_untyped(conn, fmt_in): @@ -493,6 +520,16 @@ class MyStr(str): pass +class StrNoneDumper(StrDumper): + def dump(self, obj: str) -> Optional[Buffer]: + return super().dump(obj) if obj else None + + +class StrNoneBinaryDumper(StrBinaryDumper): + def dump(self, obj: str) -> Optional[Buffer]: + return super().dump(obj) if obj else None + + def make_dumper(suffix): """Create a test dumper appending a suffix to the bytes representation.""" diff --git a/tests/types/test_array.py b/tests/types/test_array.py index 4da6a2149..226fca46b 100644 --- a/tests/types/test_array.py +++ b/tests/types/test_array.py @@ -13,6 +13,7 @@ from psycopg.types import TypeInfo from psycopg.postgres import types as builtins from psycopg.types.array import register_array +from ..test_adapt import StrNoneDumper, StrNoneBinaryDumper tests_str = [ ([[[[[["a"]]]]]], "{{{{{{a}}}}}}"), @@ -49,6 +50,16 @@ def test_dump_list_str(conn, obj, want, fmt_in): assert cur.fetchone()[0] +@pytest.mark.parametrize("fmt_in", PyFormat) +def test_dump_list_str_none(conn, fmt_in): + cur = conn.cursor() + cur.adapters.register_dumper(str, StrNoneDumper) + cur.adapters.register_dumper(str, StrNoneBinaryDumper) + + cur.execute(f"select %{fmt_in.value}::text[]", (["foo", "", "bar"],)) + assert cur.fetchone()[0] == ["foo", None, "bar"] + + @pytest.mark.parametrize("fmt_out", pq.Format) def test_load_empty_list_str(conn, fmt_out): cur = conn.cursor(binary=fmt_out) diff --git a/tests/types/test_numeric.py b/tests/types/test_numeric.py index 8e5db7668..9a71aaebb 100644 --- a/tests/types/test_numeric.py +++ b/tests/types/test_numeric.py @@ -1,14 +1,16 @@ import enum -from decimal import Decimal from math import isnan, isinf, exp +from typing import Optional +from decimal import Decimal import pytest import psycopg from psycopg import pq from psycopg import sql +from psycopg.abc import Buffer from psycopg.adapt import Transformer, PyFormat -from psycopg.types.numeric import FloatLoader +from psycopg.types.numeric import Int8, Int8Dumper, Int8BinaryDumper, FloatLoader from ..fix_crdb import is_crdb @@ -73,6 +75,22 @@ def test_dump_int_subtypes(conn, val, expr, fmt_in): assert ok +@pytest.mark.parametrize("fmt_in", [PyFormat.TEXT, PyFormat.BINARY]) +def test_int_none(conn, fmt_in): + Base: type = Int8Dumper if fmt_in == PyFormat.TEXT else Int8BinaryDumper + + class MyDumper(Base): # type: ignore + def dump(self, obj: int) -> Optional[Buffer]: + if not obj: + return None + else: + return super().dump(obj) # type: ignore + + conn.adapters.register_dumper(Int8, MyDumper) + cur = conn.execute("select %s, %s", [Int8(0), Int8(1)]) + assert cur.fetchone() == (None, 1) + + class MyEnum(enum.IntEnum): foo = 42