Method test_copy_to_table_timeout

tests/test_copy.py:555–582

(self)

            await self.con.execute('DROP TABLE copytab')

    async def test_copy_to_table_timeout(self):
        await self.con.execute('''
            CREATE TABLE copytab(a text, b text);
        ''')

        try:
            class _Source:
                def __init__(self, loop):
                    self.rowcount = 0
                    self.loop = loop

                def __aiter__(self):
                    return self

                async def __anext__(self):
                    self.rowcount += 1
                    await asyncio.sleep(60)
                    return b'a1' * 50 + b'\t' + b'b1' * 50 + b'\n'

            with self.assertRaises(asyncio.TimeoutError):
                await self.con.copy_to_table(
                    'copytab', source=_Source(self.loop), timeout=0.10)

            # Check that the protocol has recovered.
            self.assertEqual(await self.con.fetchval('SELECT 1'), 1)

        finally:
            await self.con.execute('DROP TABLE copytab')

    async def test_copy_to_table_from_file_path(self):
        await self.con.execute('''