test(crdb): fix most cursor tests to run with CockroachDB

Move some common test markers to the fix_crdb module.

Remaining cursor failing test may require some tweak:

- something with binary parameters
- something with ::text cast added by crdb dumper

diff --git a/tests/fix_crdb.py b/tests/fix_crdb.py
index 06ee3aaadf722e847da76b9993c1860a8818db80..2647da1399977e032e6eba72c6f1dc3bd69e344d 100644 (file)
--- a/tests/fix_crdb.py
+++ b/tests/fix_crdb.py
@@ -1,3 +1,5 @@
+import pytest
+
 from .utils import check_version
 
 
@@ -63,6 +65,30 @@ def check_crdb_version(got, func):
     return rv
 
 
+# Utility functions which can be imported in the test suite
+
+
+def is_crdb(conn):
+    if hasattr(conn, "pgconn"):
+        conn = conn.pgconn
+
+    return bool(conn.parameter_status(b"crdb_version"))
+
+
+def skip_crdb(*args, reason=None):
+    return pytest.param(*args, marks=pytest.mark.crdb("skip", reason=reason))
+
+
+def crdb_encoding(*args):
+    """Mark tests that fail on CockroachDB because of missing encodings"""
+    return skip_crdb(*args, reason="encoding")
+
+
+def crdb_time_precision(*args):
+    """Mark tests that fail on CockroachDB because time doesn't support precision"""
+    return skip_crdb(*args, reason="time precision")
+
+
 # mapping from reason description to ticket number
 crdb_reasons = {
     "2-phase commit": 22329,

diff --git a/tests/test_client_cursor.py b/tests/test_client_cursor.py
index e42993b7183fed1c0d09c923a98413457b470469..cc6944233f532ae8969d0f59987df8b5e4edd309 100644 (file)
--- a/tests/test_client_cursor.py
+++ b/tests/test_client_cursor.py
@@ -13,6 +13,7 @@ from psycopg.postgres import types as builtins
 
 from .utils import gc_collect
 from .test_cursor import my_row_factory
+from .fix_crdb import is_crdb, crdb_encoding, crdb_time_precision
 
 
 @pytest.fixture
@@ -251,7 +252,7 @@ def test_binary_cursor_text_override(conn):
     assert cur.pgresult.get_value(0, 0) == b"1"
 
 
-@pytest.mark.parametrize("encoding", ["utf8", "latin9"])
+@pytest.mark.parametrize("encoding", ["utf8", crdb_encoding("latin9")])
 def test_query_encode(conn, encoding):
     conn.execute(f"set client_encoding to {encoding}")
     cur = conn.cursor()
@@ -259,8 +260,9 @@ def test_query_encode(conn, encoding):
     assert res == "\u20ac"
 
 
-def test_query_badenc(conn):
-    conn.execute("set client_encoding to latin1")
+@pytest.mark.parametrize("encoding", [crdb_encoding("latin1")])
+def test_query_badenc(conn, encoding):
+    conn.execute(f"set client_encoding to {encoding}")
     cur = conn.cursor()
     with pytest.raises(UnicodeEncodeError):
         cur.execute("select '\u20ac'")
@@ -584,6 +586,7 @@ def test_query_params_executemany(conn):
     assert cur._query.params == (b"3", b"4")
 
 
+@pytest.mark.crdb("skip", reason="copy")
 @pytest.mark.parametrize("ph, params", [("%s", (10,)), ("%(n)s", {"n": 10})])
 def test_copy_out_param(conn, ph, params):
     cur = conn.cursor()
@@ -651,7 +654,10 @@ class TestColumn:
         assert c.name == "now"
         assert c.type_code == builtins["date"].oid
         assert c.display_size is None
-        assert c.internal_size == 4
+        if is_crdb(conn):
+            assert c.internal_size == 16
+        else:
+            assert c.internal_size == 4
         assert c.precision is None
         assert c.scale is None
 
@@ -671,8 +677,8 @@ class TestColumn:
             ("numeric(10)", 10, 0, None, None),
             ("numeric(10, 3)", 10, 3, None, None),
             ("time", None, None, None, 8),
-            ("time(4)", 4, None, None, 8),
-            ("time(10)", 6, None, None, 8),
+            crdb_time_precision("time(4)", 4, None, None, 8),
+            crdb_time_precision("time(10)", 6, None, None, 8),
         ],
     )
     def test_details(self, conn, type, precision, scale, dsize, isize):
