Method test_copy_from_query_with_args

tests/test_copy.py:146–169

(self)

        )

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

        res = await self.con.copy_from_query('''
            SELECT
                i,
                i * 10,
                $2::text
            FROM
                generate_series(1, 5) AS i
            WHERE
                i = $1
        ''', 3, None, output=f)

        self.assertEqual(res, 'COPY 1')

        output = f.getvalue().decode().split('\n')
        self.assertEqual(
            output,
            [
                '3\t30\t\\N',
                ''
            ]
        )

    async def test_copy_from_query_to_path(self):
        with tempfile.NamedTemporaryFile() as f: