From: Daniele Varrazzo Date: Sat, 5 Dec 2020 04:05:34 +0000 (+0000) Subject: Fixed escaping of a few chars in composite dumping X-Git-Tag: 3.0.dev0~278 X-Git-Url: http://git.ipfire.org/cgi-bin/gitweb.cgi?a=commitdiff_plain;h=75c26d8a31f7039a4b5ccebddd9514c5ac7a1008;p=thirdparty%2Fpsycopg.git Fixed escaping of a few chars in composite dumping --- diff --git a/psycopg3/psycopg3/types/composite.py b/psycopg3/psycopg3/types/composite.py index 49580d4ac..f26739e61 100644 --- a/psycopg3/psycopg3/types/composite.py +++ b/psycopg3/psycopg3/types/composite.py @@ -155,7 +155,9 @@ class TupleDumper(Dumper): dumper = self._tx.get_dumper(item, Format.TEXT) ad = dumper.dump(item) - if self._re_needs_quotes.search(ad): + if not ad: + ad = b'""' + elif self._re_needs_quotes.search(ad): ad = b'"' + self._re_escape.sub(br"\1\1", ad) + b'"' parts.append(ad) @@ -165,13 +167,8 @@ class TupleDumper(Dumper): return b"".join(parts) - _re_needs_quotes = re.compile( - br"""(?xi) - ^$ # the empty string - | [",\\\s] # or a char to escape - """ - ) - _re_escape = re.compile(br"([\"])") + _re_needs_quotes = re.compile(br'[",\\\s()]') + _re_escape = re.compile(br"([\\\"])") class BaseCompositeLoader(Loader): diff --git a/tests/types/test_composite.py b/tests/types/test_composite.py index 1ecb9c0c9..a35f06563 100644 --- a/tests/types/test_composite.py +++ b/tests/types/test_composite.py @@ -173,6 +173,20 @@ async def test_fetch_info_async(aconn, testcomp, name, fields): assert info.fields[i].type_oid == builtins[t].oid +@pytest.mark.parametrize("fmt_in", [Format.TEXT, Format.BINARY]) +def test_dump_composite_all_chars(conn, fmt_in, testcomp): + if fmt_in == Format.BINARY: + pytest.xfail("binary composite dumper not implemented") + ph = "%s" if fmt_in == Format.TEXT else "%b" + cur = conn.cursor() + for i in range(1, 256): + (res,) = cur.execute( + f"select row(chr(%s::int), 1, 1.0)::testcomp = {ph}::testcomp", + (i, (chr(i), 1, 1.0)), + ).fetchone() + assert res is True + + @pytest.mark.parametrize("fmt_out", [Format.TEXT, Format.BINARY]) def test_load_composite(conn, testcomp, fmt_out): cur = conn.cursor(format=fmt_out)