bambuddy/backend/tests/unit/test_db_pool_and_auth_cache.py

"""Tests for the DB connection-pool sizing/diagnostics and the auth-enabled
cache added for large printer farms (issue #2572)."""

import asyncio

import pytest
from sqlalchemy import text
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine


class TestPoolConfiguration:
    """P0: env-configurable, dialect-aware pool sizing."""

    def test_sqlite_defaults_when_unset(self, monkeypatch):
        """SQLite keeps 20 + 200 when no env override is set."""
        from backend.app.core import database

        for attr in ("db_pool_size", "db_max_overflow", "db_pool_timeout", "db_pool_recycle"):
            monkeypatch.setattr(database.settings, attr, None, raising=False)
        monkeypatch.setattr(database, "is_sqlite", lambda: True)

        kwargs = database._resolve_pool_kwargs()
        assert kwargs["pool_size"] == 20
        assert kwargs["max_overflow"] == 200
        # No server-socket recycle/pre-ping for a local file.
        assert "pool_pre_ping" not in kwargs
        assert "pool_recycle" not in kwargs

    def test_postgres_defaults_raise_the_old_limits(self, monkeypatch):
        """Postgres default is now 20 + 80 (was 10 + 20) with pre-ping + recycle."""
        from backend.app.core import database

        for attr in ("db_pool_size", "db_max_overflow", "db_pool_timeout", "db_pool_recycle"):
            monkeypatch.setattr(database.settings, attr, None, raising=False)
        monkeypatch.setattr(database, "is_sqlite", lambda: False)

        kwargs = database._resolve_pool_kwargs()
        assert kwargs["pool_size"] == 20
        assert kwargs["max_overflow"] == 80
        assert kwargs["pool_pre_ping"] is True
        assert kwargs["pool_recycle"] == 1800

    def test_env_overrides_win_on_postgres(self, monkeypatch):
        """DB_POOL_* overrides replace the dialect defaults."""
        from backend.app.core import database

        monkeypatch.setattr(database.settings, "db_pool_size", 100, raising=False)
        monkeypatch.setattr(database.settings, "db_max_overflow", 200, raising=False)
        monkeypatch.setattr(database.settings, "db_pool_timeout", 45, raising=False)
        monkeypatch.setattr(database.settings, "db_pool_recycle", 600, raising=False)
        monkeypatch.setattr(database, "is_sqlite", lambda: False)

        kwargs = database._resolve_pool_kwargs()
        assert kwargs["pool_size"] == 100
        assert kwargs["max_overflow"] == 200
        assert kwargs["pool_timeout"] == 45
        assert kwargs["pool_recycle"] == 600

    @pytest.mark.asyncio
    async def test_concurrent_checkouts_exceed_base_pool_size(self, tmp_path):
        """Regression (#2572): more concurrent sessions than pool_size must all
        complete by drawing from max_overflow — not deadlock or time out.

        This is the failure the farm hit: printer callbacks held every base
        connection, so unrelated requests waited on the pool. With headroom in
        max_overflow, concurrent checkouts beyond pool_size still succeed.

        Uses a file-based SQLite URL so it gets a real queue pool — the
        in-memory URL forces a single-connection StaticPool that ignores
        pool_size/max_overflow entirely.
        """
        db_file = tmp_path / "pool_regression.db"
        eng = create_async_engine(f"sqlite+aiosqlite:///{db_file}", pool_size=2, max_overflow=10)
        sm = async_sessionmaker(eng)

        async def _one():
            async with sm() as s:
                await s.execute(text("SELECT 1"))
                # Hold the checkout briefly so the calls genuinely overlap and
                # force the pool past its base size of 2.
                await asyncio.sleep(0.05)
                return (await s.execute(text("SELECT 1"))).scalar()

        try:
            results = await asyncio.gather(*[_one() for _ in range(12)])
        finally:
            await eng.dispose()

        assert results == [1] * 12


class TestPoolStatus:
    """P3: diagnostics snapshot."""

    def test_get_pool_status_shape(self):
        from backend.app.core.database import get_pool_status

        status = get_pool_status()
        assert status["dialect"] in ("sqlite", "postgresql")
        for key in ("pool_size", "max_overflow", "pool_timeout", "pool_recycle", "pool_pre_ping"):
            assert key in status["config"]
        # Live gauges are present (values are ints on a QueuePool).
        for key in ("current_size", "checked_out", "checked_in", "overflow"):
            assert key in status


class TestAuthEnabledCache:
    """P1: cache the auth-enabled probe, but only ever cache True."""

    class _Setting:
        def __init__(self, value):
            self.value = value

    class _Result:
        def __init__(self, setting):
            self._setting = setting

        def scalar_one_or_none(self):
            return self._setting

    class _CountingDB:
        def __init__(self, value):
            self._value = value
            self.calls = 0

        async def execute(self, *args, **kwargs):
            self.calls += 1
            setting = None if self._value is None else TestAuthEnabledCache._Setting(self._value)
            return TestAuthEnabledCache._Result(setting)

    @pytest.mark.asyncio
    async def test_enabled_true_is_cached(self):
        from backend.app.core import auth as auth_mod

        auth_mod.invalidate_auth_enabled_cache()
        db = self._CountingDB("true")

        assert await auth_mod.is_auth_enabled(db) is True
        assert db.calls == 1
        # Second probe served from cache — no new query.
        assert await auth_mod.is_auth_enabled(db) is True
        assert db.calls == 1

        # Invalidation forces a re-read (e.g. after set_auth_enabled).
        auth_mod.invalidate_auth_enabled_cache()
        assert await auth_mod.is_auth_enabled(db) is True
        assert db.calls == 2
        auth_mod.invalidate_auth_enabled_cache()

    @pytest.mark.asyncio
    async def test_disabled_is_never_cached(self):
        """SECURITY: a disabled result must never be cached, so staleness can
        only ever fail closed (require auth), never open."""
        from backend.app.core import auth as auth_mod

        auth_mod.invalidate_auth_enabled_cache()
        db = self._CountingDB("false")

        assert await auth_mod.is_auth_enabled(db) is False
        assert db.calls == 1
        # Every probe re-reads while disabled.
        assert await auth_mod.is_auth_enabled(db) is False
        assert db.calls == 2
        auth_mod.invalidate_auth_enabled_cache()

    @pytest.mark.asyncio
    async def test_unconfigured_returns_false_and_is_not_cached(self):
        from backend.app.core import auth as auth_mod

        auth_mod.invalidate_auth_enabled_cache()
        db = self._CountingDB(None)

        assert await auth_mod.is_auth_enabled(db) is False
        assert await auth_mod.is_auth_enabled(db) is False
        assert db.calls == 2
        auth_mod.invalidate_auth_enabled_cache()

    @pytest.mark.asyncio
    async def test_db_error_propagates_fail_closed(self):
        """A probe error must propagate (fail closed), not be swallowed."""
        from backend.app.core import auth as auth_mod

        auth_mod.invalidate_auth_enabled_cache()

        class _RaisingDB:
            async def execute(self, *args, **kwargs):
                raise RuntimeError("connection lost")

        with pytest.raises(RuntimeError):
            await auth_mod.is_auth_enabled(_RaisingDB())
        auth_mod.invalidate_auth_enabled_cache()