Method test_copy_to_table_basics

tests/test_copy.py:366–460

(self)

class TestCopyTo(tb.ConnectedTestCase):

    async def test_copy_to_table_basics(self):
        await self.con.execute('''
            CREATE TABLE copytab(a text, "b~" text, i int);
        ''')

        try:
            f = io.BytesIO()
            f.write(
                '\n'.join([
                    'a1\tb1\t1',
                    'a2\tb2\t2',
                    'a3\tb3\t3',
                    'a4\tb4\t4',
                    'a5\tb5\t5',
                    '*\t\\N\t\\N',
                    ''
                ]).encode('utf-8')
            )
            f.seek(0)

            res = await self.con.copy_to_table('copytab', source=f)
            self.assertEqual(res, 'COPY 6')

            output = await self.con.fetch("""
                SELECT * FROM copytab ORDER BY a
            """)
            self.assertEqual(
                output,
                [
                    ('*', None, None),
                    ('a1', 'b1', 1),
                    ('a2', 'b2', 2),
                    ('a3', 'b3', 3),
                    ('a4', 'b4', 4),
                    ('a5', 'b5', 5),
                ]
            )

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

            f.seek(0)
            f.truncate()

            f.write(
                '\n'.join([
                    'a|b~',
                    '*a1*|b1',
                    '*a2*|b2',
                    '*a3*|b3',
                    '*a4*|b4',
                    '*a5*|b5',
                    '*!**|*n-u-l-l*',
                    'n-u-l-l|bb',
                ]).encode('utf-8')
            )
            f.seek(0)