Method test_extra_codec_alias

tests/test_codecs.py:1113–1184

Test encoding/decoding of a builtin non-pg_catalog codec.

(self)

            await self.con.fetch("SELECT $1::int4multirange", 1)

    async def test_extra_codec_alias(self):
        """Test encoding/decoding of a builtin non-pg_catalog codec."""
        await self.con.execute('''
            CREATE DOMAIN my_dec_t AS decimal;
            CREATE EXTENSION IF NOT EXISTS hstore;
            CREATE TYPE rec_t AS ( i my_dec_t, h hstore );
        ''')

        try:
            await self.con.set_builtin_type_codec(
                'hstore', codec_name='pg_contrib.hstore')

            cases = [
                {'ham': 'spam', 'nada': None},
                {}
            ]

            st = await self.con.prepare('''
                SELECT $1::hstore AS result
            ''')

            for case in cases:
                res = await st.fetchval(case)
                self.assertEqual(res, case)

            res = await self.con.fetchval('''
                SELECT $1::hstore AS result
            ''', (('foo', '2'), ('bar', '3')))

            self.assertEqual(res, {'foo': '2', 'bar': '3'})

            with self.assertRaisesRegex(asyncpg.DataError,
                                        'null value not allowed'):
                await self.con.fetchval('''
                    SELECT $1::hstore AS result
                ''', {None: '1'})

            await self.con.set_builtin_type_codec(
                'my_dec_t', codec_name='decimal')

            res = await self.con.fetchval('''
                SELECT $1::my_dec_t AS result
            ''', 44)

            self.assertEqual(res, 44)

            # Both my_dec_t and hstore are decoded in binary
            res = await self.con.fetchval('''
                SELECT ($1::my_dec_t, 'a=>1'::hstore)::rec_t AS result
            ''', 44)

            self.assertEqual(res, (44, {'a': '1'}))

            # Now, declare only the text format for my_dec_t
            await self.con.reset_type_codec('my_dec_t')
            await self.con.set_builtin_type_codec(
                'my_dec_t', codec_name='decimal', format='text')