@@ -699,6 +705,7 @@ class TestColumn:
         unpickled = pickle.loads(pickled)
         assert [tuple(d) for d in description] == [tuple(d) for d in unpickled]
 
+    @pytest.mark.crdb("skip", reason="no col query")
     def test_no_col_query(self, conn):
         cur = conn.execute("select")
         assert cur.description == []
@@ -720,7 +727,7 @@ class TestColumn:
         assert res == "x"
         assert cur.description[0].name == "foo-bar"
 
-    @pytest.mark.parametrize("encoding", ["utf8", "latin9"])
+    @pytest.mark.parametrize("encoding", ["utf8", crdb_encoding("latin9")])
     def test_name_encode(self, conn, encoding):
         conn.execute(f"set client_encoding to {encoding}")
         cur = conn.cursor()
@@ -796,17 +803,19 @@ def test_mogrify(conn):
     q = cur.mogrify("select %s, %s", [1, dt.date(2020, 1, 1)])
     assert q == "select 1, '2020-01-01'::date"
 
-    conn.execute("set client_encoding to utf8")
-    q = cur.mogrify("select %(s)s", {"s": "\u20ac"})
-    assert q == "select '\u20ac'"
 
-    conn.execute("set client_encoding to latin9")
-    q = cur.mogrify("select %(s)s", {"s": "\u20ac"})
+@pytest.mark.parametrize("encoding", ["utf8", crdb_encoding("latin9")])
+def test_mogrify_encoding(conn, encoding):
+    conn.execute(f"set client_encoding to {encoding}")
+    q = conn.cursor().mogrify("select %(s)s", {"s": "\u20ac"})
     assert q == "select '\u20ac'"
 
-    conn.execute("set client_encoding to latin1")
+
+@pytest.mark.parametrize("encoding", [crdb_encoding("latin1")])
+def test_mogrify_badenc(conn, encoding):
+    conn.execute(f"set client_encoding to {encoding}")
     with pytest.raises(UnicodeEncodeError):
-        cur.mogrify("select %(s)s", {"s": "\u20ac"})
+        conn.cursor().mogrify("select %(s)s", {"s": "\u20ac"})
 
 
 @pytest.mark.libpq(">= 14")

diff --git a/tests/test_client_cursor_async.py b/tests/test_client_cursor_async.py
index 7bfd5666d18d7e00dbe36db1927b4b93e29e7746..20393a617acdd358f1d97a006257f106985345e1 100644 (file)
--- a/tests/test_client_cursor_async.py
+++ b/tests/test_client_cursor_async.py
@@ -11,6 +11,7 @@ from psycopg.adapt import PyFormat
 from .utils import alist, gc_collect
 from .test_cursor import my_row_factory
 from .test_cursor import execmany, _execmany  # noqa: F401
+from .fix_crdb import crdb_encoding
 
 execmany = execmany  # avoid F811 underneath
 pytestmark = pytest.mark.asyncio
@@ -255,7 +256,7 @@ async def test_binary_cursor_text_override(aconn):
     assert cur.pgresult.get_value(0, 0) == b"1"
 
 
-@pytest.mark.parametrize("encoding", ["utf8", "latin9"])
+@pytest.mark.parametrize("encoding", ["utf8", crdb_encoding("latin9")])
 async def test_query_encode(aconn, encoding):
     await aconn.execute(f"set client_encoding to {encoding}")
     cur = aconn.cursor()
@@ -264,8 +265,9 @@ async def test_query_encode(aconn, encoding):
     assert res == "\u20ac"
 
 
-async def test_query_badenc(aconn):
-    await aconn.execute("set client_encoding to latin1")
+@pytest.mark.parametrize("encoding", [crdb_encoding("latin1")])
+async def test_query_badenc(aconn, encoding):
+    await aconn.execute(f"set client_encoding to {encoding}")
     cur = aconn.cursor()
     with pytest.raises(UnicodeEncodeError):
         await cur.execute("select '\u20ac'")
@@ -579,6 +581,7 @@ async def test_query_params_executemany(aconn):
     assert cur._query.params == (b"3", b"4")
 
 
+@pytest.mark.crdb("skip", reason="copy")
 @pytest.mark.parametrize("ph, params", [("%s", (10,)), ("%(n)s", {"n": 10})])
 async def test_copy_out_param(aconn, ph, params):
     cur = aconn.cursor()
