Method test_copy_from_table_basics

tests/test_copy.py:21–83

(self)

class TestCopyFrom(tb.ConnectedTestCase):

    async def test_copy_from_table_basics(self):
        await self.con.execute('''
            CREATE TABLE copytab(a text, "b~" text, i int);
            INSERT INTO copytab (a, "b~", i) (
                SELECT 'a' || i::text, 'b' || i::text, i
                FROM generate_series(1, 5) AS i
            );
            INSERT INTO copytab (a, "b~", i) VALUES('*', NULL, NULL);
        ''')

        try:
            f = io.BytesIO()

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

            self.assertEqual(res, 'COPY 6')

            output = f.getvalue().decode().split('\n')
            self.assertEqual(
                output,
                [
                    'a1\tb1\t1',
                    'a2\tb2\t2',
                    'a3\tb3\t3',
                    'a4\tb4\t4',
                    'a5\tb5\t5',
                    '*\t\\N\t\\N',
                    ''
                ]
            )

            # Test parameters.
            await self.con.execute('SET search_path=none')

            f.seek(0)
            f.truncate()

            res = await self.con.copy_from_table(
                'copytab', output=f, columns=('a', 'b~'),
                schema_name='public', format='csv',
                delimiter='|', null='n-u-l-l', header=True,
                quote='*', escape='!', force_quote=('a',))

            output = f.getvalue().decode().split('\n')

            self.assertEqual(
                output,
                [
                    'a|b~',
                    '*a1*|b1',
                    '*a2*|b2',
                    '*a3*|b3',
                    '*a4*|b4',
                    '*a5*|b5',
                    '*!**|n-u-l-l',
                    ''
                ]