databases/tests/test_connection_options.py at 9acbd784dad20bce0b7ed7740343b8e8ff1f9eb7 · encode/databases · GitHub

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
"""
Unit tests for the backend connection arguments.
"""
import pytest

from databases.backends.aiopg import AiopgBackend
from databases.backends.mysql import MySQLBackend
from databases.backends.postgres import PostgresBackend
from databases.core import DatabaseURL
from tests.test_databases import DATABASE_URLS


def test_postgres_pool_size():
    backend = PostgresBackend("postgres://localhost/database?min_size=1&max_size=20")
    kwargs = backend._get_connection_kwargs()
    assert kwargs == {"min_size": 1, "max_size": 20}


@pytest.mark.anyio
async def test_postgres_pool_size_connect():
    for url in DATABASE_URLS:
        if DatabaseURL(url).dialect != "postgresql":
            continue
        backend = PostgresBackend(url + "?min_size=1&max_size=20")
        await backend.connect()
        await backend.disconnect()


def test_postgres_explicit_pool_size():
    backend = PostgresBackend("postgres://localhost/database", min_size=1, max_size=20)
    kwargs = backend._get_connection_kwargs()
    assert kwargs == {"min_size": 1, "max_size": 20}


def test_postgres_ssl():
    backend = PostgresBackend("postgres://localhost/database?ssl=true")
    kwargs = backend._get_connection_kwargs()
    assert kwargs == {"ssl": True}


def test_postgres_explicit_ssl():
    backend = PostgresBackend("postgres://localhost/database", ssl=True)
    kwargs = backend._get_connection_kwargs()
    assert kwargs == {"ssl": True}


def test_postgres_no_extra_options():
    backend = PostgresBackend("postgres://localhost/database")
    kwargs = backend._get_connection_kwargs()
    assert kwargs == {}


def test_postgres_password_as_callable():
    def gen_password():
        return "Foo"

    backend = PostgresBackend(
        "postgres://:password@localhost/database", password=gen_password
    )
    kwargs = backend._get_connection_kwargs()
    assert kwargs == {"password": gen_password}
    assert kwargs["password"]() == "Foo"


def test_mysql_pool_size():
    backend = MySQLBackend("mysql://localhost/database?min_size=1&max_size=20")
    kwargs = backend._get_connection_kwargs()
    assert kwargs == {"minsize": 1, "maxsize": 20}


def test_mysql_explicit_pool_size():
    backend = MySQLBackend("mysql://localhost/database", min_size=1, max_size=20)
    kwargs = backend._get_connection_kwargs()
    assert kwargs == {"minsize": 1, "maxsize": 20}


def test_mysql_ssl():
    backend = MySQLBackend("mysql://localhost/database?ssl=true")
    kwargs = backend._get_connection_kwargs()
    assert kwargs == {"ssl": True}


def test_mysql_explicit_ssl():
    backend = MySQLBackend("mysql://localhost/database", ssl=True)
    kwargs = backend._get_connection_kwargs()
    assert kwargs == {"ssl": True}


def test_mysql_pool_recycle():
    backend = MySQLBackend("mysql://localhost/database?pool_recycle=20")
    kwargs = backend._get_connection_kwargs()
    assert kwargs == {"pool_recycle": 20}


def test_aiopg_pool_size():
    backend = AiopgBackend(
        "postgresql+aiopg://localhost/database?min_size=1&max_size=20"
    )
    kwargs = backend._get_connection_kwargs()
    assert kwargs == {"minsize": 1, "maxsize": 20}


def test_aiopg_explicit_pool_size():
    backend = AiopgBackend(
        "postgresql+aiopg://localhost/database", min_size=1, max_size=20
    )
    kwargs = backend._get_connection_kwargs()
    assert kwargs == {"minsize": 1, "maxsize": 20}


def test_aiopg_ssl():
    backend = AiopgBackend("postgresql+aiopg://localhost/database?ssl=true")
    kwargs = backend._get_connection_kwargs()
    assert kwargs == {"ssl": True}


def test_aiopg_explicit_ssl():
    backend = AiopgBackend("postgresql+aiopg://localhost/database", ssl=True)
    kwargs = backend._get_connection_kwargs()
    assert kwargs == {"ssl": True}