Method test_tpc_recover_non_dbapi_connection

tests/test_tpc_async.py:291–303

(self, aconn_cls, aconn, dsn, tpc)

            await aconn.cancel_safe()

    async def test_tpc_recover_non_dbapi_connection(self, aconn_cls, aconn, dsn, tpc):
        aconn.row_factory = psycopg.rows.dict_row
        await aconn.tpc_begin("dict-connection")
        await aconn.tpc_prepare()
        await aconn.close()

        async with await aconn_cls.connect(dsn) as aconn:
            xids = await aconn.tpc_recover()
            xid = [x for x in xids if x.database == aconn.info.dbname][0]

        assert xid.format_id is None
        assert xid.gtrid == "dict-connection"
        assert xid.bqual is None