tests/test_deps.py

import importlib
from unittest.mock import MagicMock, patch

import pytest
from fastapi import HTTPException

import app.api.deps as deps
from app.api.deps import get_db


def test_get_db_closes_session():
    mock_session = MagicMock()
    with patch('app.api.deps.SessionLocal', return_value=mock_session):
        db_gen = get_db()
        db = next(db_gen)
        assert db == mock_session
        # Simulate an exception during request handling
        with pytest.raises(Exception):
            db_gen.throw(Exception("test error"))
        mock_session.close.assert_called_once()


# verify_internal_key tests below are called directly, not through
# TestClient: tests/conftest.py globally overrides verify_internal_key
# (`fastapi_app.dependency_overrides[verify_internal_key] = mock_verify_internal_key`)
# so that already-protected routes (routes_governance.py) can be exercised
# in tests without the gateway-injected X-Internal-Key header. That override
# makes the real fail-closed behavior untestable through the app for any
# router that uses it -- this is the only place it's actually verified to
# reject what it should reject, rather than just trusted to work because
# it's wired in.
#
# Newly relevant as of the auth fix to routes_risk.py, routes_intents.py,
# routes_history.py, routes_memory.py (see docs/authentication.md) -- those
# four routers now depend on this function passing correctly.

_NEWLY_PROTECTED_ROUTER_MODULES = [
    "app.api.routes_risk",
    "app.api.routes_intents",
    "app.api.routes_history",
    "app.api.routes_memory",
]


@pytest.mark.asyncio
async def test_verify_internal_key_rejects_missing_header(monkeypatch):
    monkeypatch.setattr(deps, "INTERNAL_API_KEY", "real-secret")
    with pytest.raises(HTTPException) as exc_info:
        await deps.verify_internal_key(x_internal_key=None)
    assert exc_info.value.status_code == 401


@pytest.mark.asyncio
async def test_verify_internal_key_rejects_wrong_key(monkeypatch):
    monkeypatch.setattr(deps, "INTERNAL_API_KEY", "real-secret")
    with pytest.raises(HTTPException) as exc_info:
        await deps.verify_internal_key(x_internal_key="wrong-key")
    assert exc_info.value.status_code == 401


@pytest.mark.asyncio
async def test_verify_internal_key_fails_closed_when_unset(monkeypatch):
    """The env var being unset must reject every request, not let them
    through -- this is the specific property that makes this safe to add
    to a router without also needing to guarantee the env var is always
    set."""
    monkeypatch.setattr(deps, "INTERNAL_API_KEY", "")
    with pytest.raises(HTTPException) as exc_info:
        await deps.verify_internal_key(x_internal_key="anything")
    assert exc_info.value.status_code == 401


@pytest.mark.asyncio
async def test_verify_internal_key_accepts_correct_key(monkeypatch):
    monkeypatch.setattr(deps, "INTERNAL_API_KEY", "real-secret")
    result = await deps.verify_internal_key(x_internal_key="real-secret")
    assert result is None


@pytest.mark.parametrize("router_module_name", _NEWLY_PROTECTED_ROUTER_MODULES)
def test_router_requires_verify_internal_key(router_module_name):
    """Structural check, independent of the function-level tests above:
    proves each router actually declares verify_internal_key as a
    router-level dependency, not just that the function itself works in
    isolation. Mirrors the pattern routes_governance.py already uses
    (`APIRouter(dependencies=[Depends(verify_internal_key)])`)."""
    module = importlib.import_module(router_module_name)
    dependency_callables = [d.dependency for d in module.router.dependencies]
    assert deps.verify_internal_key in dependency_callables


# ---------------------------------------------------------------------------
# Rate limit resolution
# ---------------------------------------------------------------------------

@pytest.mark.parametrize(
    "configured",
    ["", "   ", "garbage", "100/minutes/extra", None],
    ids=["blank", "whitespace", "unparseable", "malformed", "unset"],
)
def test_bad_rate_limit_degrades_to_the_default(configured):
    """A bad RATE_LIMIT must not be able to take the service down.

    Limiter accepts any string and only parses inside the middleware, per
    request -- so an unparseable value boots clean, reports healthy, and
    then returns 500 on every route including /health. Blank is the easy
    way in: clearing the variable in a dashboard overrides the default
    with "" rather than restoring it.
    """
    from app.api.deps import _RATE_LIMIT_FALLBACK, _resolve_rate_limit

    assert _resolve_rate_limit(configured) == _RATE_LIMIT_FALLBACK


@pytest.mark.parametrize("configured", ["50/minute", "100 per minute", "10/second"])
def test_valid_rate_limit_is_passed_through_unchanged(configured):
    from app.api.deps import _resolve_rate_limit

    assert _resolve_rate_limit(configured) == configured


@pytest.mark.parametrize(
    "configured", ["", "   ", "garbage", None, "50/minute", "100 per minute"]
)
def test_resolved_rate_limit_always_parses_per_request(configured):
    """The property that actually matters: whatever comes back must survive
    the per-request parse SlowAPIMiddleware performs, for every input."""
    from slowapi import Limiter
    from slowapi.util import get_remote_address

    from app.api.deps import _resolve_rate_limit

    limiter = Limiter(
        key_func=get_remote_address, default_limits=[_resolve_rate_limit(configured)]
    )
    for group in limiter._default_limits:
        assert list(group)


def test_module_limiter_parses_per_request():
    """The real module-level limiter, not a rebuilt one -- this is the
    object SlowAPIMiddleware actually uses."""
    from app.api.deps import limiter

    for group in limiter._default_limits:
        assert list(group)