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
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
|
"""
Support module for test_connection[_async].py
"""
from __future__ import annotations
from typing import Any
from dataclasses import dataclass
import pytest
import psycopg
from psycopg.conninfo import conninfo_to_dict
try:
from psycopg.conninfo import _DEFAULT_CONNECT_TIMEOUT as DEFAULT_TIMEOUT
except ImportError:
# Allow tests to import (not necessarily to pass all) if the psycopg module
# imported is not the one expected (e.g. running psycopg pool tests on the
# master branch with psycopg 3.1.x imported).
DEFAULT_TIMEOUT = 130
@pytest.fixture
def testctx(svcconn):
svcconn.execute("create table if not exists testctx (id int primary key)")
svcconn.execute("delete from testctx")
return None
@dataclass
class ParamDef:
name: str
guc: str
values: list[Any]
non_default: str
param_isolation = ParamDef(
name="isolation_level",
guc="isolation",
values=list(psycopg.IsolationLevel),
non_default="serializable",
)
param_read_only = ParamDef(
name="read_only",
guc="read_only",
values=[True, False],
non_default="on",
)
param_deferrable = ParamDef(
name="deferrable",
guc="deferrable",
values=[True, False],
non_default="on",
)
# Map Python values to Postgres values for the tx_params possible values
tx_values_map = {
v.name.lower().replace("_", " "): v.value for v in psycopg.IsolationLevel
}
tx_values_map["on"] = True
tx_values_map["off"] = False
tx_params = [
param_isolation,
param_read_only,
pytest.param(param_deferrable, marks=pytest.mark.crdb_skip("deferrable")),
]
tx_params_isolation = [
pytest.param(
param_isolation,
id="isolation_level",
marks=pytest.mark.crdb("skip", reason="transaction isolation"),
),
pytest.param(
param_read_only, id="read_only", marks=pytest.mark.crdb_skip("begin_read_only")
),
pytest.param(
param_deferrable, id="deferrable", marks=pytest.mark.crdb_skip("deferrable")
),
]
conninfo_params_timeout = [
(
"",
{"dbname": "mydb", "connect_timeout": None},
({"dbname": "mydb"}, DEFAULT_TIMEOUT),
),
(
"",
{"dbname": "mydb", "connect_timeout": 1},
({"dbname": "mydb", "connect_timeout": 1}, 2),
),
(
"dbname=postgres",
{},
({"dbname": "postgres"}, DEFAULT_TIMEOUT),
),
(
"dbname=postgres connect_timeout=2",
{},
({"dbname": "postgres", "connect_timeout": "2"}, 2),
),
(
"postgresql:///postgres?connect_timeout=2",
{"connect_timeout": 10},
({"dbname": "postgres", "connect_timeout": 10}, 10),
),
]
def drop_default_args_from_conninfo(conninfo):
if isinstance(conninfo, str):
params = conninfo_to_dict(conninfo)
else:
params = conninfo.copy()
def removeif(key, value):
if params.get(key) == value:
params.pop(key)
removeif("connect_timeout", str(DEFAULT_TIMEOUT))
return params
|
