Method test_copy_from_query_timeout_2

tests/test_copy.py:340–361

(self)

        self.assertEqual(await self.con.fetchval('SELECT 1'), 1)

    async def test_copy_from_query_timeout_2(self):
        async def writer(data):
            try:
                await asyncio.sleep(10)
            except asyncio.TimeoutError:
                raise
            else:
                self.fail('TimeoutError not raised')

        coro = self.con.copy_from_query('''
            SELECT
                repeat('a', 500)
            FROM
                generate_series(1, 5000) AS i
        ''', output=writer, timeout=0.10)

        task = self.loop.create_task(coro)

        with self.assertRaises(asyncio.TimeoutError):
            await task

        self.assertEqual(await self.con.fetchval('SELECT 1'), 1)


class TestCopyTo(tb.ConnectedTestCase):