import psycopg
from psycopg import rows
+from .utils import eur
+
def test_tuple_row(conn):
conn.row_factory = rows.dict_row
assert r2.number == 1
assert not cur.nextset()
assert type(r1) is not type(r2)
- cur.execute('select 1 as üåäö, 1 as _, 1 as "123"')
+
+ cur.execute(f'select 1 as üåäö, 2 as _, 3 as "123", 4 as "a-b", 5 as "{eur}eur"')
(r3,) = cur.fetchall()
- assert "üåäö" in r3._fields
- assert "f_" in r3._fields
- assert "f123" in r3._fields
+ assert r3.üåäö == 1
+ assert r3.f_ == 2
+ assert r3.f123 == 3
+ assert r3.a_b == 4
+ assert r3.f_eur == 5
def test_class_row(conn):
assert obj == got
-@pytest.mark.parametrize("name", ["a-b", f"{eur}", "order"])
+@pytest.mark.parametrize("name", ["a-b", f"{eur}", "order", "1", "'"])
def test_literal_invalid_name(conn, name):
conn.execute("set client_encoding to utf8")
- conn.execute(f'create type "{name}" as (foo text)')
- info = CompositeInfo.fetch(conn, f'"{name}"')
+ conn.execute(
+ sql.SQL("create type {name} as (foo text)").format(name=sql.Identifier(name))
+ )
+ info = CompositeInfo.fetch(conn, sql.Identifier(name).as_string(conn))
register_composite(info, conn)
obj = info.python_type("hello")
assert sql.Literal(obj).as_string(conn) == f"'(hello)'::\"{name}\""
got = cur.fetchone()[0]
assert got == obj
assert type(got) is type(obj)
+
+
+@pytest.mark.parametrize(
+ "name, attr",
+ [
+ ("a-b", "a_b"),
+ (f"{eur}", "f_"),
+ ("üåäö", "üåäö"),
+ ("order", "order"),
+ ("1", "f1"),
+ ],
+)
+def test_literal_invalid_attr(conn, name, attr):
+ conn.execute("set client_encoding to utf8")
+ conn.execute(
+ sql.SQL("create type test_attr as ({name} text)").format(
+ name=sql.Identifier(name)
+ )
+ )
+ info = CompositeInfo.fetch(conn, "test_attr")
+ register_composite(info, conn)
+ obj = info.python_type("hello")
+ assert getattr(obj, attr) == "hello"
+ cur = conn.execute(sql.SQL("select {}").format(obj))
+ got = cur.fetchone()[0]
+ assert got == obj
+ assert type(got) is type(obj)