Test overriding core codecs.
(self)
| 1414 | await self.con.execute('DROP TYPE custom_codec_t') |
| 1415 | |
| 1416 | async def test_custom_codec_override_binary(self): |
| 1417 | """Test overriding core codecs.""" |
| 1418 | import json |
| 1419 | |
| 1420 | conn = await self.connect() |
| 1421 | try: |
| 1422 | def _encoder(value): |
| 1423 | return json.dumps(value).encode('utf-8') |
| 1424 | |
| 1425 | def _decoder(value): |
| 1426 | return json.loads(value.decode('utf-8')) |
| 1427 | |
| 1428 | await conn.set_type_codec( |
| 1429 | 'json', encoder=_encoder, decoder=_decoder, |
| 1430 | schema='pg_catalog', format='binary' |
| 1431 | ) |
| 1432 | |
| 1433 | data = {'foo': 'bar', 'spam': 1} |
| 1434 | res = await conn.fetchval('SELECT $1::json', data) |
| 1435 | self.assertEqual(data, res) |
| 1436 | |
| 1437 | finally: |
| 1438 | await conn.close() |
| 1439 | |
| 1440 | async def test_custom_codec_override_text(self): |
| 1441 | """Test overriding core codecs.""" |
nothing calls this directly
no test coverage detected