LegacyHUB/tests/test_ingest_auth.py

"""Tests for mandatory service-key auth on the ingest endpoints."""

from __future__ import annotations

import importlib
import os

import pytest
from fastapi import HTTPException
from fastapi.testclient import TestClient

from app.api.security import require_service_api_key


class _Req:
    def __init__(self, headers: dict[str, str]) -> None:
        self.headers = headers


def test_open_when_no_key_configured(monkeypatch):
    from app.config import settings

    monkeypatch.setattr(settings, "ingest_api_key", "")
    monkeypatch.setattr(settings, "api_key", "")
    require_service_api_key(_Req({}))  # must not raise


def test_rejects_missing_key(monkeypatch):
    from app.config import settings

    monkeypatch.setattr(settings, "ingest_api_key", "svc-key")
    monkeypatch.setattr(settings, "api_key", "")
    with pytest.raises(HTTPException) as exc:
        require_service_api_key(_Req({}))
    assert exc.value.status_code == 401


def test_accepts_valid_x_api_key(monkeypatch):
    from app.config import settings

    monkeypatch.setattr(settings, "ingest_api_key", "svc-key")
    monkeypatch.setattr(settings, "api_key", "")
    require_service_api_key(_Req({"x-api-key": "svc-key"}))  # must not raise


def test_falls_back_to_global_api_key_via_bearer(monkeypatch):
    from app.config import settings

    monkeypatch.setattr(settings, "ingest_api_key", "")
    monkeypatch.setattr(settings, "api_key", "glob-key")
    require_service_api_key(_Req({"authorization": "Bearer glob-key"}))  # must not raise


def test_rejects_wrong_key(monkeypatch):
    from app.config import settings

    monkeypatch.setattr(settings, "ingest_api_key", "svc-key")
    monkeypatch.setattr(settings, "api_key", "")
    with pytest.raises(HTTPException):
        require_service_api_key(_Req({"x-api-key": "nope"}))


# ---- integration ----

_MANIFEST = {
    "manifest": {
        "manifest_version": "1.0",
        "asset": {
            "asset_id": "asset_x",
            "owner_module": "qms",
            "owner_record_type": "doc",
            "owner_record_id": "1",
            "asset_kind": "document",
            "content_type": "application/pdf",
            "size_bytes": 10,
            "sha256": "a" * 64,
        },
        "storage": {
            "provider": "minio",
            "bucket": "teamhub-qms-originals",
            "object_key": "qms/2026/01/01/asset_x/original/f.pdf",
        },
        # pending gate -> handler short-circuits before any storage access.
        "security": {"gate_status": "pending"},
        "retention": {"policy_id": "default", "legal_hold": False},
    }
}


@pytest.fixture
def ingest_secured_app(monkeypatch):
    monkeypatch.setenv("INGEST_API_KEY", "svc-key")
    monkeypatch.delenv("API_KEY", raising=False)
    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
    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 test_knowledge_ingest_rejects_without_service_key(ingest_secured_app):
    from app.config import settings

    client = TestClient(ingest_secured_app)
    res = client.post(f"{settings.app_api_prefix}/knowledge-ingest", json=_MANIFEST)
    assert res.status_code == 401


def test_knowledge_ingest_accepts_with_service_key(ingest_secured_app):
    from app.config import settings

    client = TestClient(ingest_secured_app)
    res = client.post(
        f"{settings.app_api_prefix}/knowledge-ingest",
        headers={"X-API-Key": "svc-key"},
        json=_MANIFEST,
    )
    # Auth passed; the pending gate makes the handler return a clean rejection.
    assert res.status_code == 200
    assert res.json()["status"] == "rejected"
    assert res.json()["reason_code"] == "security_gate_not_approved"