Make sure set_type_codec() overrides core codec for all formats.

When a custom codec is set for a type, it should be called regardless of
whether its format matches the preferred format requested by the
pipeline.

Fixes: #140
Fixes: #148

Author: elprans

asyncpg/protocol/codecs/base.pxd

@@ -161,3 +161,5 @@ cdef class DataCodecConfig:
         dict _local_type_codecs
 
     cdef inline Codec get_codec(self, uint32_t oid, CodecFormat format)
+    cdef inline Codec get_local_codec(
+        self, uint32_t oid, CodecFormat preferred_format=*)

asyncpg/protocol/codecs/base.pyx

@@ -563,17 +563,39 @@ cdef class DataCodecConfig:
     cdef inline Codec get_codec(self, uint32_t oid, CodecFormat format):
         cdef Codec codec
 
-        try:
-            return self._local_type_codecs[oid, format]
-        except KeyError:
-            codec = get_core_codec(oid, format)
-            if codec is not None:
+        codec = self.get_local_codec(oid, format)
+        if codec is not None:
+            if codec.format != format:
+                # The codec for this OID has been overridden by
+                # set_{builtin}_type_codec with a different format.
+                # We must respect that and not return a core codec.
+                return None
+            else:
                 return codec
+
+        codec = get_core_codec(oid, format)
+        if codec is not None:
+            return codec
+        else:
+            try:
+                return self._type_codecs_cache[oid, format]
+            except KeyError:
+                return None
+
+    cdef inline Codec get_local_codec(
+            self, uint32_t oid, CodecFormat preferred_format=PG_FORMAT_BINARY):
+        cdef Codec codec
+
+        codec = self._local_type_codecs.get((oid, preferred_format))
+        if codec is None:
+            if preferred_format == PG_FORMAT_BINARY:
+                alt_format = PG_FORMAT_TEXT
             else:
-                try:
-                    return self._type_codecs_cache[oid, format]
-                except KeyError:
-                    return None
+                alt_format = PG_FORMAT_BINARY
+
+            codec = self._local_type_codecs.get((oid, alt_format))
+
+        return codec
 
 
 cdef inline Codec get_core_codec(uint32_t oid, CodecFormat format):

tests/test_codecs.py

@@ -924,7 +924,7 @@ def hstore_encoder(obj):
                 DROP EXTENSION hstore
             ''')
 
-    async def test_custom_codec_override(self):
+    async def test_custom_codec_override_binary(self):
         """Test overriding core codecs."""
         import json
 
@@ -948,6 +948,44 @@ def _decoder(value):
         finally:
             await conn.close()
 
+    async def test_custom_codec_override_text(self):
+        """Test overriding core codecs."""
+        import json
+
+        conn = await self.cluster.connect(database='postgres', loop=self.loop)
+        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', binary=False
+            )
+
+            data = {'foo': 'bar', 'spam': 1}
+            res = await conn.fetchval('SELECT $1::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', binary=False
+            )
+
+            data = '14058ad9-0118-4b7e-ac15-01bc13e2ccd1'
+            res = await conn.fetchval('SELECT $1::uuid', data)
+            self.assertEqual(res, data)
+        finally:
+            await conn.close()
+
     async def test_composites_in_arrays(self):
         await self.con.execute('''
             CREATE TYPE t AS (a text, b int);