166 lines
4.6 KiB
Python
166 lines
4.6 KiB
Python
"""Tests for the optional API-key auth middleware."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import importlib
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
|
|
KEY = "test-secret-key-DO-NOT-USE-IN-PROD"
|
|
|
|
|
|
@pytest.fixture
|
|
def secured_app(monkeypatch):
|
|
"""Reload the FastAPI application with API_KEY set so the middleware
|
|
installs itself before the lifespan starts. Returns a TestClient bound to
|
|
that fresh app instance.
|
|
"""
|
|
monkeypatch.setenv("API_KEY", KEY)
|
|
|
|
# Drop cached Settings and main so the new env vars are picked up.
|
|
import app.config as cfg
|
|
import app.main as main_module
|
|
|
|
cfg.get_settings.cache_clear()
|
|
importlib.reload(cfg)
|
|
importlib.reload(main_module)
|
|
return main_module.app
|
|
|
|
|
|
def _patch_health(monkeypatch, module):
|
|
from app.api.schemas import ComponentHealth
|
|
|
|
def _ok(name):
|
|
return ComponentHealth(name=name, status="ok", detail={})
|
|
|
|
for name in (
|
|
"_check_postgres",
|
|
"_check_minio",
|
|
"_check_opensearch",
|
|
"_check_qdrant",
|
|
"_check_redis",
|
|
):
|
|
monkeypatch.setattr(module, name, lambda n=name: _ok(n.removeprefix("_check_")))
|
|
|
|
|
|
def test_health_remains_open_when_key_required(secured_app, monkeypatch):
|
|
from app.api import routes_health
|
|
from app.config import settings
|
|
|
|
_patch_health(monkeypatch, routes_health)
|
|
client = TestClient(secured_app)
|
|
res = client.get(f"{settings.app_api_prefix}/health")
|
|
assert res.status_code == 200
|
|
|
|
|
|
def test_protected_route_rejects_missing_key(secured_app, monkeypatch):
|
|
from app.config import settings
|
|
from app.indexing import hybrid_search
|
|
|
|
monkeypatch.setattr(hybrid_search, "run_search", lambda req: pytest.fail("must not run"))
|
|
|
|
client = TestClient(secured_app)
|
|
res = client.post(
|
|
f"{settings.app_api_prefix}/search",
|
|
json={
|
|
"query": "anything",
|
|
"limit": 1,
|
|
"filters": {
|
|
"document_id": None,
|
|
"source_path": None,
|
|
"block_type": None,
|
|
"min_ocr_confidence": None,
|
|
},
|
|
"search_mode": "hybrid",
|
|
},
|
|
)
|
|
assert res.status_code == 401
|
|
assert res.json()["detail"].startswith("invalid")
|
|
|
|
|
|
def test_protected_route_accepts_x_api_key_header(secured_app, monkeypatch):
|
|
from app.api.schemas import SearchResponse
|
|
from app.config import settings
|
|
from app.indexing import hybrid_search
|
|
|
|
monkeypatch.setattr(
|
|
hybrid_search,
|
|
"run_search",
|
|
lambda req: SearchResponse(
|
|
query=req.query, mode=req.search_mode, total_candidates=0, reranked=False, results=[]
|
|
),
|
|
)
|
|
|
|
client = TestClient(secured_app)
|
|
res = client.post(
|
|
f"{settings.app_api_prefix}/search",
|
|
headers={"X-API-Key": KEY},
|
|
json={
|
|
"query": "x",
|
|
"limit": 1,
|
|
"filters": {
|
|
"document_id": None,
|
|
"source_path": None,
|
|
"block_type": None,
|
|
"min_ocr_confidence": None,
|
|
},
|
|
"search_mode": "hybrid",
|
|
},
|
|
)
|
|
assert res.status_code == 200
|
|
|
|
|
|
def test_protected_route_accepts_bearer_token(secured_app, monkeypatch):
|
|
from app.api.schemas import SearchResponse
|
|
from app.config import settings
|
|
from app.indexing import hybrid_search
|
|
|
|
monkeypatch.setattr(
|
|
hybrid_search,
|
|
"run_search",
|
|
lambda req: SearchResponse(
|
|
query=req.query, mode=req.search_mode, total_candidates=0, reranked=False, results=[]
|
|
),
|
|
)
|
|
|
|
client = TestClient(secured_app)
|
|
res = client.post(
|
|
f"{settings.app_api_prefix}/search",
|
|
headers={"Authorization": f"Bearer {KEY}"},
|
|
json={
|
|
"query": "x",
|
|
"limit": 1,
|
|
"filters": {
|
|
"document_id": None,
|
|
"source_path": None,
|
|
"block_type": None,
|
|
"min_ocr_confidence": None,
|
|
},
|
|
"search_mode": "hybrid",
|
|
},
|
|
)
|
|
assert res.status_code == 200
|
|
|
|
|
|
def test_protected_route_rejects_wrong_key(secured_app):
|
|
from app.config import settings
|
|
|
|
client = TestClient(secured_app)
|
|
res = client.post(
|
|
f"{settings.app_api_prefix}/search",
|
|
headers={"X-API-Key": "wrong"},
|
|
json={
|
|
"query": "x",
|
|
"limit": 1,
|
|
"filters": {
|
|
"document_id": None,
|
|
"source_path": None,
|
|
"block_type": None,
|
|
"min_ocr_confidence": None,
|
|
},
|
|
"search_mode": "hybrid",
|
|
},
|
|
)
|
|
assert res.status_code == 401
|