(self)
| 1939 | self.assertTupleEqual(st.get_attributes(), ()) |
| 1940 | |
| 1941 | async def test_array_with_custom_json_text_codec(self): |
| 1942 | import json |
| 1943 | |
| 1944 | await self.con.execute('CREATE TABLE tab (id serial, val json[]);') |
| 1945 | insert_sql = 'INSERT INTO tab (val) VALUES (cast($1 AS json[]));' |
| 1946 | query_sql = 'SELECT val FROM tab ORDER BY id DESC;' |
| 1947 | try: |
| 1948 | for custom_codec in [False, True]: |
| 1949 | if custom_codec: |
| 1950 | await self.con.set_type_codec( |
| 1951 | 'json', |
| 1952 | encoder=lambda v: v, |
| 1953 | decoder=json.loads, |
| 1954 | schema="pg_catalog", |
| 1955 | ) |
| 1956 | |
| 1957 | for val in ['"null"', '22', 'null', '[2]', '{"a": null}']: |
| 1958 | await self.con.execute(insert_sql, [val]) |
| 1959 | result = await self.con.fetchval(query_sql) |
| 1960 | if custom_codec: |
| 1961 | self.assertEqual(result, [json.loads(val)]) |
| 1962 | else: |
| 1963 | self.assertEqual(result, [val]) |
| 1964 | |
| 1965 | await self.con.execute(insert_sql, [None]) |
| 1966 | result = await self.con.fetchval(query_sql) |
| 1967 | self.assertEqual(result, [None]) |
| 1968 | |
| 1969 | await self.con.execute(insert_sql, None) |
| 1970 | result = await self.con.fetchval(query_sql) |
| 1971 | self.assertEqual(result, None) |
| 1972 | |
| 1973 | finally: |
| 1974 | await self.con.execute(''' |
| 1975 | DROP TABLE tab; |
| 1976 | ''') |
| 1977 | |
| 1978 | |
| 1979 | @unittest.skipIf(os.environ.get('PGHOST'), 'using remote cluster for testing') |
nothing calls this directly
no test coverage detected