@@ -670,17 +673,19 @@ async def test_mogrify(aconn):
     q = cur.mogrify("select %s, %s", [1, dt.date(2020, 1, 1)])
     assert q == "select 1, '2020-01-01'::date"
 
-    await aconn.execute("set client_encoding to utf8")
-    q = cur.mogrify("select %(s)s", {"s": "\u20ac"})
-    assert q == "select '\u20ac'"
 
-    await aconn.execute("set client_encoding to latin9")
-    q = cur.mogrify("select %(s)s", {"s": "\u20ac"})
+@pytest.mark.parametrize("encoding", ["utf8", crdb_encoding("latin9")])
+async def test_mogrify_encoding(aconn, encoding):
+    await aconn.execute(f"set client_encoding to {encoding}")
+    q = aconn.cursor().mogrify("select %(s)s", {"s": "\u20ac"})
     assert q == "select '\u20ac'"
 
-    await aconn.execute("set client_encoding to latin1")
+
+@pytest.mark.parametrize("encoding", [crdb_encoding("latin1")])
+async def test_mogrify_badenc(aconn, encoding):
+    await aconn.execute(f"set client_encoding to {encoding}")
     with pytest.raises(UnicodeEncodeError):
-        cur.mogrify("select %(s)s", {"s": "\u20ac"})
+        aconn.cursor().mogrify("select %(s)s", {"s": "\u20ac"})
 
 
 @pytest.mark.libpq(">= 14")

diff --git a/tests/test_conninfo.py b/tests/test_conninfo.py
index 8531cd573157920237f3890745a63885603b7d5c..a94a101bd7f0ededabc61b523cc089eb3720cd41 100644 (file)
--- a/tests/test_conninfo.py
+++ b/tests/test_conninfo.py
@@ -10,11 +10,9 @@ from psycopg.conninfo import make_conninfo, conninfo_to_dict, ConnectionInfo
 from psycopg.conninfo import resolve_hostaddr_async
 from psycopg._encodings import pg2pyenc
 
-snowman = "\u2603"
-
+from .fix_crdb import crdb_encoding
 
-def skip_crdb(*args, reason=None):
-    return pytest.param(*args, marks=pytest.mark.crdb("skip", reason=reason))
+snowman = "\u2603"
 
 
 class MyString(str):
@@ -230,7 +228,7 @@ class TestConnectionInfo:
         with pytest.raises(psycopg.OperationalError):
             conn.info.error_message
 
-    @pytest.mark.crdb("skip", reason="always 0 on crdb")
+    @pytest.mark.crdb("skip", reason="backend pid")
     def test_backend_pid(self, conn):
         assert conn.info.backend_pid
         assert conn.info.backend_pid == conn.pgconn.backend_pid
@@ -292,8 +290,8 @@ class TestConnectionInfo:
             ("utf8", "UTF8", "utf-8"),
             ("utf-8", "UTF8", "utf-8"),
             ("utf_8", "UTF8", "utf-8"),
-            skip_crdb("eucjp", "EUC_JP", "euc_jp", reason="encoding"),
-            skip_crdb("euc-jp", "EUC_JP", "euc_jp", reason="encoding"),
+            crdb_encoding("eucjp", "EUC_JP", "euc_jp"),
+            crdb_encoding("euc-jp", "EUC_JP", "euc_jp"),
         ],
     )
     def test_encoding_env_var(self, dsn, monkeypatch, enc, out, codec):

diff --git a/tests/test_cursor.py b/tests/test_cursor.py
index 7c7db7b95208a536588101d24998b1d760909454..da0576ac3893f75b3462a57808c9322c3b6ece49 100644 (file)
--- a/tests/test_cursor.py
+++ b/tests/test_cursor.py
@@ -13,6 +13,7 @@ from psycopg.postgres import types as builtins
 from psycopg.rows import RowMaker
 
 from .utils import gc_collect
+from .fix_crdb import is_crdb, crdb_encoding, crdb_time_precision
 
 
 def test_init(conn):
@@ -232,7 +233,7 @@ def test_binary_cursor_text_override(conn):
     assert cur.pgresult.get_value(0, 0) == b"1"
 
 
-@pytest.mark.parametrize("encoding", ["utf8", "latin9"])
+@pytest.mark.parametrize("encoding", ["utf8", crdb_encoding("latin9")])
 def test_query_encode(conn, encoding):
     conn.execute(f"set client_encoding to {encoding}")
     cur = conn.cursor()
