1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
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",)
|
