File size: 1,529 Bytes
7c6ffa6
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
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
from types import SimpleNamespace

from app.core.database import build_engine_options


def test_sqlite_engine_options_keep_thread_check_disabled() -> None:
    settings = SimpleNamespace(database_url="sqlite:///test.db")

    options = build_engine_options(settings)

    assert options == {
        "pool_pre_ping": True,
        "connect_args": {"check_same_thread": False},
    }


def test_postgres_engine_options_use_small_pool_defaults() -> None:
    settings = SimpleNamespace(
        database_url="postgresql+psycopg://user:pass@example.com/db",
        database_pool_size=2,
        database_max_overflow=3,
        database_pool_timeout_seconds=10,
        database_pool_recycle_seconds=1800,
    )

    options = build_engine_options(settings)

    assert options["pool_pre_ping"] is True
    assert options["connect_args"] == {}
    assert options["pool_size"] == 2
    assert options["max_overflow"] == 3
    assert options["pool_timeout"] == 10
    assert options["pool_recycle"] == 1800


def test_postgres_engine_options_clamp_unsafe_values() -> None:
    settings = SimpleNamespace(
        database_url="postgresql+psycopg://user:pass@example.com/db",
        database_pool_size=0,
        database_max_overflow=-2,
        database_pool_timeout_seconds=0,
        database_pool_recycle_seconds=1,
    )

    options = build_engine_options(settings)

    assert options["pool_size"] == 1
    assert options["max_overflow"] == 0
    assert options["pool_timeout"] == 1
    assert options["pool_recycle"] == 60