NonAsciiEnum = Enum(
- "NonAsciiEnum", {"X\xe0": "x\xe0", "X\xe1": "x\xe1", "COMMA": "foo,bar"}, type=str
+ "NonAsciiEnum",
+ {"X\xe0": "x\xe0", "X\xe1": "x\xe1", "COMMA": "foo,bar"},
+ type=str,
)
THREE = 3
-enum_cases = [PureTestEnum, StrTestEnum, NonAsciiEnum, IntTestEnum]
-ascii_cases = [PureTestEnum, StrTestEnum, IntTestEnum]
-
+enum_cases = [PureTestEnum, StrTestEnum, IntTestEnum]
encodings = ["utf8", "latin1"]
-@pytest.fixture(scope="session", params=enum_cases)
-def testenum(request, svcconn):
- enum = request.param
- return ensure_enum(enum, svcconn)
-
-
@pytest.fixture(scope="session", autouse=True)
def make_test_enums(request, svcconn):
- for enum in enum_cases:
+ for enum in enum_cases + [NonAsciiEnum]:
ensure_enum(enum, svcconn)
def ensure_enum(enum, conn):
name = enum.__name__.lower()
- labels = list(enum.__members__.keys())
+ labels = list(enum.__members__)
conn.execute(
sql.SQL(
"""
return name, enum, labels
-def test_fetch_info(conn, testenum):
- name, enum, labels = testenum
+def test_fetch_info(conn):
+ info = EnumInfo.fetch(conn, "StrTestEnum")
+ assert info.name == "strtestenum"
+ assert info.oid > 0
+ assert info.oid != info.array_oid > 0
+ assert len(info.labels) == len(StrTestEnum)
+ assert info.labels == list(StrTestEnum.__members__)
+
- info = EnumInfo.fetch(conn, name)
- assert info.name == name
+@pytest.mark.asyncio
+async def test_fetch_info_async(aconn):
+ info = await EnumInfo.fetch(aconn, "PureTestEnum")
+ assert info.name == "puretestenum"
assert info.oid > 0
assert info.oid != info.array_oid > 0
- assert len(info.labels) == len(labels)
- assert info.labels == labels
+ assert len(info.labels) == len(PureTestEnum)
+ assert info.labels == list(PureTestEnum.__members__)
-def test_register_makes_a_type(conn, testenum):
- name, enum, labels = testenum
- info = EnumInfo.fetch(conn, name)
+def test_register_makes_a_type(conn):
+ info = EnumInfo.fetch(conn, "IntTestEnum")
assert info
assert info.enum is None
register_enum(info, context=conn)
assert info.enum is not None
- assert [e.name for e in info.enum] == [e.name for e in enum]
+ assert [e.name for e in info.enum] == list(IntTestEnum.__members__)
+
+
+@pytest.mark.parametrize("enum", enum_cases)
+@pytest.mark.parametrize("fmt_in", PyFormat)
+@pytest.mark.parametrize("fmt_out", pq.Format)
+def test_enum_loader(conn, enum, fmt_in, fmt_out):
+ info = EnumInfo.fetch(conn, enum.__name__)
+ register_enum(info, conn, enum=enum)
+
+ for label in info.labels:
+ cur = conn.execute(
+ f"select %{fmt_in}::{enum.__name__}", [label], binary=fmt_out
+ )
+ assert cur.fetchone()[0] == enum[label]
@pytest.mark.parametrize("encoding", encodings)
@pytest.mark.parametrize("fmt_in", PyFormat)
@pytest.mark.parametrize("fmt_out", pq.Format)
-def test_enum_loader(conn, testenum, encoding, fmt_in, fmt_out):
+def test_enum_loader_nonascii(conn, encoding, fmt_in, fmt_out):
+ enum = NonAsciiEnum
conn.execute(f"set client_encoding to {encoding}")
- name, enum, labels = testenum
- register_enum(EnumInfo.fetch(conn, name), conn, enum=enum)
+ info = EnumInfo.fetch(conn, enum.__name__)
+ register_enum(info, conn, enum=enum)
- for label in labels:
- cur = conn.execute(f"select %{fmt_in}::{name}", [label], binary=fmt_out)
+ for label in info.labels:
+ cur = conn.execute(f"select %{fmt_in}::{info.name}", [label], binary=fmt_out)
assert cur.fetchone()[0] == enum[label]
+@pytest.mark.parametrize("enum", enum_cases)
@pytest.mark.parametrize("fmt_in", PyFormat)
@pytest.mark.parametrize("fmt_out", pq.Format)
-@pytest.mark.parametrize("enum", ascii_cases)
def test_enum_loader_sqlascii(conn, enum, fmt_in, fmt_out):
- info = EnumInfo.fetch(conn, enum.__name__.lower())
+ info = EnumInfo.fetch(conn, enum.__name__)
register_enum(info, conn, enum)
conn.execute("set client_encoding to sql_ascii")
assert cur.fetchone()[0] == enum[label]
+@pytest.mark.parametrize("enum", enum_cases)
+@pytest.mark.parametrize("fmt_in", PyFormat)
+@pytest.mark.parametrize("fmt_out", pq.Format)
+def test_enum_dumper(conn, enum, fmt_in, fmt_out):
+ info = EnumInfo.fetch(conn, enum.__name__)
+ register_enum(info, conn, enum)
+
+ for item in enum:
+ cur = conn.execute(f"select %{fmt_in}", [item], binary=fmt_out)
+ assert cur.fetchone()[0] == item
+
+
@pytest.mark.parametrize("encoding", encodings)
@pytest.mark.parametrize("fmt_in", PyFormat)
@pytest.mark.parametrize("fmt_out", pq.Format)
-def test_enum_dumper(conn, testenum, encoding, fmt_in, fmt_out):
+def test_enum_dumper_nonascii(conn, encoding, fmt_in, fmt_out):
+ enum = NonAsciiEnum
conn.execute(f"set client_encoding to {encoding}")
- name, enum, labels = testenum
- register_enum(EnumInfo.fetch(conn, name), conn, enum)
+ info = EnumInfo.fetch(conn, enum.__name__)
+ register_enum(info, conn, enum)
for item in enum:
cur = conn.execute(f"select %{fmt_in}", [item], binary=fmt_out)
assert cur.fetchone()[0] == item
+@pytest.mark.parametrize("enum", enum_cases)
@pytest.mark.parametrize("fmt_in", PyFormat)
@pytest.mark.parametrize("fmt_out", pq.Format)
-@pytest.mark.parametrize("enum", ascii_cases)
def test_enum_dumper_sqlascii(conn, enum, fmt_in, fmt_out):
- info = EnumInfo.fetch(conn, enum.__name__.lower())
+ info = EnumInfo.fetch(conn, enum.__name__)
register_enum(info, conn, enum)
conn.execute("set client_encoding to sql_ascii")
assert cur.fetchone()[0] == item
-@pytest.mark.parametrize("encoding", encodings)
+@pytest.mark.parametrize("enum", enum_cases)
@pytest.mark.parametrize("fmt_in", PyFormat)
@pytest.mark.parametrize("fmt_out", pq.Format)
-def test_generic_enum_dumper(conn, testenum, encoding, fmt_in, fmt_out):
- conn.execute(f"set client_encoding to {encoding}")
-
- name, enum, labels = testenum
-
+def test_generic_enum_dumper(conn, enum, fmt_in, fmt_out):
for item in enum:
if enum is PureTestEnum:
want = item.name
@pytest.mark.parametrize("encoding", encodings)
@pytest.mark.parametrize("fmt_in", PyFormat)
@pytest.mark.parametrize("fmt_out", pq.Format)
-def test_generic_enum_loader(conn, testenum, encoding, fmt_in, fmt_out):
+def test_generic_enum_dumper_nonascii(conn, encoding, fmt_in, fmt_out):
conn.execute(f"set client_encoding to {encoding}")
+ for item in NonAsciiEnum:
+ cur = conn.execute(f"select %{fmt_in}", [item.value], binary=fmt_out)
+ assert cur.fetchone()[0] == item.value
+
- name, enum, labels = testenum
- for label in labels:
- cur = conn.execute(f"select %{fmt_in}::{name}", [label], binary=fmt_out)
+@pytest.mark.parametrize("enum", enum_cases)
+@pytest.mark.parametrize("fmt_in", PyFormat)
+@pytest.mark.parametrize("fmt_out", pq.Format)
+def test_generic_enum_loader(conn, enum, fmt_in, fmt_out):
+ for label in enum.__members__:
+ cur = conn.execute(
+ f"select %{fmt_in}::{enum.__name__}", [label], binary=fmt_out
+ )
want = enum[label].name
if fmt_out == pq.Format.BINARY:
- want = want.encode(encoding)
+ want = want.encode()
assert cur.fetchone()[0] == want
@pytest.mark.parametrize("encoding", encodings)
@pytest.mark.parametrize("fmt_in", PyFormat)
@pytest.mark.parametrize("fmt_out", pq.Format)
-def test_enum_array_loader(conn, testenum, encoding, fmt_in, fmt_out):
+def test_generic_enum_loader_nonascii(conn, encoding, fmt_in, fmt_out):
conn.execute(f"set client_encoding to {encoding}")
- name, enum, labels = testenum
- register_enum(EnumInfo.fetch(conn, name), conn, enum)
-
- cur = conn.execute(f"select %{fmt_in}::{name}[]", [labels], binary=fmt_out)
- assert cur.fetchone()[0] == list(enum)
+ for label in NonAsciiEnum.__members__:
+ cur = conn.execute(f"select %{fmt_in}::nonasciienum", [label], binary=fmt_out)
+ if fmt_out == pq.Format.TEXT:
+ assert cur.fetchone()[0] == label
+ else:
+ assert cur.fetchone()[0] == label.encode(encoding)
-@pytest.mark.parametrize("encoding", encodings)
@pytest.mark.parametrize("fmt_in", PyFormat)
@pytest.mark.parametrize("fmt_out", pq.Format)
-def test_enum_array_dumper(conn, testenum, encoding, fmt_in, fmt_out):
- conn.execute(f"set client_encoding to {encoding}")
-
- name, enum, labels = testenum
- register_enum(EnumInfo.fetch(conn, name), conn, enum)
+def test_enum_array_loader(conn, fmt_in, fmt_out):
+ enum = PureTestEnum
+ info = EnumInfo.fetch(conn, enum.__name__)
+ register_enum(info, conn, enum)
- cur = conn.execute(f"select %{fmt_in}", [list(enum)], binary=fmt_out)
+ labels = list(enum.__members__)
+ cur = conn.execute(f"select %{fmt_in}::{info.name}[]", [labels], binary=fmt_out)
assert cur.fetchone()[0] == list(enum)
-@pytest.mark.parametrize("encoding", encodings)
@pytest.mark.parametrize("fmt_in", PyFormat)
@pytest.mark.parametrize("fmt_out", pq.Format)
-def test_generic_enum_array_loader(conn, testenum, encoding, fmt_in, fmt_out):
- conn.execute(f"set client_encoding to {encoding}")
-
- name, enum, labels = testenum
- info = TypeInfo.fetch(conn, name)
- info.register(conn)
- want = [member.name for member in enum]
- if fmt_out == pq.Format.BINARY:
- want = [item.encode(encoding) for item in want]
- cur = conn.execute(f"select %{fmt_in}::{name}[]", [labels], binary=fmt_out)
- assert cur.fetchone()[0] == want
+def test_enum_array_dumper(conn, fmt_in, fmt_out):
+ enum = StrTestEnum
+ info = EnumInfo.fetch(conn, enum.__name__)
+ register_enum(info, conn, enum)
+ cur = conn.execute(f"select %{fmt_in}::text[]", [list(enum)], binary=fmt_out)
+ assert cur.fetchone()[0] == list(enum.__members__)
-@pytest.mark.asyncio
-async def test_fetch_info_async(aconn, testenum):
- name, enum, labels = testenum
- info = await EnumInfo.fetch(aconn, name)
- assert info.name == name
- assert info.oid > 0
- assert info.oid != info.array_oid > 0
- assert len(info.labels) == len(labels)
- assert info.labels == labels
-
-
-@pytest.mark.asyncio
-@pytest.mark.parametrize("encoding", encodings)
@pytest.mark.parametrize("fmt_in", PyFormat)
@pytest.mark.parametrize("fmt_out", pq.Format)
-async def test_enum_async(aconn, testenum, encoding, fmt_in, fmt_out):
- await aconn.execute(f"set client_encoding to {encoding}")
-
- name, enum, labels = testenum
- register_enum(await EnumInfo.fetch(aconn, name), aconn, enum)
-
- for label in labels:
- cur = await aconn.execute(f"select %{fmt_in}::{name}", [label], binary=fmt_out)
- assert (await cur.fetchone())[0] == enum[label]
-
- cur = await cur.execute(f"select %{fmt_in}", [list(enum)])
- assert (await cur.fetchone())[0] == list(enum)
+def test_generic_enum_array_loader(conn, fmt_in, fmt_out):
+ enum = IntTestEnum
+ info = TypeInfo.fetch(conn, enum.__name__)
+ info.register(conn)
+ labels = list(enum.__members__)
+ cur = conn.execute(f"select %{fmt_in}::{info.name}[]", [labels], binary=fmt_out)
+ if fmt_out == pq.Format.TEXT:
+ assert cur.fetchone()[0] == labels
+ else:
+ assert cur.fetchone()[0] == [item.encode() for item in labels]
-@pytest.mark.parametrize("fmt_in", PyFormat)
-@pytest.mark.parametrize("fmt_out", pq.Format)
-def test_enum_error(conn, fmt_in, fmt_out):
+def test_enum_error(conn):
conn.autocommit = True
info = EnumInfo.fetch(conn, "puretestenum")