"""0.9.6: the pooled engine — connections are reused, pinged only after idle, and a dead one is replaced.
The live half needs a real Postgres (ERP_TEST_PG_URL=postgresql+psycopg://user@host:port/db)."""
import os
from urllib.parse import urlparse
import pytest
from sqlalchemy import create_engine, event, text
from erp_platform.pool import pooled_engine, watch_idle
PG_URL = os.environ.get("ERP_TEST_PG_URL", "")
def _pings(engine):
seen = []
orig = engine.dialect.do_ping
engine.dialect.do_ping = lambda c: (seen.append(1), orig(c))[1]
return seen
def test_a_connection_is_pinged_only_after_it_sat_idle():
e = create_engine("sqlite://")
watch_idle(e, ping_after=3600)
pings = _pings(e)
for _ in range(3):
with e.connect() as c:
c.execute(text("SELECT 1"))
assert pings == [] # hot: no extra round trip per checkout
e2 = create_engine("sqlite://")
watch_idle(e2, ping_after=0)
pings2 = _pings(e2)
for _ in range(3):
with e2.connect() as c:
c.execute(text("SELECT 1"))
assert len(pings2) == 2 # every checkout after the first
@pytest.mark.skipif(not PG_URL, reason="needs ERP_TEST_PG_URL (a real Postgres)")
def test_a_pooled_connection_is_reused_and_a_dead_one_replaced():
import psycopg
u = urlparse(PG_URL.replace("postgresql+psycopg", "postgresql"))
kw = dict(host=u.hostname, port=u.port or 5432, dbname=u.path.lstrip("/"), user=u.username or "postgres")
made = []
def creator():
made.append(1) # where a fresh token would be resolved
return psycopg.connect(**kw)
e = pooled_engine(creator, pool_size=2, ping_after=0)
with e.connect() as c:
pid = c.execute(text("SELECT pg_backend_pid()")).scalar()
with e.connect() as c:
assert c.execute(text("SELECT pg_backend_pid()")).scalar() == pid and len(made) == 1
with psycopg.connect(**kw, autocommit=True) as killer:
killer.execute("SELECT pg_terminate_backend(%s)", (pid,))
with e.connect() as c: # the ping finds it dead: a new one, no error
assert c.execute(text("SELECT pg_backend_pid()")).scalar() != pid
assert len(made) == 2
e.dispose()
def test_the_pg_wire_engine_never_probes_for_hstore(monkeypatch):
"""0.9.11: psycopg's hstore lookup is a pg_catalog query, O(tables) on DataK3 (dodilio/tables#494)."""
monkeypatch.setenv("BUCKET", "not-used")
from erp_platform import db
from erp_platform.pool import pooled_engine
assert pooled_engine(lambda: None, pool_size=1).dialect.use_native_hstore is False
assert db.engine.dialect.use_native_hstore is False