Method test_custom_codec_override_text

tests/test_codecs.py:1440–1485

Test overriding core codecs.

(self)

            await conn.close()

    async def test_custom_codec_override_text(self):
        """Test overriding core codecs."""
        import json

        conn = await self.connect()
        try:
            def _encoder(value):
                return json.dumps(value)

            def _decoder(value):
                return json.loads(value)

            await conn.set_type_codec(
                'json', encoder=_encoder, decoder=_decoder,
                schema='pg_catalog', format='text'
            )

            data = {'foo': 'bar', 'spam': 1}
            res = await conn.fetchval('SELECT $1::json', data)
            self.assertEqual(data, res)

            res = await conn.fetchval('SELECT $1::json[]', [data])
            self.assertEqual([data], res)

            await conn.execute('CREATE DOMAIN my_json AS json')

            res = await conn.fetchval('SELECT $1::my_json', data)
            self.assertEqual(data, res)

            def _encoder(value):
                return value

            def _decoder(value):
                return value

            await conn.set_type_codec(
                'uuid', encoder=_encoder, decoder=_decoder,
                schema='pg_catalog', format='text'
            )

            data = '14058ad9-0118-4b7e-ac15-01bc13e2ccd1'
            res = await conn.fetchval('SELECT $1::uuid', data)
            self.assertEqual(res, data)
        finally:
            await conn.execute('DROP DOMAIN IF EXISTS my_json')
            await conn.close()

    async def test_custom_codec_override_tuple(self):
        """Test overriding core codecs."""