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
|
from os import environ
import pytest
from test.utils import parse_server_version
import pg8000.dbapi
@pytest.fixture(scope="class")
def db_kwargs():
db_connect = {"user": "postgres", "password": "pw"}
for kw, var, f in [
("host", "PGHOST", str),
("password", "PGPASSWORD", str),
("port", "PGPORT", int),
]:
try:
db_connect[kw] = f(environ[var])
except KeyError:
pass
return db_connect
@pytest.fixture
def con(request, db_kwargs):
conn = pg8000.dbapi.connect(**db_kwargs)
def fin():
try:
conn.rollback()
except pg8000.dbapi.InterfaceError:
pass
try:
conn.close()
except pg8000.dbapi.InterfaceError:
pass
request.addfinalizer(fin)
return conn
@pytest.fixture
def cursor(request, con):
cursor = con.cursor()
def fin():
cursor.close()
request.addfinalizer(fin)
return cursor
@pytest.fixture
def pg_version(cursor):
cursor.execute("select current_setting('server_version')")
retval = cursor.fetchall()
version = retval[0][0]
major = parse_server_version(version)
return major
|
