Unable to code a pytest case to check my FastAPI API key authorizaton

Question

The following tests fail under pytest since the response is always 200:

# Create a test client
client = TestClient(app)

# Mock Redis client
mock_redis = AsyncMock()

@pytest.fixture(scope="function", autouse=True)
def mock_redis_client():
    with patch("mema4.server.REDIS_CLIENT", mock_redis):
        mock_redis.lrange.return_value = []  # Ensure lrange returns an empty list
        yield mock_redis

        
@pytest.fixture(scope="function", autouse=True)
def override_get_api_key():
    def mock_get_api_key(api_key: str = Security(api_key_header)):
        print(f"Received API key: {api_key}")  # Debugging line
        print(f"Expected API key: {MEMA4_API_KEY}")  # Debugging line
        if api_key == MEMA4_API_KEY:
            return api_key
        raise HTTPException(status_code=403, detail="Invalid API Key")

    app.dependency_overrides[get_api_key] = mock_get_api_key
    try:
        yield
    finally:
        del app.dependency_overrides[get_api_key]


@pytest.mark.parametrize("api_key,expected_status", [
    (str(MEMA4_API_KEY), 200),
    ("invalid_key", 403),
    (None, 403)
])
def test_api_key_scenarios(mock_redis_client, api_key, expected_status):
    mock_redis_client.lrange.return_value = []
    headers = {"X-API-Key": api_key} if api_key is not None else {}
    response = client.get("/queueFetch", headers=headers)
    print(f"Response status code: {response.status_code}")  # Debugging line
    print(f"Response content: {response.content}")  # Debugging line
    assert response.status_code == expected_status
    if expected_status == 403:
        assert response.json()["detail"] == "Invalid API Key"

on the contrary the real endpoint that is tested /fetchQueue behaves as desired with a 200 with the good API key and a 403 in case it is invalid or missing:

# Security setup
api_key_header = APIKeyHeader(name="X-API-Key", auto_error=True)
......
@app.get(
    "/queueFetch",
    response_model=QueueItemsResponse,
    responses={
        200: {"description": "Successfully retrieved queue items"},
        403: {"description": "Invalid API Key"},
        503: {"description": "Service unavailable, queue is down"},
    },
    dependencies=[Security(api_key_header)],
)

I am totally out of ideas. Thanks for any help.

Answer 1

You are overriding get_api_key dependency, but do you use it in your endpoint function? In the part of code you shared you are using only api_key_header.

I created runnable code example (simplified your code and added endpoint definition) and it works:

from typing import Annotated
from unittest.mock import AsyncMock, patch
from fastapi import Depends, FastAPI, HTTPException, Security
from fastapi.security import APIKeyHeader
from fastapi.testclient import TestClient
import pytest

api_key_header = APIKeyHeader(name="X-API-Key", auto_error=True)

app = FastAPI()

def get_api_key(api_key: str = Security(api_key_header)):
    return api_key

@app.get(
    "/queueFetch",
    responses={
        200: {"description": "Successfully retrieved queue items"},
        403: {"description": "Invalid API Key"},
        503: {"description": "Service unavailable, queue is down"},
    },
    dependencies=[Security(api_key_header)],
)
def asd(api_key: Annotated[str, Depends(get_api_key)]):
    pass

client = TestClient(app)

MEMA4_API_KEY = "123"

@pytest.fixture(scope="function", autouse=True)
def override_get_api_key():
    def mock_get_api_key(api_key: str = Security(api_key_header)):
        print(f"Received API key: {api_key}")  # Debugging line
        print(f"Expected API key: {MEMA4_API_KEY}")  # Debugging line
        if api_key == MEMA4_API_KEY:
            return api_key
        raise HTTPException(status_code=403, detail="Invalid API Key")

    app.dependency_overrides[get_api_key] = mock_get_api_key
    try:
        yield
    finally:
        del app.dependency_overrides[get_api_key]


@pytest.mark.parametrize("api_key,expected_status", [
    (str(MEMA4_API_KEY), 200),
    ("invalid_key", 403),
    (None, 403)
])
def test_api_key_scenarios(api_key, expected_status):
    headers = {"X-API-Key": api_key} if api_key is not None else {}
    response = client.get("/queueFetch", headers=headers)
    print(f"Response status code: {response.status_code}")  # Debugging line
    print(f"Response content: {response.content}")  # Debugging line
    assert response.status_code == expected_status
    if expected_status == 403:
        assert response.json()["detail"] in ("Invalid API Key", "Not authenticated")