LegacyHUB/tests/test_api_security.py

"""Tests for the optional API-key auth middleware."""

from __future__ import annotations

import importlib
import os

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)
    yield main_module.app
    # Restore pristine config so the reloaded singleton does not leak to later tests.
    for _k in ("API_KEY", "INGEST_API_KEY", "AUTH_REQUIRE_IDENTITY"):
        os.environ.pop(_k, None)
    cfg.get_settings.cache_clear()
    importlib.reload(cfg)
    importlib.reload(main_module)


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