@pytest.mark.parametrize("name", ["text", sql.Identifier("text")])
-@pytest.mark.parametrize("status", ["IDLE", "INTRANS"])
+@pytest.mark.parametrize("status", ["IDLE", "INTRANS", None])
@pytest.mark.parametrize(
"encoding", ["utf8", crdb_encoding("latin1"), crdb_encoding("sql_ascii")]
)
with conn.transaction():
conn.execute("select set_config('client_encoding', %s, false)", [encoding])
- status = getattr(TransactionStatus, status)
- if status == TransactionStatus.INTRANS:
- conn.execute("select 1")
+ if status:
+ status = getattr(TransactionStatus, status)
+ if status == TransactionStatus.INTRANS:
+ conn.execute("select 1")
+ else:
+ conn.autocommit = True
+ status = TransactionStatus.IDLE
assert conn.info.transaction_status == status
info = TypeInfo.fetch(conn, name)
@pytest.mark.asyncio
@pytest.mark.parametrize("name", ["text", sql.Identifier("text")])
-@pytest.mark.parametrize("status", ["IDLE", "INTRANS"])
+@pytest.mark.parametrize("status", ["IDLE", "INTRANS", None])
@pytest.mark.parametrize(
"encoding", ["utf8", crdb_encoding("latin1"), crdb_encoding("sql_ascii")]
)
"select set_config('client_encoding', %s, false)", [encoding]
)
- status = getattr(TransactionStatus, status)
- if status == TransactionStatus.INTRANS:
- await aconn.execute("select 1")
+ if status:
+ status = getattr(TransactionStatus, status)
+ if status == TransactionStatus.INTRANS:
+ await aconn.execute("select 1")
+ else:
+ await aconn.set_autocommit(True)
+ status = TransactionStatus.IDLE
assert aconn.info.transaction_status == status
info = await TypeInfo.fetch(aconn, name)