psycopg/psycopg2
1
1
Fork 0
mirror of https://github.com/psycopg/psycopg2.git synced 2024-11-22 08:56:34 +03:00
Code Issues Packages Projects Releases Wiki Activity

Types tests adapted to CockroachDB

Browse Source
This commit is contained in:
Daniele Varrazzo 2020-07-23 01:53:37 +01:00
parent c8697e6c67
commit 5e957daa82
2 changed files with 29 additions and 14 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
tests/test_types_basic.py
View File

@ -32,6 +32,7 @@ import platform
from . import testutils
import unittest
from .testutils import PY2, long, text_type, ConnectingTestCase, restore_types
from .testutils import skip_if_crdb
import psycopg2
from psycopg2.extensions import AsIs, adapt, register_adapter
@ -148,12 +149,14 @@ class TypesBasicTests(ConnectingTestCase):
buf2 = self.execute("SELECT %s::bytea AS foo", (buf,))
self.assertEqual(s, buf2.tobytes())
@skip_if_crdb
def testArray(self):
s = self.execute("SELECT %s AS foo", ([[1, 2], [3, 4]],))
self.failUnlessEqual(s, [[1, 2], [3, 4]])
s = self.execute("SELECT %s AS foo", (['one', 'two', 'three'],))
self.failUnlessEqual(s, ['one', 'two', 'three'])
@skip_if_crdb
def testEmptyArrayRegression(self):
# ticket #42
curs = self.conn.cursor()
@ -170,6 +173,7 @@ class TypesBasicTests(ConnectingTestCase):
curs.execute("select col from array_test where id = 2")
self.assertEqual(curs.fetchone()[0], [])
@skip_if_crdb
@testutils.skip_before_postgres(8, 4)
def testNestedEmptyArray(self):
# issue #788
@ -235,6 +239,7 @@ class TypesBasicTests(ConnectingTestCase):
self.assert_(isinstance(x[0], bytes))
self.assertEqual(x, [b'a', b'b', b'c'])
@skip_if_crdb
@testutils.skip_before_postgres(8, 2)
def testArrayOfNulls(self):
curs = self.conn.cursor()
@ -271,6 +276,7 @@ class TypesBasicTests(ConnectingTestCase):
curs.execute("insert into na (boolaa) values (%s)", ([[True, None]],))
curs.execute("insert into na (boolaa) values (%s)", ([[None, None]],))
@skip_if_crdb
@testutils.skip_before_postgres(8, 2)
def testNestedArrays(self):
curs = self.conn.cursor()
@ -400,6 +406,7 @@ class TypesBasicTests(ConnectingTestCase):
a = self.execute("select '{1, 2, NULL}'::int4[]")
self.assertEqual(a, [2, 4, 'nada'])
@skip_if_crdb
@testutils.skip_before_postgres(8, 2)
def testNetworkArray(self):
# we don't know these types, but we know their arrays

36
tests/test_types_extras.py
View File

@ -27,7 +27,7 @@ from pickle import dumps, loads
import unittest
from .testutils import (PY2, text_type, skip_if_no_uuid, skip_before_postgres,
ConnectingTestCase, py3_raises_typeerror, slow, skip_from_python,
restore_types)
restore_types, skip_if_crdb, crdb_version)
import psycopg2
import psycopg2.extras
@ -134,6 +134,7 @@ class TypesExtrasTests(ConnectingTestCase):
def skip_if_no_hstore(f):
@wraps(f)
@skip_if_crdb
def skip_if_no_hstore_(self):
oids = HstoreAdapter.get_oids(self.conn)
if oids is None or not oids[0]:
@ -417,6 +418,7 @@ class HstoreTestCase(ConnectingTestCase):
def skip_if_no_composite(f):
@wraps(f)
@skip_if_crdb
def skip_if_no_composite_(self):
if self.conn.info.server_version < 80000:
return self.skipTest(
@ -786,6 +788,7 @@ def skip_if_no_json_type(f):
return skip_if_no_json_type_
@skip_if_crdb
class JsonTestCase(ConnectingTestCase):
def test_adapt(self):
objs = [None, "te'xt", 123, 123.45,
@ -990,8 +993,9 @@ class JsonbTestCase(ConnectingTestCase):
curs.execute("""select '{"a": 100.0, "b": null}'::jsonb""")
self.assertEqual(curs.fetchone()[0], {'a': 100.0, 'b': None})
curs.execute("""select array['{"a": 100.0, "b": null}']::jsonb[]""")
self.assertEqual(curs.fetchone()[0], [{'a': 100.0, 'b': None}])
if crdb_version(self.conn) is None:
curs.execute("""select array['{"a": 100.0, "b": null}']::jsonb[]""")
self.assertEqual(curs.fetchone()[0], [{'a': 100.0, 'b': None}])
def test_register_on_connection(self):
psycopg2.extras.register_json(self.conn, loads=self.myloads, name='jsonb')
@ -1025,11 +1029,12 @@ class JsonbTestCase(ConnectingTestCase):
data = curs.fetchone()[0]
self.assert_(isinstance(data['a'], Decimal))
self.assertEqual(data['a'], Decimal('100.0'))
# sure we are not manling json too?
curs.execute("""select '{"a": 100.0, "b": null}'::json""")
data = curs.fetchone()[0]
self.assert_(isinstance(data['a'], float))
self.assertEqual(data['a'], 100.0)
# sure we are not mangling json too?
if crdb_version(self.conn) is None:
curs.execute("""select '{"a": 100.0, "b": null}'::json""")
data = curs.fetchone()[0]
self.assert_(isinstance(data['a'], float))
self.assertEqual(data['a'], 100.0)
def test_register_default(self):
curs = self.conn.cursor()
@ -1044,17 +1049,19 @@ class JsonbTestCase(ConnectingTestCase):
self.assert_(isinstance(data['a'], Decimal))
self.assertEqual(data['a'], Decimal('100.0'))
curs.execute("""select array['{"a": 100.0, "b": null}']::jsonb[]""")
data = curs.fetchone()[0]
self.assert_(isinstance(data[0]['a'], Decimal))
self.assertEqual(data[0]['a'], Decimal('100.0'))
if crdb_version(self.conn) is None:
curs.execute("""select array['{"a": 100.0, "b": null}']::jsonb[]""")
data = curs.fetchone()[0]
self.assert_(isinstance(data[0]['a'], Decimal))
self.assertEqual(data[0]['a'], Decimal('100.0'))
def test_null(self):
curs = self.conn.cursor()
curs.execute("""select NULL::jsonb""")
self.assertEqual(curs.fetchone()[0], None)
curs.execute("""select NULL::jsonb[]""")
self.assertEqual(curs.fetchone()[0], None)
if crdb_version(self.conn) is None:
curs.execute("""select NULL::jsonb[]""")
self.assertEqual(curs.fetchone()[0], None)
class RangeTestCase(unittest.TestCase):
@ -1325,6 +1332,7 @@ class RangeTestCase(unittest.TestCase):
self.assertEqual(result, expected)
@skip_if_crdb
@skip_before_postgres(9, 2, "range not supported before postgres 9.2")
class RangeCasterTestCase(ConnectingTestCase):