Coverage for tests/test_db_engine.py: 100%

1""" 

2Connection-pool hardening checks. 

3 

4These do not need a live database: the engine is built lazily, and the one 

5test that connects short-circuits libpq with a ``do_connect`` handler so we 

6can read back the parameters psycopg would have been handed. 

7""" 

8 

9import pytest 

10from sqlalchemy import create_engine, event 

11 

12from bbconf.db_utils import CONNECT_ARGS, POOL_RECYCLE_SECONDS, BaseEngine 

13 

14KEEPALIVE_KEYS = { 

15 "connect_timeout", 

16 "keepalives", 

17 "keepalives_idle", 

18 "keepalives_interval", 

19 "keepalives_count", 

20 "tcp_user_timeout", 

21} 

22 

23 

24def test_connect_args_cover_dead_peer_detection(): 

25 assert KEEPALIVE_KEYS <= set(CONNECT_ARGS) 

26 assert CONNECT_ARGS["keepalives"] == 1 

27 # keepalives_idle + keepalives_interval * keepalives_count: a dead idle 

28 # peer must be noticed in about a minute, not the ~2 h Linux default. 

29 detection = CONNECT_ARGS["keepalives_idle"] + ( 

30 CONNECT_ARGS["keepalives_interval"] * CONNECT_ARGS["keepalives_count"] 

31 ) 

32 assert detection <= 120 

33 assert CONNECT_ARGS["connect_timeout"] <= 10 

34 assert CONNECT_ARGS["tcp_user_timeout"] <= 60_000 

35 

36 

37def test_base_engine_hardens_the_pool(monkeypatch): 

38 """BaseEngine must build its engine with pre-ping, recycle, and keepalives.""" 

39 captured = {} 

40 

41 def fake_create_engine(dsn, **kwargs): 

42 captured["dsn"] = dsn 

43 captured["kwargs"] = kwargs 

44 return create_engine("postgresql+psycopg://u:p@localhost:5432/db", **kwargs) 

45 

46 monkeypatch.setattr("bbconf.db_utils.create_engine", fake_create_engine) 

47 monkeypatch.setattr(BaseEngine, "create_schema", lambda self, engine=None: None) 

48 monkeypatch.setattr(BaseEngine, "check_db_connection", lambda self: None) 

49 

50 base = BaseEngine(host="localhost", database="bedbase", user="u", password="p") 

51 

52 assert captured["kwargs"]["pool_pre_ping"] is True 

53 assert captured["kwargs"]["pool_recycle"] == POOL_RECYCLE_SECONDS 

54 assert KEEPALIVE_KEYS <= set(captured["kwargs"]["connect_args"]) 

55 

56 # And the settings survive onto the live pool object. 

57 assert base.engine.pool._pre_ping is True 

58 assert base.engine.pool._recycle == POOL_RECYCLE_SECONDS 

59 

60 

61def test_connect_args_reach_the_driver(): 

62 """The libpq keywords are actually handed to psycopg, not silently dropped.""" 

63 engine = create_engine( 

64 "postgresql+psycopg://u:p@localhost:5432/db", 

65 connect_args=dict(CONNECT_ARGS), 

66 ) 

67 

68 seen = {} 

69 

70 class _Stop(Exception): 

71 pass 

72 

73 @event.listens_for(engine, "do_connect") 

74 def _capture(dialect, conn_rec, cargs, cparams): 

75 seen.update(cparams) 

76 raise _Stop() 

77 

78 with pytest.raises(Exception): 

79 engine.connect() 

80 

81 for key, value in CONNECT_ARGS.items(): 

82 assert seen[key] == value