Method test_copy_to_table_large_rows

tests/test_copy.py:462–487

(self)

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

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

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

                def __aiter__(self):
                    return self

                async def __anext__(self):
                    if self.rowcount >= 100:
                        raise StopAsyncIteration
                    else:
                        self.rowcount += 1
                        return b'a1' * 500000 + b'\t' + b'b1' * 500000 + b'\n'

            res = await self.con.copy_to_table('copytab', source=_Source())

            self.assertEqual(res, 'COPY 100')

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

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