Method test_copy_from_table_large_rows

tests/test_copy.py:85–118

(self)

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

    async def test_copy_from_table_large_rows(self):
        await self.con.execute('''
            CREATE TABLE copytab(a text, b text);
            INSERT INTO copytab (a, b) (
                SELECT
                    repeat('a' || i::text, 500000),
                    repeat('b' || i::text, 500000)
                FROM
                    generate_series(1, 5) AS i
            );
        ''')

        try:
            f = io.BytesIO()

            # Basic functionality.
            res = await self.con.copy_from_table('copytab', output=f)

            self.assertEqual(res, 'COPY 5')

            output = f.getvalue().decode().split('\n')
            self.assertEqual(
                output,
                [
                    'a1' * 500000 + '\t' + 'b1' * 500000,
                    'a2' * 500000 + '\t' + 'b2' * 500000,
                    'a3' * 500000 + '\t' + 'b3' * 500000,
                    'a4' * 500000 + '\t' + 'b4' * 500000,
                    'a5' * 500000 + '\t' + 'b5' * 500000,
                    ''
                ]
            )
        finally:
            await self.con.execute('DROP TABLE public.copytab')

    async def test_copy_from_query_basics(self):
        f = io.BytesIO()