@pytest.fixture
-def aconn(loop, dsn):
+def aconn(loop, dsn, pq):
"""Return an `AsyncConnection` connected to the ``--test-dsn`` database."""
from psycopg3 import AsyncConnection
@pytest.fixture
def pgconn(pq, dsn):
"""Return a PGconn connection open to `--test-dsn`."""
- return pq.PGconn.connect(dsn.encode("utf8"))
+ conn = pq.PGconn.connect(dsn.encode("utf8"))
+ if conn.status != 0:
+ pytest.fail(
+ f"bad connection: {conn.error_message.decode('utf8', 'replace')}"
+ )
+ return conn
@pytest.fixture
def test_exec_params_nulls(pq, pgconn):
+ if pgconn.server_version < 100000:
+ pytest.xfail("it doesn't work on pg < 10")
res = pgconn.exec_params(b"select $1, $2, $3", [b"hi", b"", None])
assert res.status == pq.ExecStatus.TUPLES_OK
assert res.get_value(0, 0) == b"hi"
+import pytest
+
+
def test_execute_many(aconn, loop):
cur = aconn.cursor()
rv = loop.run_until_complete(cur.execute("select 'foo'; select 'bar'"))
def test_execute_sequence(aconn, loop):
+ if aconn.pgconn.server_version < 100000:
+ pytest.xfail("it doesn't work on pg < 10")
cur = aconn.cursor()
rv = loop.run_until_complete(
cur.execute("select %s, %s, %s", [1, "foo", None])
+import pytest
+
+
def test_execute_many(conn):
cur = conn.cursor()
rv = cur.execute("select 'foo'; select 'bar'")
def test_execute_sequence(conn):
+ if conn.pgconn.server_version < 100000:
+ pytest.xfail("it doesn't work on pg < 10")
cur = conn.cursor()
rv = cur.execute("select %s, %s, %s", [1, "foo", None])
assert rv is cur