@@ -240,8 +241,9 @@ def test_query_encode(conn, encoding):
     assert res == "\u20ac"
 
 
-def test_query_badenc(conn):
-    conn.execute("set client_encoding to latin1")
+@pytest.mark.parametrize("encoding", [crdb_encoding("latin1")])
+def test_query_badenc(conn, encoding):
+    conn.execute(f"set client_encoding to {encoding}")
     cur = conn.cursor()
     with pytest.raises(UnicodeEncodeError):
         cur.execute("select '\u20ac'")
@@ -601,6 +603,7 @@ def test_stream_no_row(conn):
     assert recs == []
 
 
+@pytest.mark.crdb("skip", reason="no col query")
 def test_stream_no_col(conn):
     cur = conn.cursor()
     recs = list(cur.stream("select"))
@@ -727,7 +730,10 @@ class TestColumn:
         assert c.name == "now"
         assert c.type_code == builtins["date"].oid
         assert c.display_size is None
-        assert c.internal_size == 4
+        if is_crdb(conn):
+            assert c.internal_size == 16
+        else:
+            assert c.internal_size == 4
         assert c.precision is None
         assert c.scale is None
 
@@ -747,8 +753,8 @@ class TestColumn:
             ("numeric(10)", 10, 0, None, None),
             ("numeric(10, 3)", 10, 3, None, None),
             ("time", None, None, None, 8),
-            ("time(4)", 4, None, None, 8),
-            ("time(10)", 6, None, None, 8),
+            crdb_time_precision("time(4)", 4, None, None, 8),
+            crdb_time_precision("time(10)", 6, None, None, 8),
         ],
     )
     def test_details(self, conn, type, precision, scale, dsize, isize):
@@ -775,6 +781,7 @@ class TestColumn:
         unpickled = pickle.loads(pickled)
         assert [tuple(d) for d in description] == [tuple(d) for d in unpickled]
 
+    @pytest.mark.crdb("skip", reason="no col query")
     def test_no_col_query(self, conn):
         cur = conn.execute("select")
         assert cur.description == []
@@ -796,7 +803,7 @@ class TestColumn:
         assert res == "x"
         assert cur.description[0].name == "foo-bar"
 
-    @pytest.mark.parametrize("encoding", ["utf8", "latin9"])
+    @pytest.mark.parametrize("encoding", ["utf8", crdb_encoding("latin9")])
     def test_name_encode(self, conn, encoding):
         conn.execute(f"set client_encoding to {encoding}")
         cur = conn.cursor()

diff --git a/tests/test_cursor_async.py b/tests/test_cursor_async.py
index 232f1ae4214b386a9740f12f9e552868ccd1689e..6b3b5974e4f013bcc7c4c0d741db3779208cc77f 100644 (file)
--- a/tests/test_cursor_async.py
+++ b/tests/test_cursor_async.py
@@ -11,6 +11,7 @@ from psycopg.adapt import PyFormat
 from .utils import gc_collect
 from .test_cursor import my_row_factory
 from .test_cursor import execmany, _execmany  # noqa: F401
+from .fix_crdb import crdb_encoding
 
 execmany = execmany  # avoid F811 underneath
 pytestmark = pytest.mark.asyncio
@@ -234,7 +235,7 @@ async def test_binary_cursor_text_override(aconn):
     assert cur.pgresult.get_value(0, 0) == b"1"
 
 
-@pytest.mark.parametrize("encoding", ["utf8", "latin9"])
+@pytest.mark.parametrize("encoding", ["utf8", crdb_encoding("latin9")])
 async def test_query_encode(aconn, encoding):
     await aconn.execute(f"set client_encoding to {encoding}")
     cur = aconn.cursor()
@@ -243,8 +244,9 @@ async def test_query_encode(aconn, encoding):
     assert res == "\u20ac"
 
 
-async def test_query_badenc(aconn):
-    await aconn.execute("set client_encoding to latin1")
+@pytest.mark.parametrize("encoding", [crdb_encoding("latin1")])
+async def test_query_badenc(aconn, encoding):
+    await aconn.execute(f"set client_encoding to {encoding}")
     cur = aconn.cursor()
     with pytest.raises(UnicodeEncodeError):
         await cur.execute("select '\u20ac'")