import psycopg
from psycopg import pq, sql, postgres
from psycopg import errors as e
-from psycopg.adapt import Transformer, PyFormat as Format, Dumper, Loader
+from psycopg.adapt import Transformer, PyFormat, Dumper, Loader
from psycopg._cmodule import _psycopg
from psycopg.postgres import types as builtins, TEXT_OID
@pytest.mark.parametrize(
"data, format, result, type",
[
- (1, Format.TEXT, b"1", "int2"),
- ("hello", Format.TEXT, b"hello", "text"),
- ("hello", Format.BINARY, b"hello", "text"),
+ (1, PyFormat.TEXT, b"1", "int2"),
+ ("hello", PyFormat.TEXT, b"hello", "text"),
+ ("hello", PyFormat.BINARY, b"hello", "text"),
],
)
def test_dump(data, format, result, type):
t = Transformer()
dumper = t.get_dumper(data, format)
assert dumper.dump(data) == result
- if type == "text" and format != Format.BINARY:
+ if type == "text" and format != PyFormat.BINARY:
assert dumper.oid == 0
else:
assert dumper.oid == builtins[type].oid
)
def test_quote(data, result):
t = Transformer()
- dumper = t.get_dumper(data, Format.TEXT)
+ 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, Format.TEXT) is not dumper
+ assert conn.adapters.get_dumper(MyStr, PyFormat.TEXT) is not dumper
conn.adapters.register_dumper(MyStr, dumper)
- assert conn.adapters.get_dumper(MyStr, Format.TEXT) is dumper
+ assert conn.adapters.get_dumper(MyStr, PyFormat.TEXT) is dumper
def test_register_dumper_by_class_name(conn):
dumper = make_dumper("x")
- assert conn.adapters.get_dumper(MyStr, Format.TEXT) is not dumper
+ assert conn.adapters.get_dumper(MyStr, PyFormat.TEXT) is not dumper
conn.adapters.register_dumper(
f"{MyStr.__module__}.{MyStr.__qualname__}", dumper
)
- assert conn.adapters.get_dumper(MyStr, Format.TEXT) is dumper
+ assert conn.adapters.get_dumper(MyStr, PyFormat.TEXT) is dumper
def test_dump_global_ctx(dsn, global_adapters):
assert cur.fetchone() == ("hellob",)
-@pytest.mark.parametrize("fmt_out", [Format.TEXT, Format.BINARY])
-def test_dump_subclass(conn, fmt_out):
+def test_dump_subclass(conn):
class MyString(str):
pass
"sql, obj",
[("'{hello}'::text[]", ["helloc"]), ("row('hello'::text)", ("helloc",))],
)
-@pytest.mark.parametrize("fmt_out", [pq.Format.TEXT, pq.Format.BINARY])
+@pytest.mark.parametrize("fmt_out", pq.Format)
def test_load_cursor_ctx_nested(conn, sql, obj, fmt_out):
cur = conn.cursor(binary=fmt_out == pq.Format.BINARY)
if fmt_out == pq.Format.TEXT:
assert res == obj
-@pytest.mark.parametrize("fmt_out", [pq.Format.TEXT, pq.Format.BINARY])
+@pytest.mark.parametrize("fmt_out", pq.Format)
def test_array_dumper(conn, fmt_out):
t = Transformer(conn)
- fmt_in = Format.from_pq(fmt_out)
+ fmt_in = PyFormat.from_pq(fmt_out)
dint = t.get_dumper([0], fmt_in)
if fmt_out == pq.Format.BINARY:
assert dint.oid == builtins["int2"].array_oid
assert dint.sub_dumper is None
dstr = t.get_dumper([""], fmt_in)
- if fmt_in == Format.BINARY:
+ if fmt_in == PyFormat.BINARY:
assert dstr.oid == builtins["text"].array_oid
assert dstr.sub_dumper.oid == builtins["text"].oid
else:
assert cur.execute("select %s", ["hello"]).fetchone()[0] == "hellob"
-@pytest.mark.parametrize("fmt_in", [Format.TEXT, Format.BINARY])
+@pytest.mark.parametrize("fmt_in", [PyFormat.TEXT, PyFormat.BINARY])
def test_none_type_argument(conn, fmt_in):
cur = conn.cursor()
cur.execute("create table none_args (id serial primary key, num integer)")
assert cur.fetchone()[0]
-@pytest.mark.parametrize("fmt_in", [Format.AUTO, Format.TEXT, Format.BINARY])
+@pytest.mark.parametrize("fmt_in", PyFormat)
def test_return_untyped(conn, fmt_in):
# Analyze and check for changes using strings in untyped/typed contexts
cur = conn.cursor()
assert cur.fetchone() == ("hello", 10)
cur.execute("create table testjson(data jsonb)")
- if fmt_in != Format.BINARY:
+ if fmt_in != PyFormat.BINARY:
cur.execute(f"insert into testjson (data) values (%{fmt_in})", ["{}"])
assert cur.execute("select data from testjson").fetchone() == ({},)
else:
)
-@pytest.mark.parametrize("fmt_in", [Format.AUTO, Format.TEXT, Format.BINARY])
+@pytest.mark.parametrize("fmt_in", PyFormat.AUTO)
def test_no_cast_needed(conn, fmt_in):
# Verify that there is no need of cast in certain common scenario
cur = conn.execute("select '2021-01-01'::date + %s", [3])
@pytest.mark.slow
-@pytest.mark.parametrize("fmt", [Format.AUTO, Format.TEXT, Format.BINARY])
+@pytest.mark.parametrize("fmt", PyFormat)
@pytest.mark.parametrize("fmt_out", [pq.Format.TEXT, pq.Format.BINARY])
def test_random(conn, faker, fmt, fmt_out):
faker.format = fmt
import psycopg
from psycopg import pq, sql, rows
-from psycopg.adapt import PyFormat as Format
+from psycopg.adapt import PyFormat
from psycopg.postgres import types as builtins
from .utils import gc_collect
cur.executemany(query, [(10, "hello"), (20, "world")])
-@pytest.mark.parametrize("fmt_in", [Format.AUTO, Format.TEXT, Format.BINARY])
+@pytest.mark.parametrize("fmt_in", PyFormat)
def test_executemany_null_first(conn, fmt_in):
cur = conn.cursor()
cur.execute("create table testmany (a bigint, b bigint)")
@pytest.mark.slow
-@pytest.mark.parametrize("fmt", [Format.AUTO, Format.TEXT, Format.BINARY])
-@pytest.mark.parametrize("fmt_out", [pq.Format.TEXT, pq.Format.BINARY])
+@pytest.mark.parametrize("fmt", PyFormat)
+@pytest.mark.parametrize("fmt_out", pq.Format)
@pytest.mark.parametrize("fetch", ["one", "many", "all", "iter"])
@pytest.mark.parametrize(
"row_factory", ["tuple_row", "dict_row", "namedtuple_row"]
import psycopg
from psycopg import pq, sql, rows
-from psycopg.adapt import PyFormat as Format
+from psycopg.adapt import PyFormat
from .utils import gc_collect
from .test_cursor import my_row_factory
await cur.executemany(query, [(10, "hello"), (20, "world")])
-@pytest.mark.parametrize("fmt_in", [Format.AUTO, Format.TEXT, Format.BINARY])
+@pytest.mark.parametrize("fmt_in", PyFormat)
async def test_executemany_null_first(aconn, fmt_in):
cur = aconn.cursor()
await cur.execute("create table testmany (a bigint, b bigint)")
@pytest.mark.slow
-@pytest.mark.parametrize("fmt", [Format.AUTO, Format.TEXT, Format.BINARY])
-@pytest.mark.parametrize("fmt_out", [pq.Format.TEXT, pq.Format.BINARY])
+@pytest.mark.parametrize("fmt", PyFormat)
+@pytest.mark.parametrize("fmt_out", pq.Format)
@pytest.mark.parametrize("fetch", ["one", "many", "all", "iter"])
@pytest.mark.parametrize(
"row_factory", ["tuple_row", "dict_row", "namedtuple_row"]
import psycopg
from psycopg import pq
-from psycopg.adapt import Transformer, PyFormat as Format
+from psycopg.adapt import Transformer, PyFormat
from psycopg._queries import PostgresQuery, _split_query
@pytest.mark.parametrize(
"input, want",
[
- (b"", [(b"", 0, Format.AUTO)]),
- (b"foo bar", [(b"foo bar", 0, Format.AUTO)]),
- (b"foo %% bar", [(b"foo % bar", 0, Format.AUTO)]),
- (b"%s", [(b"", 0, Format.AUTO), (b"", 0, Format.AUTO)]),
- (b"%s foo", [(b"", 0, Format.AUTO), (b" foo", 0, Format.AUTO)]),
- (b"%b foo", [(b"", 0, Format.BINARY), (b" foo", 0, Format.AUTO)]),
- (b"foo %s", [(b"foo ", 0, Format.AUTO), (b"", 0, Format.AUTO)]),
+ (b"", [(b"", 0, PyFormat.AUTO)]),
+ (b"foo bar", [(b"foo bar", 0, PyFormat.AUTO)]),
+ (b"foo %% bar", [(b"foo % bar", 0, PyFormat.AUTO)]),
+ (b"%s", [(b"", 0, PyFormat.AUTO), (b"", 0, PyFormat.AUTO)]),
+ (b"%s foo", [(b"", 0, PyFormat.AUTO), (b" foo", 0, PyFormat.AUTO)]),
+ (b"%b foo", [(b"", 0, PyFormat.BINARY), (b" foo", 0, PyFormat.AUTO)]),
+ (b"foo %s", [(b"foo ", 0, PyFormat.AUTO), (b"", 0, PyFormat.AUTO)]),
(
b"foo %%%s bar",
- [(b"foo %", 0, Format.AUTO), (b" bar", 0, Format.AUTO)],
+ [(b"foo %", 0, PyFormat.AUTO), (b" bar", 0, PyFormat.AUTO)],
),
(
b"foo %(name)s bar",
- [(b"foo ", "name", Format.AUTO), (b" bar", 0, Format.AUTO)],
+ [(b"foo ", "name", PyFormat.AUTO), (b" bar", 0, PyFormat.AUTO)],
),
(
b"foo %(name)s %(name)b bar",
[
- (b"foo ", "name", Format.AUTO),
- (b" ", "name", Format.BINARY),
- (b" bar", 0, Format.AUTO),
+ (b"foo ", "name", PyFormat.AUTO),
+ (b" ", "name", PyFormat.BINARY),
+ (b" bar", 0, PyFormat.AUTO),
],
),
(
b"foo %s%b bar %s baz",
[
- (b"foo ", 0, Format.AUTO),
- (b"", 1, Format.BINARY),
- (b" bar ", 2, Format.AUTO),
- (b" baz", 0, Format.AUTO),
+ (b"foo ", 0, PyFormat.AUTO),
+ (b"", 1, PyFormat.BINARY),
+ (b" bar ", 2, PyFormat.AUTO),
+ (b" baz", 0, PyFormat.AUTO),
],
),
],
import pytest
from psycopg import pq, sql, ProgrammingError
-from psycopg.adapt import PyFormat as Format
+from psycopg.adapt import PyFormat
eur = "\u20ac"
def test_class(self):
assert issubclass(sql.Placeholder, sql.Composable)
- @pytest.mark.parametrize("format", Format)
+ @pytest.mark.parametrize("format", PyFormat)
def test_repr_format(self, conn, format):
ph = sql.Placeholder(format=format)
- add = f"format={format.name}" if format != Format.AUTO else ""
+ add = f"format={format.name}" if format != PyFormat.AUTO else ""
assert str(ph) == repr(ph) == f"Placeholder({add})"
- @pytest.mark.parametrize("format", Format)
+ @pytest.mark.parametrize("format", PyFormat)
def test_repr_name_format(self, conn, format):
ph = sql.Placeholder("foo", format=format)
- add = f", format={format.name}" if format != Format.AUTO else ""
+ add = f", format={format.name}" if format != PyFormat.AUTO else ""
assert str(ph) == repr(ph) == f"Placeholder('foo'{add})"
def test_bad_name(self):
assert sql.Placeholder("foo") != sql.Placeholder()
assert sql.Placeholder("foo") != sql.Literal("foo")
- @pytest.mark.parametrize("format", Format)
+ @pytest.mark.parametrize("format", PyFormat)
def test_as_string(self, conn, format):
ph = sql.Placeholder(format=format)
assert ph.as_string(conn) == f"%{format}"
ph = sql.Placeholder(name="foo", format=format)
assert ph.as_string(conn) == f"%(foo){format}"
- @pytest.mark.parametrize("format", Format)
+ @pytest.mark.parametrize("format", PyFormat)
def test_as_bytes(self, conn, format):
ph = sql.Placeholder(format=format)
assert ph.as_bytes(conn) == f"%{format}".encode("ascii")
import psycopg
from psycopg import pq
from psycopg import sql
-from psycopg.adapt import PyFormat as Format, Transformer, Dumper
+from psycopg.adapt import PyFormat, Transformer, Dumper
from psycopg.types import TypeInfo
from psycopg.postgres import types as builtins
),
]
-fmts_in = [Format.AUTO, Format.TEXT, Format.BINARY]
-
-@pytest.mark.parametrize("fmt_in", fmts_in)
+@pytest.mark.parametrize("fmt_in", PyFormat)
@pytest.mark.parametrize("obj, want", tests_str)
def test_dump_list_str(conn, obj, want, fmt_in):
cur = conn.cursor()
assert cur.fetchone()[0]
-@pytest.mark.parametrize("fmt_out", [pq.Format.TEXT, pq.Format.BINARY])
+@pytest.mark.parametrize("fmt_out", pq.Format)
@pytest.mark.parametrize("want, obj", tests_str)
def test_load_list_str(conn, obj, want, fmt_out):
cur = conn.cursor(binary=fmt_out)
assert cur.fetchone()[0] == want
-@pytest.mark.parametrize("fmt_in", fmts_in)
-@pytest.mark.parametrize("fmt_out", [pq.Format.TEXT, pq.Format.BINARY])
+@pytest.mark.parametrize("fmt_in", PyFormat)
+@pytest.mark.parametrize("fmt_out", pq.Format)
def test_all_chars(conn, fmt_in, fmt_out):
cur = conn.cursor(binary=fmt_out)
for i in range(1, 256):
def test_bad_binary_array(input):
tx = Transformer()
with pytest.raises(psycopg.DataError):
- tx.get_dumper(input, Format.BINARY).dump(input)
+ tx.get_dumper(input, PyFormat.BINARY).dump(input)
-@pytest.mark.parametrize("fmt_out", [pq.Format.TEXT, pq.Format.BINARY])
+@pytest.mark.parametrize("fmt_out", pq.Format)
@pytest.mark.parametrize("want, obj", tests_int)
def test_load_list_int(conn, obj, want, fmt_out):
cur = conn.cursor(binary=fmt_out)
)
def test_array_mixed_numbers(array, type):
tx = Transformer()
- dumper = tx.get_dumper(array, Format.BINARY)
+ dumper = tx.get_dumper(array, PyFormat.BINARY)
dumper.dump(array)
assert dumper.oid == builtins[type].array_oid
@pytest.mark.parametrize(
"wrapper", "Int2 Int4 Int8 Float4 Float8 Decimal".split()
)
-@pytest.mark.parametrize("fmt_in", fmts_in)
-@pytest.mark.parametrize("fmt_out", [pq.Format.TEXT, pq.Format.BINARY])
+@pytest.mark.parametrize("fmt_in", PyFormat)
+@pytest.mark.parametrize("fmt_out", pq.Format)
def test_list_number_wrapper(conn, wrapper, fmt_in, fmt_out):
wrapper = getattr(psycopg.types.numeric, wrapper)
if wrapper is Decimal:
assert cur.description[0].type_code == builtins["numeric"].array_oid
-@pytest.mark.parametrize("fmt_in", fmts_in)
+@pytest.mark.parametrize("fmt_in", PyFormat)
def test_empty_list_mix(conn, fmt_in):
objs = list(range(3))
conn.execute("create table testarrays (col1 bigint[], col2 bigint[])")
assert f2 == []
-@pytest.mark.parametrize("fmt_in", fmts_in)
+@pytest.mark.parametrize("fmt_in", PyFormat)
def test_empty_list(conn, fmt_in):
cur = conn.cursor()
cur.execute("create table test (id serial primary key, data date[])")
assert not cur.fetchone()
-@pytest.mark.parametrize("fmt_in", fmts_in)
+@pytest.mark.parametrize("fmt_in", PyFormat)
def test_empty_list_after_choice(conn, fmt_in):
cur = conn.cursor()
cur.execute("create table test (id serial primary key, data float[])")
from psycopg import pq
from psycopg import sql
-from psycopg.adapt import Transformer, PyFormat as Format
+from psycopg.adapt import Transformer, PyFormat
from psycopg.postgres import types as builtins
-@pytest.mark.parametrize("fmt_in", [Format.AUTO, Format.TEXT, Format.BINARY])
-@pytest.mark.parametrize("fmt_out", [pq.Format.TEXT, pq.Format.BINARY])
+@pytest.mark.parametrize("fmt_in", PyFormat)
+@pytest.mark.parametrize("fmt_out", pq.Format)
@pytest.mark.parametrize("b", [True, False])
def test_roundtrip_bool(conn, b, fmt_in, fmt_out):
cur = conn.cursor(binary=fmt_out)
def test_quote_bool(conn, val):
tx = Transformer()
- assert tx.get_dumper(val, Format.TEXT).quote(val) == str(
+ assert tx.get_dumper(val, PyFormat.TEXT).quote(val) == str(
val
).lower().encode("ascii")
def test_quote_none(conn):
tx = Transformer()
- assert tx.get_dumper(None, Format.TEXT).quote(None) == b"NULL"
+ assert tx.get_dumper(None, PyFormat.TEXT).quote(None) == b"NULL"
cur = conn.cursor()
cur.execute(sql.SQL("select {v}").format(v=sql.Literal(None)))
from psycopg import pq, postgres
from psycopg.sql import Identifier
-from psycopg.adapt import PyFormat as Format
+from psycopg.adapt import PyFormat
from psycopg.postgres import types as builtins
from psycopg.types.range import Range
from psycopg.types.composite import CompositeInfo, register_composite
assert res == obj
-@pytest.mark.parametrize("fmt_out", [pq.Format.TEXT, pq.Format.BINARY])
+@pytest.mark.parametrize("fmt_out", pq.Format)
def test_load_all_chars(conn, fmt_out):
cur = conn.cursor(binary=fmt_out)
for i in range(1, 256):
assert res == (s,)
-@pytest.mark.parametrize("fmt_in", [Format.AUTO, Format.TEXT, Format.BINARY])
+@pytest.mark.parametrize("fmt_in", PyFormat)
def test_dump_builtin_empty_range(conn, fmt_in):
conn.execute(
"""
assert info.field_types[i] == builtins[t].oid
-@pytest.mark.parametrize("fmt_in", [Format.AUTO, Format.TEXT])
+@pytest.mark.parametrize("fmt_in", [PyFormat.AUTO, PyFormat.TEXT])
def test_dump_tuple_all_chars(conn, fmt_in, testcomp):
cur = conn.cursor()
for i in range(1, 256):
assert res is True
-@pytest.mark.parametrize("fmt_in", [Format.AUTO, Format.TEXT, Format.BINARY])
+@pytest.mark.parametrize("fmt_in", PyFormat)
def test_dump_composite_all_chars(conn, fmt_in, testcomp):
cur = conn.cursor()
register_composite(testcomp, cur)
assert res is True
-@pytest.mark.parametrize("fmt_in", [Format.AUTO, Format.TEXT, Format.BINARY])
+@pytest.mark.parametrize("fmt_in", PyFormat)
def test_dump_composite_null(conn, fmt_in, testcomp):
cur = conn.cursor()
register_composite(testcomp, cur)
assert rec[0] is True, rec[1]
-@pytest.mark.parametrize("fmt_out", [pq.Format.TEXT, pq.Format.BINARY])
+@pytest.mark.parametrize("fmt_out", pq.Format)
def test_load_composite(conn, testcomp, fmt_out):
info = CompositeInfo.fetch(conn, "testcomp")
register_composite(info, conn)
assert isinstance(res[0].baz, float)
-@pytest.mark.parametrize("fmt_out", [pq.Format.TEXT, pq.Format.BINARY])
+@pytest.mark.parametrize("fmt_out", pq.Format)
def test_load_composite_factory(conn, testcomp, fmt_out):
info = CompositeInfo.fetch(conn, "testcomp")
def test_register_scope(conn, testcomp):
info = CompositeInfo.fetch(conn, "testcomp")
register_composite(info)
- for fmt in (pq.Format.TEXT, pq.Format.BINARY):
+ for fmt in pq.Format:
for oid in (info.oid, info.array_oid):
assert postgres.adapters._loaders[fmt].pop(oid)
- for fmt in Format:
+ for fmt in PyFormat:
assert postgres.adapters._dumpers[fmt].pop(info.python_type)
cur = conn.cursor()
register_composite(info, cur)
- for fmt in (pq.Format.TEXT, pq.Format.BINARY):
+ for fmt in pq.Format:
for oid in (info.oid, info.array_oid):
assert oid not in postgres.adapters._loaders[fmt]
assert oid not in conn.adapters._loaders[fmt]
assert oid in cur.adapters._loaders[fmt]
register_composite(info, conn)
- for fmt in (pq.Format.TEXT, pq.Format.BINARY):
+ for fmt in pq.Format:
for oid in (info.oid, info.array_oid):
assert oid not in postgres.adapters._loaders[fmt]
assert oid in conn.adapters._loaders[fmt]
import pytest
from psycopg import DataError, pq, sql
-from psycopg.adapt import PyFormat as Format
+from psycopg.adapt import PyFormat
class TestDate:
("max", "9999-12-31"),
],
)
- @pytest.mark.parametrize(
- "fmt_in", [Format.AUTO, Format.TEXT, Format.BINARY]
- )
+ @pytest.mark.parametrize("fmt_in", PyFormat)
def test_dump_date(self, conn, val, expr, fmt_in):
val = as_date(val)
cur = conn.cursor()
("max", "9999-12-31"),
],
)
- @pytest.mark.parametrize("fmt_out", [pq.Format.TEXT, pq.Format.BINARY])
+ @pytest.mark.parametrize("fmt_out", pq.Format)
def test_load_date(self, conn, val, expr, fmt_out):
cur = conn.cursor(binary=fmt_out)
cur.execute(f"select '{expr}'::date")
("max", "9999-12-31 23:59:59.999999"),
],
)
- @pytest.mark.parametrize(
- "fmt_in", [Format.AUTO, Format.TEXT, Format.BINARY]
- )
+ @pytest.mark.parametrize("fmt_in", PyFormat)
def test_dump_datetime(self, conn, val, expr, fmt_in):
cur = conn.cursor()
cur.execute("set timezone to '+02:00'")
("max~2", "9999-12-31 23:59:59.999999"),
],
)
- @pytest.mark.parametrize(
- "fmt_in", [Format.AUTO, Format.TEXT, Format.BINARY]
- )
+ @pytest.mark.parametrize("fmt_in", PyFormat)
def test_dump_datetimetz(self, conn, val, expr, fmt_in):
cur = conn.cursor()
cur.execute("set timezone to '-02:00'")
("NOSUCH0", "2000-1-1", 0),
],
)
- @pytest.mark.parametrize("fmt_out", [pq.Format.TEXT, pq.Format.BINARY])
+ @pytest.mark.parametrize("fmt_out", pq.Format)
def test_load_datetimetz_tz(self, conn, fmt_out, tzname, expr, tzoff):
if "/" in tzname and sys.platform == "win32":
pytest.skip("no IANA db on Windows")
("2000,1,2,3,4,5,6~2", "timestamptz"),
],
)
- @pytest.mark.parametrize(
- "fmt_in", [Format.AUTO, Format.TEXT, Format.BINARY]
- )
+ @pytest.mark.parametrize("fmt_in", PyFormat)
def test_dump_datetime_tz_or_not_tz(self, conn, val, type, fmt_in):
val = as_dt(val)
cur = conn.cursor()
pytest.param("min", "+09:18:59", "Asia/Tokyo", marks=[tz_sec]),
],
)
- @pytest.mark.parametrize("fmt_out", [pq.Format.TEXT, pq.Format.BINARY])
+ @pytest.mark.parametrize("fmt_out", pq.Format)
def test_max_with_timezone(self, conn, fmt_out, valname, tzval, tzname):
# This happens e.g. in Django when it caches forever.
# e.g. see Django test cache.tests.DBCacheTests.test_forever_timeout
("max", "23:59:59.999999"),
],
)
- @pytest.mark.parametrize(
- "fmt_in", [Format.AUTO, Format.TEXT, Format.BINARY]
- )
+ @pytest.mark.parametrize("fmt_in", PyFormat)
def test_dump_time(self, conn, val, expr, fmt_in):
cur = conn.cursor()
cur.execute(
("max", "23:59:59.999999"),
],
)
- @pytest.mark.parametrize("fmt_out", [pq.Format.TEXT, pq.Format.BINARY])
+ @pytest.mark.parametrize("fmt_out", pq.Format)
def test_load_time(self, conn, val, expr, fmt_out):
cur = conn.cursor(binary=fmt_out)
cur.execute(f"select '{expr}'::time")
assert cur.fetchone()[0] == as_time(val)
- @pytest.mark.parametrize("fmt_out", [pq.Format.TEXT, pq.Format.BINARY])
+ @pytest.mark.parametrize("fmt_out", pq.Format)
def test_load_time_24(self, conn, fmt_out):
cur = conn.cursor(binary=fmt_out)
cur.execute("select '24:00'::time")
("max~+12", "23:59:59.999999+12:00"),
],
)
- @pytest.mark.parametrize(
- "fmt_in", [Format.AUTO, Format.TEXT, Format.BINARY]
- )
+ @pytest.mark.parametrize("fmt_in", PyFormat)
def test_dump_timetz(self, conn, val, expr, fmt_in):
cur = conn.cursor()
cur.execute("set timezone to '-02:00'")
("3,0,0,456789~-2", "03:00:00.456789", "+02:00"),
],
)
- @pytest.mark.parametrize("fmt_out", [pq.Format.TEXT, pq.Format.BINARY])
+ @pytest.mark.parametrize("fmt_out", pq.Format)
def test_load_timetz(self, conn, val, timezone, expr, fmt_out):
cur = conn.cursor(binary=fmt_out)
cur.execute(f"set timezone to '{timezone}'")
cur.execute(f"select '{expr}'::timetz")
assert cur.fetchone()[0] == as_time(val)
- @pytest.mark.parametrize("fmt_out", [pq.Format.TEXT, pq.Format.BINARY])
+ @pytest.mark.parametrize("fmt_out", pq.Format)
def test_load_timetz_24(self, conn, fmt_out):
cur = conn.cursor()
cur.execute("select '24:00'::timetz")
("3,4,5,6~2", "timetz"),
],
)
- @pytest.mark.parametrize(
- "fmt_in", [Format.AUTO, Format.TEXT, Format.BINARY]
- )
+ @pytest.mark.parametrize("fmt_in", PyFormat)
def test_dump_time_tz_or_not_tz(self, conn, val, type, fmt_in):
val = as_time(val)
cur = conn.cursor()
("-90d", "-3 month"),
],
)
- @pytest.mark.parametrize("fmt_out", [pq.Format.TEXT, pq.Format.BINARY])
+ @pytest.mark.parametrize("fmt_out", pq.Format)
def test_load_interval(self, conn, val, expr, fmt_out):
cur = conn.cursor(binary=fmt_out)
cur.execute(f"select '{expr}'::interval")
import psycopg.types
from psycopg import pq
from psycopg import sql
-from psycopg.adapt import PyFormat as Format
+from psycopg.adapt import PyFormat
from psycopg.types.json import Json, Jsonb, set_json_dumps, set_json_loads
samples = [
@pytest.mark.parametrize("val", samples)
-@pytest.mark.parametrize("fmt_in", [Format.AUTO, Format.TEXT, Format.BINARY])
+@pytest.mark.parametrize("fmt_in", PyFormat)
def test_json_dump(conn, val, fmt_in):
obj = json.loads(val)
cur = conn.cursor()
assert cur.fetchone()[0] is True
-@pytest.mark.parametrize("fmt_in", [Format.AUTO, Format.TEXT, Format.BINARY])
+@pytest.mark.parametrize("fmt_in", PyFormat)
@pytest.mark.parametrize("val", samples)
def test_jsonb_dump(conn, val, fmt_in):
obj = json.loads(val)
@pytest.mark.parametrize("val", samples)
@pytest.mark.parametrize("jtype", ["json", "jsonb"])
-@pytest.mark.parametrize("fmt_out", [pq.Format.TEXT, pq.Format.BINARY])
+@pytest.mark.parametrize("fmt_out", pq.Format)
def test_json_load(conn, val, jtype, fmt_out):
cur = conn.cursor(binary=fmt_out)
cur.execute(f"select %s::{jtype}", (val,))
@pytest.mark.parametrize("val", samples)
@pytest.mark.parametrize("jtype", ["json", "jsonb"])
-@pytest.mark.parametrize("fmt_out", [pq.Format.TEXT, pq.Format.BINARY])
+@pytest.mark.parametrize("fmt_out", pq.Format)
def test_json_load_copy(conn, val, jtype, fmt_out):
cur = conn.cursor()
stmt = sql.SQL("copy (select {}::{}) to stdout (format {})").format(
assert got == json.loads(val)
-@pytest.mark.parametrize("fmt_in", [Format.AUTO, Format.TEXT, Format.BINARY])
+@pytest.mark.parametrize("fmt_in", PyFormat)
@pytest.mark.parametrize("wrapper", ["Json", "Jsonb"])
def test_json_dump_customise(conn, wrapper, fmt_in):
wrapper = getattr(psycopg.types.json, wrapper)
set_json_dumps(json.dumps)
-@pytest.mark.parametrize("fmt_in", [Format.AUTO, Format.TEXT, Format.BINARY])
+@pytest.mark.parametrize("fmt_in", PyFormat)
@pytest.mark.parametrize("wrapper", ["Json", "Jsonb"])
def test_json_dump_customise_context(conn, wrapper, fmt_in):
wrapper = getattr(psycopg.types.json, wrapper)
assert cur2.fetchone()[0] == "qux"
-@pytest.mark.parametrize("fmt_in", [Format.AUTO, Format.TEXT, Format.BINARY])
+@pytest.mark.parametrize("fmt_in", PyFormat)
@pytest.mark.parametrize("wrapper", ["Json", "Jsonb"])
def test_json_dump_customise_wrapper(conn, wrapper, fmt_in):
wrapper = getattr(psycopg.types.json, wrapper)
from psycopg import pq
from psycopg import sql
-from psycopg.adapt import PyFormat as Format
+from psycopg.adapt import PyFormat
-@pytest.mark.parametrize("fmt_in", [Format.AUTO, Format.TEXT, Format.BINARY])
+@pytest.mark.parametrize("fmt_in", PyFormat)
@pytest.mark.parametrize("val", ["192.168.0.1", "2001:db8::"])
def test_address_dump(conn, fmt_in, val):
cur = conn.cursor()
assert cur.fetchone()[0] is True
-@pytest.mark.parametrize("fmt_in", [Format.AUTO, Format.TEXT, Format.BINARY])
+@pytest.mark.parametrize("fmt_in", PyFormat)
@pytest.mark.parametrize("val", ["127.0.0.1/24", "::ffff:102:300/128"])
def test_interface_dump(conn, fmt_in, val):
cur = conn.cursor()
assert cur.fetchone()[0] is True
-@pytest.mark.parametrize("fmt_in", [Format.AUTO, Format.TEXT, Format.BINARY])
+@pytest.mark.parametrize("fmt_in", PyFormat)
@pytest.mark.parametrize("val", ["127.0.0.0/24", "::ffff:102:300/128"])
def test_network_dump(conn, fmt_in, val):
cur = conn.cursor()
assert cur.fetchone()[0] is True
-@pytest.mark.parametrize("fmt_in", [Format.AUTO, Format.TEXT, Format.BINARY])
+@pytest.mark.parametrize("fmt_in", PyFormat)
def test_network_mixed_size_array(conn, fmt_in):
val = [
ipaddress.IPv4Network("192.168.0.1/32"),
assert val == got
-@pytest.mark.parametrize("fmt_out", [pq.Format.TEXT, pq.Format.BINARY])
+@pytest.mark.parametrize("fmt_out", pq.Format)
@pytest.mark.parametrize("val", ["127.0.0.1/32", "::ffff:102:300/128"])
def test_inet_load_address(conn, fmt_out, val):
addr = ipaddress.ip_address(val.split("/", 1)[0])
assert got == addr
-@pytest.mark.parametrize("fmt_out", [pq.Format.TEXT, pq.Format.BINARY])
+@pytest.mark.parametrize("fmt_out", pq.Format)
@pytest.mark.parametrize("val", ["127.0.0.1/24", "::ffff:102:300/127"])
def test_inet_load_network(conn, fmt_out, val):
pyval = ipaddress.ip_interface(val)
assert got == pyval
-@pytest.mark.parametrize("fmt_out", [pq.Format.TEXT, pq.Format.BINARY])
+@pytest.mark.parametrize("fmt_out", pq.Format)
@pytest.mark.parametrize("val", ["127.0.0.0/24", "::ffff:102:300/128"])
def test_cidr_load(conn, fmt_out, val):
pyval = ipaddress.ip_network(val)
from psycopg import sql
-from psycopg.adapt import Transformer, PyFormat as Format
+from psycopg.adapt import Transformer, PyFormat
def test_quote_none(conn):
tx = Transformer()
- assert tx.get_dumper(None, Format.TEXT).quote(None) == b"NULL"
+ assert tx.get_dumper(None, PyFormat.TEXT).quote(None) == b"NULL"
cur = conn.cursor()
cur.execute(sql.SQL("select {v}").format(v=sql.Literal(None)))
import psycopg
from psycopg import pq
from psycopg import sql
-from psycopg.adapt import Transformer, PyFormat as Format
+from psycopg.adapt import Transformer, PyFormat
from psycopg.types.numeric import FloatLoader
(int(-(2 ** 63)), "'-9223372036854775808'::bigint"),
],
)
-@pytest.mark.parametrize("fmt_in", [Format.TEXT, Format.BINARY])
+@pytest.mark.parametrize("fmt_in", PyFormat)
def test_dump_int(conn, val, expr, fmt_in):
assert isinstance(val, int)
cur = conn.cursor()
(int(-(2 ** 63) - 1), f"'{-2 ** 63 - 1}'::numeric"),
],
)
-@pytest.mark.parametrize("fmt_in", [Format.AUTO, Format.TEXT, Format.BINARY])
+@pytest.mark.parametrize("fmt_in", PyFormat)
def test_dump_int_subtypes(conn, val, expr, fmt_in):
cur = conn.cursor()
cur.execute(f"select pg_typeof({expr}) = pg_typeof(%{fmt_in})", (val,))
assert ok
-@pytest.mark.parametrize("fmt_in", [Format.AUTO, Format.TEXT, Format.BINARY])
+@pytest.mark.parametrize("fmt_in", PyFormat)
def test_dump_enum(conn, fmt_in):
import enum
)
def test_quote_int(conn, val, expr):
tx = Transformer()
- assert tx.get_dumper(val, Format.TEXT).quote(val) == expr
+ assert tx.get_dumper(val, PyFormat.TEXT).quote(val) == expr
cur = conn.cursor()
cur.execute(sql.SQL("select {v}, -{v}").format(v=sql.Literal(val)))
("4294967295", "oid", 4294967295),
],
)
-@pytest.mark.parametrize("fmt_out", [pq.Format.TEXT, pq.Format.BINARY])
+@pytest.mark.parametrize("fmt_out", pq.Format)
def test_load_int(conn, val, pgtype, want, fmt_out):
cur = conn.cursor(binary=fmt_out)
cur.execute(f"select %s::{pgtype}", (val,))
(float("-inf"), "'-Infinity'"),
],
)
-@pytest.mark.parametrize("fmt_in", [Format.TEXT, Format.BINARY])
+@pytest.mark.parametrize("fmt_in", PyFormat)
def test_dump_float(conn, val, expr, fmt_in):
assert isinstance(val, float)
cur = conn.cursor()
)
def test_quote_float(conn, val, expr):
tx = Transformer()
- assert tx.get_dumper(val, Format.TEXT).quote(val) == expr
+ assert tx.get_dumper(val, PyFormat.TEXT).quote(val) == expr
cur = conn.cursor()
cur.execute(sql.SQL("select {v}, -{v}").format(v=sql.Literal(val)))
("-inf", "float8", -float("inf")),
],
)
-@pytest.mark.parametrize("fmt_out", [pq.Format.TEXT, pq.Format.BINARY])
+@pytest.mark.parametrize("fmt_out", pq.Format)
def test_load_float(conn, val, pgtype, want, fmt_out):
cur = conn.cursor(binary=fmt_out)
cur.execute(f"select %s::{pgtype}", (val,))
("-1.42e40", "float8", -1.42e40),
],
)
-@pytest.mark.parametrize("fmt_out", [pq.Format.TEXT, pq.Format.BINARY])
+@pytest.mark.parametrize("fmt_out", pq.Format)
def test_load_float_approx(conn, expr, pgtype, want, fmt_out):
cur = conn.cursor(binary=fmt_out)
cur.execute("select %s::%s" % (expr, pgtype))
"snan",
],
)
-@pytest.mark.parametrize("fmt_in", [Format.TEXT, Format.BINARY])
-@pytest.mark.parametrize("fmt_out", [pq.Format.TEXT, pq.Format.BINARY])
+@pytest.mark.parametrize("fmt_in", PyFormat)
+@pytest.mark.parametrize("fmt_out", pq.Format)
def test_roundtrip_numeric(conn, val, fmt_in, fmt_out):
cur = conn.cursor(binary=fmt_out)
val = Decimal(val)
def test_quote_numeric(conn, val, expr):
val = Decimal(val)
tx = Transformer()
- assert tx.get_dumper(val, Format.TEXT).quote(val) == expr
+ assert tx.get_dumper(val, PyFormat.TEXT).quote(val) == expr
cur = conn.cursor()
cur.execute(sql.SQL("select {v}, -{v}").format(v=sql.Literal(val)))
@pytest.mark.slow
-@pytest.mark.parametrize("fmt_in", [Format.TEXT, Format.BINARY])
+@pytest.mark.parametrize("fmt_in", PyFormat)
def test_dump_numeric_exhaustive(conn, fmt_in):
cur = conn.cursor()
@pytest.mark.slow
-@pytest.mark.parametrize("fmt_out", [pq.Format.TEXT, pq.Format.BINARY])
+@pytest.mark.parametrize("fmt_out", pq.Format)
def test_load_numeric_exhaustive(conn, fmt_out):
cur = conn.cursor(binary=fmt_out)
@pytest.mark.parametrize("wrapper", "Int2 Int4 Int8 Oid Float4 Float8".split())
-@pytest.mark.parametrize("fmt_in", [Format.AUTO, Format.TEXT, Format.BINARY])
+@pytest.mark.parametrize("fmt_in", PyFormat)
def test_repr_wrapper(conn, wrapper, fmt_in):
wrapper = getattr(psycopg.types.numeric, wrapper)
cur = conn.execute(f"select pg_typeof(%{fmt_in})::oid", [wrapper(0)])
import psycopg.errors
from psycopg import pq
from psycopg.sql import Identifier
-from psycopg.adapt import PyFormat as Format
+from psycopg.adapt import PyFormat
from psycopg.types import range as range_module
from psycopg.types.range import Range, RangeInfo, register_range
"pgtype",
"int4range int8range numrange daterange tsrange tstzrange".split(),
)
-@pytest.mark.parametrize("fmt_in", [Format.AUTO, Format.TEXT, Format.BINARY])
+@pytest.mark.parametrize("fmt_in", PyFormat)
def test_dump_builtin_empty(conn, pgtype, fmt_in):
r = Range(empty=True)
cur = conn.execute(f"select 'empty'::{pgtype} = %{fmt_in}", (r,))
Int4Range Int8Range NumericRange DateRange TimestampRange TimestamptzRange
""".split(),
)
-@pytest.mark.parametrize("fmt_in", [Format.AUTO, Format.TEXT, Format.BINARY])
+@pytest.mark.parametrize("fmt_in", PyFormat)
def test_dump_builtin_empty_wrapper(conn, wrapper, fmt_in):
wrapper = getattr(range_module, wrapper)
r = wrapper(empty=True)
@pytest.mark.parametrize(
"fmt_in",
[
- Format.AUTO,
- Format.TEXT,
+ PyFormat.AUTO,
+ PyFormat.TEXT,
# There are many ways to work around this (use text, use a cast on the
# placeholder, use specific Range subclasses).
pytest.param(
- Format.BINARY,
+ PyFormat.BINARY,
marks=pytest.mark.xfail(
reason="can't dump an array of untypes binary range without cast"
),
"pgtype",
"int4range int8range numrange daterange tsrange tstzrange".split(),
)
-@pytest.mark.parametrize("fmt_in", [Format.AUTO, Format.TEXT, Format.BINARY])
+@pytest.mark.parametrize("fmt_in", PyFormat)
def test_dump_builtin_array_with_cast(conn, pgtype, fmt_in):
r1 = Range(empty=True)
r2 = Range(bounds="()")
Int4Range Int8Range NumericRange DateRange TimestampRange TimestamptzRange
""".split(),
)
-@pytest.mark.parametrize("fmt_in", [Format.AUTO, Format.TEXT, Format.BINARY])
+@pytest.mark.parametrize("fmt_in", PyFormat)
def test_dump_builtin_array_wrapper(conn, wrapper, fmt_in):
wrapper = getattr(range_module, wrapper)
r1 = wrapper(empty=True)
@pytest.mark.parametrize("pgtype, min, max, bounds", samples)
-@pytest.mark.parametrize("fmt_in", [Format.AUTO, Format.TEXT, Format.BINARY])
+@pytest.mark.parametrize("fmt_in", PyFormat)
def test_dump_builtin_range(conn, pgtype, min, max, bounds, fmt_in):
r = Range(min, max, bounds)
sub = type2sub[pgtype]
"pgtype",
"int4range int8range numrange daterange tsrange tstzrange".split(),
)
-@pytest.mark.parametrize("fmt_out", [pq.Format.TEXT, pq.Format.BINARY])
+@pytest.mark.parametrize("fmt_out", pq.Format)
def test_load_builtin_empty(conn, pgtype, fmt_out):
r = Range(empty=True)
cur = conn.cursor(binary=fmt_out)
"pgtype",
"int4range int8range numrange daterange tsrange tstzrange".split(),
)
-@pytest.mark.parametrize("fmt_out", [pq.Format.TEXT, pq.Format.BINARY])
+@pytest.mark.parametrize("fmt_out", pq.Format)
def test_load_builtin_inf(conn, pgtype, fmt_out):
r = Range(bounds="()")
cur = conn.cursor(binary=fmt_out)
"pgtype",
"int4range int8range numrange daterange tsrange tstzrange".split(),
)
-@pytest.mark.parametrize("fmt_out", [pq.Format.TEXT, pq.Format.BINARY])
+@pytest.mark.parametrize("fmt_out", pq.Format)
def test_load_builtin_array(conn, pgtype, fmt_out):
r1 = Range(empty=True)
r2 = Range(bounds="()")
@pytest.mark.parametrize("pgtype, min, max, bounds", samples)
-@pytest.mark.parametrize("fmt_out", [pq.Format.TEXT, pq.Format.BINARY])
+@pytest.mark.parametrize("fmt_out", pq.Format)
def test_load_builtin_range(conn, pgtype, min, max, bounds, fmt_out):
r = Range(min, max, bounds)
sub = type2sub[pgtype]
(None, None, "empty"),
],
)
-@pytest.mark.parametrize("format", [pq.Format.TEXT, pq.Format.BINARY])
+@pytest.mark.parametrize("format", pq.Format)
def test_copy_in_empty(conn, min, max, bounds, format):
cur = conn.cursor()
cur.execute("create table copyrange (id serial primary key, r daterange)")
Int4Range Int8Range NumericRange DateRange TimestampRange TimestamptzRange
""".split(),
)
-@pytest.mark.parametrize("format", [pq.Format.TEXT, pq.Format.BINARY])
+@pytest.mark.parametrize("format", pq.Format)
def test_copy_in_empty_wrappers(conn, bounds, wrapper, format):
cur = conn.cursor()
cur.execute("create table copyrange (id serial primary key, r daterange)")
"pgtype",
"int4range int8range numrange daterange tsrange tstzrange".split(),
)
-@pytest.mark.parametrize("format", [pq.Format.TEXT, pq.Format.BINARY])
+@pytest.mark.parametrize("format", pq.Format)
def test_copy_in_empty_set_type(conn, bounds, pgtype, format):
cur = conn.cursor()
cur.execute(f"create table copyrange (id serial primary key, r {pgtype})")
assert cur.fetchone()[0] is True
-@pytest.mark.parametrize("fmt_out", [pq.Format.TEXT, pq.Format.BINARY])
+@pytest.mark.parametrize("fmt_out", pq.Format)
def test_load_custom_empty(conn, testrange, fmt_out):
info = RangeInfo.fetch(conn, "testrange")
register_range(info, conn)
assert got.isempty
-@pytest.mark.parametrize("fmt_out", [pq.Format.TEXT, pq.Format.BINARY])
+@pytest.mark.parametrize("fmt_out", pq.Format)
def test_load_quoting(conn, testrange, fmt_out):
info = RangeInfo.fetch(conn, "testrange")
register_range(info, conn)
assert ord(got.upper) == i + 1
-@pytest.mark.parametrize("fmt_out", [pq.Format.TEXT, pq.Format.BINARY])
+@pytest.mark.parametrize("fmt_out", pq.Format)
def test_mixed_array_types(conn, fmt_out):
conn.execute("create table testmix (a daterange[], b tstzrange[])")
r1 = Range(dt.date(2000, 1, 1), dt.date(2001, 1, 1), "[)")
from psycopg import pq
from psycopg import sql
from psycopg import errors as e
-from psycopg.adapt import PyFormat as Format
+from psycopg.adapt import PyFormat
from psycopg import Binary
eur = "\u20ac"
-
#
# tests with text
#
-@pytest.mark.parametrize("fmt_in", [Format.AUTO, Format.TEXT, Format.BINARY])
+@pytest.mark.parametrize("fmt_in", PyFormat)
def test_dump_1char(conn, fmt_in):
cur = conn.cursor()
for i in range(1, 256):
assert not messages
-@pytest.mark.parametrize("fmt_in", [Format.AUTO, Format.TEXT, Format.BINARY])
+@pytest.mark.parametrize("fmt_in", PyFormat)
def test_dump_zero(conn, fmt_in):
cur = conn.cursor()
s = "foo\x00bar"
@pytest.mark.parametrize(
"typename", ["text", "varchar", "name", "bpchar", '"char"']
)
-@pytest.mark.parametrize("fmt_out", [pq.Format.TEXT, pq.Format.BINARY])
+@pytest.mark.parametrize("fmt_out", pq.Format)
def test_load_1char(conn, typename, fmt_out):
cur = conn.cursor(binary=fmt_out)
for i in range(1, 256):
assert cur.pgresult.fformat(0) == fmt_out
-@pytest.mark.parametrize("fmt_in", [Format.AUTO, Format.TEXT, Format.BINARY])
+@pytest.mark.parametrize("fmt_in", PyFormat)
@pytest.mark.parametrize("encoding", ["utf8", "latin9", "ascii"])
def test_dump_enc(conn, fmt_in, encoding):
cur = conn.cursor()
assert res == ord(eur)
-@pytest.mark.parametrize("fmt_in", [Format.AUTO, Format.TEXT, Format.BINARY])
+@pytest.mark.parametrize("fmt_in", PyFormat)
def test_dump_badenc(conn, fmt_in):
cur = conn.cursor()
cur.execute(f"select %{fmt_in}::bytea", (eur,))
-@pytest.mark.parametrize("fmt_in", [Format.AUTO, Format.TEXT, Format.BINARY])
+@pytest.mark.parametrize("fmt_in", PyFormat)
def test_dump_utf8_badenc(conn, fmt_in):
cur = conn.cursor()
cur.execute(f"select %{fmt_in}", ("\uddf8",))
-@pytest.mark.parametrize("fmt_in", [Format.AUTO, Format.TEXT])
+@pytest.mark.parametrize("fmt_in", [PyFormat.AUTO, PyFormat.TEXT])
def test_dump_enum(conn, fmt_in):
from enum import Enum
assert res == "foo"
-@pytest.mark.parametrize("fmt_in", [Format.AUTO, Format.TEXT])
+@pytest.mark.parametrize("fmt_in", [PyFormat.AUTO, PyFormat.TEXT])
def test_dump_text_oid(conn, fmt_in):
conn.autocommit = True
assert cur.fetchone()[0] == "foobar"
-@pytest.mark.parametrize("fmt_out", [pq.Format.TEXT, pq.Format.BINARY])
+@pytest.mark.parametrize("fmt_out", pq.Format)
@pytest.mark.parametrize("encoding", ["utf8", "latin9"])
@pytest.mark.parametrize("typename", ["text", "varchar", "name", "bpchar"])
def test_load_enc(conn, typename, encoding, fmt_out):
assert res == eur
-@pytest.mark.parametrize("fmt_out", [pq.Format.TEXT, pq.Format.BINARY])
+@pytest.mark.parametrize("fmt_out", pq.Format)
@pytest.mark.parametrize("typename", ["text", "varchar", "name", "bpchar"])
def test_load_badenc(conn, typename, fmt_out):
conn.autocommit = True
copy.read_row()
-@pytest.mark.parametrize("fmt_out", [pq.Format.TEXT, pq.Format.BINARY])
+@pytest.mark.parametrize("fmt_out", pq.Format)
@pytest.mark.parametrize("typename", ["text", "varchar", "name", "bpchar"])
def test_load_ascii(conn, typename, fmt_out):
cur = conn.cursor(binary=fmt_out)
assert res == eur.encode()
-@pytest.mark.parametrize("fmt_in", [Format.AUTO, Format.TEXT, Format.BINARY])
-@pytest.mark.parametrize("fmt_out", [pq.Format.TEXT, pq.Format.BINARY])
+@pytest.mark.parametrize("fmt_in", PyFormat)
+@pytest.mark.parametrize("fmt_out", pq.Format)
@pytest.mark.parametrize("typename", ["text", "varchar", "name", "bpchar"])
def test_text_array(conn, typename, fmt_in, fmt_out):
cur = conn.cursor(binary=fmt_out)
assert res == a
-@pytest.mark.parametrize("fmt_in", [Format.AUTO, Format.TEXT, Format.BINARY])
-@pytest.mark.parametrize("fmt_out", [pq.Format.TEXT, pq.Format.BINARY])
+@pytest.mark.parametrize("fmt_in", PyFormat)
+@pytest.mark.parametrize("fmt_out", pq.Format)
def test_text_array_ascii(conn, fmt_in, fmt_out):
conn.client_encoding = "ascii"
cur = conn.cursor(binary=fmt_out)
#
-@pytest.mark.parametrize("fmt_in", [Format.AUTO, Format.TEXT, Format.BINARY])
+@pytest.mark.parametrize("fmt_in", PyFormat)
@pytest.mark.parametrize("pytype", [bytes, bytearray, memoryview, Binary])
def test_dump_1byte(conn, fmt_in, pytype):
cur = conn.cursor()
assert not messages
-@pytest.mark.parametrize("fmt_out", [pq.Format.TEXT, pq.Format.BINARY])
+@pytest.mark.parametrize("fmt_out", pq.Format)
def test_load_1byte(conn, fmt_out):
cur = conn.cursor(binary=fmt_out)
for i in range(0, 256):
assert cur.pgresult.fformat(0) == fmt_out
-@pytest.mark.parametrize("fmt_in", [Format.AUTO, Format.TEXT, Format.BINARY])
-@pytest.mark.parametrize("fmt_out", [pq.Format.TEXT, pq.Format.BINARY])
+@pytest.mark.parametrize("fmt_in", PyFormat)
+@pytest.mark.parametrize("fmt_out", pq.Format)
def test_bytea_array(conn, fmt_in, fmt_out):
cur = conn.cursor(binary=fmt_out)
a = [bytes(range(0, 256))]
from psycopg import pq
from psycopg import sql
-from psycopg.adapt import PyFormat as Format
+from psycopg.adapt import PyFormat
-@pytest.mark.parametrize("fmt_in", [Format.AUTO, Format.TEXT, Format.BINARY])
+@pytest.mark.parametrize("fmt_in", PyFormat)
def test_uuid_dump(conn, fmt_in):
val = "12345678123456781234567812345679"
cur = conn.cursor()
assert cur.fetchone()[0] is True
-@pytest.mark.parametrize("fmt_out", [pq.Format.TEXT, pq.Format.BINARY])
+@pytest.mark.parametrize("fmt_out", pq.Format)
def test_uuid_load(conn, fmt_out):
cur = conn.cursor(binary=fmt_out)
val = "12345678123456781234567812345679"