-
Notifications
You must be signed in to change notification settings - Fork 14
Expand file tree
/
Copy pathtest_ssl_mode.py
More file actions
100 lines (89 loc) · 2.26 KB
/
test_ssl_mode.py
File metadata and controls
100 lines (89 loc) · 2.26 KB
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
import pytest
from psqlpy import ConnectionPool, SslMode
from psqlpy._internal import ConnectionPoolBuilder
pytestmark = pytest.mark.anyio
@pytest.mark.parametrize(
"ssl_mode",
[
SslMode.Disable,
SslMode.Allow,
SslMode.Prefer,
SslMode.Require,
SslMode.VerifyCa,
SslMode.VerifyFull,
],
)
async def test_ssl_mode_require(
ssl_mode: SslMode,
postgres_host: str,
postgres_user: str,
postgres_password: str,
postgres_port: int,
postgres_dbname: str,
ssl_cert_file: str,
) -> None:
pg_pool = ConnectionPool(
username=postgres_user,
password=postgres_password,
host=postgres_host,
port=postgres_port,
db_name=postgres_dbname,
ssl_mode=ssl_mode,
ca_file=ssl_cert_file,
)
conn = await pg_pool.connection()
await conn.execute("SELECT 1")
@pytest.mark.parametrize(
"ssl_mode",
[
SslMode.Disable,
SslMode.Allow,
SslMode.Prefer,
SslMode.Require,
SslMode.VerifyCa,
SslMode.VerifyFull,
],
)
async def test_ssl_mode_require_pool_builder(
ssl_mode: SslMode,
postgres_host: str,
postgres_user: str,
postgres_password: str,
postgres_port: int,
postgres_dbname: str,
ssl_cert_file: str,
) -> None:
builder = (
ConnectionPoolBuilder()
.max_pool_size(10)
.host(postgres_host)
.port(postgres_port)
.user(postgres_user)
.password(postgres_password)
.dbname(postgres_dbname)
.ssl_mode(ssl_mode)
.ca_file(ssl_cert_file)
)
pool = builder.build()
connection = await pool.connection()
await connection.execute("SELECT 1")
async def test_ssl_mode_require_without_ca_file(
postgres_host: str,
postgres_user: str,
postgres_password: str,
postgres_port: int,
postgres_dbname: str,
) -> None:
builder = (
ConnectionPoolBuilder()
.max_pool_size(10)
.host(postgres_host)
.port(postgres_port)
.user(postgres_user)
.password(postgres_password)
.dbname(postgres_dbname)
.ssl_mode(SslMode.Require)
)
pool = builder.build()
connection = await pool.connection()
await connection.execute("SELECT 1")