conn.tpc_recover()
+@pytest.mark.slow
def test_broken_connection(conn):
cur = conn.cursor()
with pytest.raises(psycopg.DatabaseError):
assert conn.closed
+@pytest.mark.slow
def test_broken(conn):
(session_id,) = conn.execute("show session_id").fetchone()
with pytest.raises(psycopg.OperationalError):
await conn.tpc_recover()
+@pytest.mark.slow
async def test_broken_connection(aconn):
cur = aconn.cursor()
with pytest.raises(psycopg.DatabaseError):
assert aconn.closed
+@pytest.mark.slow
async def test_broken(aconn):
cur = await aconn.execute("show session_id")
(session_id,) = await cur.fetchone()