Method test_copy_from_query_to_sink

tests/test_copy.py:234–259

(self)

            ''', 3, output=1)

    async def test_copy_from_query_to_sink(self):
        with tempfile.NamedTemporaryFile() as f:
            async def writer(data):
                # Sleeping here to simulate slow output sink to test
                # backpressure.
                await asyncio.sleep(0.05)
                f.write(data)

            await self.con.copy_from_query('''
                SELECT
                    repeat('a', 500)
                FROM
                    generate_series(1, 5000) AS i
            ''', output=writer)

            f.seek(0)

            output = f.read().decode().split('\n')
            self.assertEqual(
                output,
                [
                    'a' * 500
                ] * 5000 + ['']
            )

        self.assertEqual(await self.con.fetchval('SELECT 1'), 1)

    async def test_copy_from_query_cancellation_explicit(self):
        async def writer(data):