1 files changed, 78 insertions, 0 deletions
diff --git a/tests/crdb/test_adapt.py b/tests/crdb/test_adapt.py
new file mode 100644
index 0000000..ce5bacf
--- /dev/null
+++ b/tests/crdb/test_adapt.py
@@ -0,0 +1,78 @@
+from copy import deepcopy
+
+import pytest
+
+from psycopg.crdb import adapters, CrdbConnection
+
+from psycopg.adapt import PyFormat, Transformer
+from psycopg.types.array import ListDumper
+from psycopg.postgres import types as builtins
+
+from ..test_adapt import MyStr, make_dumper, make_bin_dumper
+from ..test_adapt import make_loader, make_bin_loader
+
+pytestmark = pytest.mark.crdb
+
+
+@pytest.mark.parametrize("fmt_in", PyFormat)
+def test_return_untyped(conn, fmt_in):
+    # Analyze and check for changes using strings in untyped/typed contexts
+    cur = conn.cursor()
+    # Currently string are passed as text oid to CockroachDB, unlike Postgres,
+    # to which strings are passed as unknown. This is because CRDB doesn't
+    # allow the unknown oid to be emitted; execute("SELECT %s", ["str"]) raises
+    # an error. However, unlike PostgreSQL, text can be cast to any other type.
+    cur.execute(f"select %{fmt_in.value}, %{fmt_in.value}", ["hello", 10])
+    assert cur.fetchone() == ("hello", 10)
+
+    cur.execute("create table testjson(data jsonb)")
+    cur.execute(f"insert into testjson (data) values (%{fmt_in.value})", ["{}"])
+    assert cur.execute("select data from testjson").fetchone() == ({},)
+
+
+def test_str_list_dumper_text(conn):
+    t = Transformer(conn)
+    dstr = t.get_dumper([""], PyFormat.TEXT)
+    assert isinstance(dstr, ListDumper)
+    assert dstr.oid == builtins["text"].array_oid
+    assert dstr.sub_dumper and dstr.sub_dumper.oid == builtins["text"].oid
+
+
+@pytest.fixture
+def crdb_adapters():
+    """Restore the crdb adapters after a test has changed them."""
+    dumpers = deepcopy(adapters._dumpers)
+    dumpers_by_oid = deepcopy(adapters._dumpers_by_oid)
+    loaders = deepcopy(adapters._loaders)
+    types = list(adapters.types)
+
+    yield None
+
+    adapters._dumpers = dumpers
+    adapters._dumpers_by_oid = dumpers_by_oid
+    adapters._loaders = loaders
+    adapters.types.clear()
+    for t in types:
+        adapters.types.add(t)
+
+
+def test_dump_global_ctx(dsn, crdb_adapters, pgconn):
+    adapters.register_dumper(MyStr, make_bin_dumper("gb"))
+    adapters.register_dumper(MyStr, make_dumper("gt"))
+    with CrdbConnection.connect(dsn) as conn:
+        cur = conn.execute("select %s", [MyStr("hello")])
+        assert cur.fetchone() == ("hellogt",)
+        cur = conn.execute("select %b", [MyStr("hello")])
+        assert cur.fetchone() == ("hellogb",)
+        cur = conn.execute("select %t", [MyStr("hello")])
+        assert cur.fetchone() == ("hellogt",)
+
+
+def test_load_global_ctx(dsn, crdb_adapters):
+    adapters.register_loader("text", make_loader("gt"))
+    adapters.register_loader("text", make_bin_loader("gb"))
+    with CrdbConnection.connect(dsn) as conn:
+        cur = conn.cursor(binary=False).execute("select 'hello'::text")
+        assert cur.fetchone() == ("hellogt",)
+        cur = conn.cursor(binary=True).execute("select 'hello'::text")
+        assert cur.fetchone() == ("hellogb",)