Method test_pool_11

tests/test_pool.py:233–288

(self)

        await pool.close()

    async def test_pool_11(self):
        pool = await self.create_pool(database='postgres',
                                      min_size=1, max_size=1)

        async with pool.acquire() as con:
            self.assertIn(repr(con._con), repr(con))  # Test __repr__.

            ps = await con.prepare('SELECT 1')
            txn = con.transaction()
            async with con.transaction():
                cur = await con.cursor('SELECT 1')
                ps_cur = await ps.cursor()

        self.assertIn('[released]', repr(con))

        with self.assertRaisesRegex(
                asyncpg.InterfaceError,
                r'cannot call Connection\.execute.*released back to the pool'):

            con.execute('select 1')

        for meth in ('fetchval', 'fetchrow', 'fetch', 'explain',
                     'get_query', 'get_statusmsg', 'get_parameters',
                     'get_attributes'):
            with self.assertRaisesRegex(
                    asyncpg.InterfaceError,
                    r'cannot call PreparedStatement\.{meth}.*released '
                    r'back to the pool'.format(meth=meth)):

                getattr(ps, meth)()

        for c in (cur, ps_cur):
            for meth in ('fetch', 'fetchrow'):
                with self.assertRaisesRegex(
                        asyncpg.InterfaceError,
                        r'cannot call Cursor\.{meth}.*released '
                        r'back to the pool'.format(meth=meth)):

                    getattr(c, meth)()

            with self.assertRaisesRegex(
                    asyncpg.InterfaceError,
                    r'cannot call Cursor\.forward.*released '
                    r'back to the pool'):

                c.forward(1)

        for meth in ('start', 'commit', 'rollback'):
            with self.assertRaisesRegex(
                    asyncpg.InterfaceError,
                    r'cannot call Transaction\.{meth}.*released '
                    r'back to the pool'.format(meth=meth)):

                getattr(txn, meth)()

        await pool.close()

    async def test_pool_12(self):