deerflow-code/offline-backend-20260512/backend/tests/test_external_llmwiki_search.py
2026-09-07 18:24:55 +08:00

52 lines
2.3 KiB
Python

from __future__ import annotations
from types import SimpleNamespace
import pytest
from fastapi import HTTPException
from starlette.requests import Request
from app.gateway.auth_middleware import _authorize_external_llmwiki
from app.gateway.routers.external_llmwiki import _external_mappings
def _request(key: str) -> Request:
return Request({"type": "http", "method": "POST", "path": "/api/external/llmwiki/wiki/vector-search", "headers": [(b"x-api-key", key.encode())]})
def test_external_api_key_uses_isolated_fail_closed_auth(monkeypatch) -> None:
config = SimpleNamespace(enabled=True, api_key="current", previous_api_key="previous")
monkeypatch.setattr("deerflow.config.get_app_config", lambda: SimpleNamespace(llmwiki=SimpleNamespace(local_wiki_index=SimpleNamespace(external_api=config))))
assert _authorize_external_llmwiki(_request("current")) is None
assert _authorize_external_llmwiki(_request("previous")) is None
rejected = _authorize_external_llmwiki(_request("wrong"))
assert rejected is not None
assert rejected.status_code == 401
@pytest.mark.asyncio
async def test_external_mappings_require_explicit_public_external_permission() -> None:
rows = [
{"id": "allowed", "publication_status": "published", "external_search_enabled": True, "wiki_index_enabled": True},
{"id": "private", "publication_status": "private", "external_search_enabled": True, "wiki_index_enabled": True},
{"id": "not-opted-in", "publication_status": "published", "external_search_enabled": False, "wiki_index_enabled": True},
{
"id": "deposit",
"name": "对话沉淀",
"owner_user_id": "system",
"publication_status": "published",
"external_search_enabled": True,
"wiki_index_enabled": True,
},
]
class Store:
async def list_visible(self, *_args, **_kwargs):
return rows
request = Request({"type": "http", "method": "POST", "path": "/", "headers": [], "app": SimpleNamespace(state=SimpleNamespace(llmwiki_store=Store()))})
assert [row["id"] for row in await _external_mappings(request, [])] == ["allowed"]
with pytest.raises(HTTPException) as exc_info:
await _external_mappings(request, ["not-opted-in"])
assert exc_info.value.status_code == 404