337 lines
11 KiB
Python
337 lines
11 KiB
Python
from __future__ import annotations
|
|
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
from fastapi import HTTPException
|
|
from starlette.requests import Request
|
|
|
|
from app.gateway.routers import llmwiki
|
|
from app.gateway.weknora_embed import verify_weknora_signed_payload
|
|
|
|
|
|
def _request(store: object) -> Request:
|
|
return Request(
|
|
{
|
|
"type": "http",
|
|
"method": "GET",
|
|
"path": "/api/public/llmwiki/knowledge-bases/kb-one",
|
|
"headers": [],
|
|
"app": SimpleNamespace(state=SimpleNamespace(llmwiki_store=store)),
|
|
}
|
|
)
|
|
|
|
|
|
class PublicStore:
|
|
def __init__(self, row: dict[str, object] | None) -> None:
|
|
self.row = row
|
|
self.calls: list[tuple[str, str, bool, bool]] = []
|
|
|
|
async def get_authorized(
|
|
self,
|
|
mapping_id: str,
|
|
user_id: str,
|
|
*,
|
|
write: bool,
|
|
is_admin: bool,
|
|
) -> dict[str, object] | None:
|
|
self.calls.append((mapping_id, user_id, write, is_admin))
|
|
return self.row
|
|
|
|
|
|
class PublicListStore:
|
|
def __init__(self, rows: list[dict[str, object]]) -> None:
|
|
self.rows = rows
|
|
self.calls: list[tuple[str, str, bool]] = []
|
|
|
|
async def list_visible(
|
|
self,
|
|
user_id: str,
|
|
*,
|
|
scope: str,
|
|
is_admin: bool,
|
|
) -> list[dict[str, object]]:
|
|
self.calls.append((user_id, scope, is_admin))
|
|
return self.rows
|
|
|
|
|
|
def _published_row() -> dict[str, object]:
|
|
return {
|
|
"id": "kb-one",
|
|
"weknora_id": "remote-one",
|
|
"owner_user_id": "owner-one",
|
|
"name": "公开知识库",
|
|
"description": "公开说明",
|
|
"kb_type": "document",
|
|
"publication_status": "published",
|
|
}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_public_knowledge_base_list_exposes_only_published_wiki_identity() -> None:
|
|
store = PublicListStore(
|
|
[
|
|
{
|
|
**_published_row(),
|
|
"id": "1063edee-332a-43b4-bbd9-73bf0a708b94",
|
|
"owner_user_id": "secret-owner",
|
|
},
|
|
{**_published_row(), "id": "private-kb", "publication_status": "private"},
|
|
{
|
|
**_published_row(),
|
|
"id": "conversation-deposit",
|
|
"owner_user_id": "system",
|
|
"name": "对话沉淀",
|
|
},
|
|
]
|
|
)
|
|
|
|
payload = await llmwiki.list_public_knowledge_bases(_request(store))
|
|
|
|
assert store.calls == [("__public_wiki__", "public", False)]
|
|
assert payload == {
|
|
"knowledge_bases": [
|
|
{
|
|
"id": "1063edee-332a-43b4-bbd9-73bf0a708b94",
|
|
"name": "公开知识库",
|
|
}
|
|
],
|
|
"total": 1,
|
|
}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_public_wiki_reader_uses_anonymous_non_admin_authorization(monkeypatch) -> None:
|
|
store = PublicStore(_published_row())
|
|
|
|
class Client:
|
|
async def get_knowledge_base(self, knowledge_base_id: str):
|
|
assert knowledge_base_id == "remote-one"
|
|
return {"id": knowledge_base_id, "name": "公开知识库", "knowledge_count": 2}
|
|
|
|
monkeypatch.setattr(llmwiki, "_client_or_503", lambda: Client())
|
|
|
|
payload = await llmwiki.get_public_knowledge_base(_request(store), "kb-one")
|
|
|
|
assert store.calls == [("kb-one", "__public_wiki__", False, False)]
|
|
assert payload["publication_status"] == "published"
|
|
assert payload["owner_user_id"] is None
|
|
assert payload["can_write"] is False
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize(
|
|
"row",
|
|
[
|
|
{**_published_row(), "publication_status": "private"},
|
|
{
|
|
**_published_row(),
|
|
"owner_user_id": "system",
|
|
"name": "对话沉淀",
|
|
},
|
|
None,
|
|
],
|
|
)
|
|
async def test_public_wiki_reader_rejects_non_public_resources(row) -> None:
|
|
with pytest.raises(HTTPException) as exc_info:
|
|
await llmwiki._public_mapping(_request(PublicStore(row)), "kb-one")
|
|
|
|
assert exc_info.value.status_code == 404
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_public_wiki_page_list_is_normalized(monkeypatch) -> None:
|
|
store = PublicStore(_published_row())
|
|
|
|
class Client:
|
|
async def list_wiki_pages(self, knowledge_base_id: str, **kwargs):
|
|
assert knowledge_base_id == "remote-one"
|
|
assert kwargs == {"page": 1, "page_size": 50, "query": "目录"}
|
|
return {
|
|
"pages": [
|
|
{
|
|
"id": "page-one",
|
|
"slug": "guide/start",
|
|
"title": "开始使用",
|
|
"page_type": "article",
|
|
"status": "completed",
|
|
"content": "# 开始使用",
|
|
}
|
|
],
|
|
"total": 1,
|
|
"page": 1,
|
|
"page_size": 50,
|
|
"total_pages": 1,
|
|
}
|
|
|
|
monkeypatch.setattr(llmwiki, "_client_or_503", lambda: Client())
|
|
|
|
payload = await llmwiki.list_public_wiki_pages(
|
|
_request(store),
|
|
"kb-one",
|
|
page=1,
|
|
page_size=50,
|
|
query=" 目录 ",
|
|
)
|
|
|
|
assert payload["pages"][0]["slug"] == "guide/start"
|
|
assert payload["pages"][0]["aliases"] == []
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_public_wiki_file_is_authorized_and_streamed(monkeypatch) -> None:
|
|
store = PublicStore(_published_row())
|
|
|
|
class Client:
|
|
async def get_knowledge_base_file(self, knowledge_base_id: str, file_path: str):
|
|
assert knowledge_base_id == "remote-one"
|
|
assert file_path == "storage://tenant/resource://wiki/figure.png"
|
|
return b"png-bytes", "image/png"
|
|
|
|
monkeypatch.setattr(llmwiki, "_client_or_503", lambda: Client())
|
|
|
|
response = await llmwiki.get_public_wiki_file(
|
|
_request(store),
|
|
"kb-one",
|
|
"storage://tenant/resource://wiki/figure.png",
|
|
)
|
|
|
|
assert store.calls == [("kb-one", "__public_wiki__", False, False)]
|
|
assert response.body == b"png-bytes"
|
|
assert response.media_type == "image/png"
|
|
assert response.headers["cache-control"] == "public, max-age=300"
|
|
assert response.headers["content-disposition"] == "inline; filename*=UTF-8''figure.png"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_private_wiki_file_uses_signed_http_access_instead_of_blob_url(monkeypatch) -> None:
|
|
async def authorized_mapping(*_args, **_kwargs):
|
|
return _published_row(), "user-one", False
|
|
|
|
class Client:
|
|
async def get_knowledge_base_file(self, knowledge_base_id: str, file_path: str):
|
|
assert knowledge_base_id == "remote-one"
|
|
assert file_path == "storage://tenant/resource://wiki/figure.png"
|
|
return b"png-bytes", "image/png"
|
|
|
|
monkeypatch.setattr(llmwiki, "_authorized_mapping", authorized_mapping)
|
|
monkeypatch.setattr(llmwiki, "_client_or_503", lambda: Client())
|
|
|
|
access = await llmwiki.create_wiki_file_access(
|
|
_request(PublicStore(_published_row())),
|
|
"kb-one",
|
|
"storage://tenant/resource://wiki/figure.png",
|
|
)
|
|
payload = verify_weknora_signed_payload(access["token"])
|
|
assert payload["kind"] == "wiki_file"
|
|
assert payload["mapping_id"] == "kb-one"
|
|
assert payload["weknora_id"] == "remote-one"
|
|
assert payload["file_path"] == "storage://tenant/resource://wiki/figure.png"
|
|
|
|
response = await llmwiki.get_signed_wiki_file(access["token"])
|
|
assert response.body == b"png-bytes"
|
|
assert response.headers["cache-control"] == "private, max-age=3600"
|
|
assert response.headers["content-disposition"] == "inline; filename*=UTF-8''figure.png"
|
|
|
|
|
|
def test_wiki_file_proxy_rejects_non_weknora_urls() -> None:
|
|
with pytest.raises(HTTPException) as exc_info:
|
|
llmwiki._validate_protected_wiki_file_path("https://example.test/private.png")
|
|
|
|
assert exc_info.value.status_code == 422
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_wiki_file_proxy_adds_image_extension_for_opaque_resource_ids(monkeypatch) -> None:
|
|
class Client:
|
|
async def get_knowledge_base_file(self, knowledge_base_id: str, file_path: str):
|
|
return b"png-bytes", "image/png"
|
|
|
|
monkeypatch.setattr(llmwiki, "_client_or_503", lambda: Client())
|
|
|
|
response = await llmwiki._wiki_file_response(
|
|
{"weknora_id": "remote-one"},
|
|
"resource://opaque-image-id",
|
|
public=True,
|
|
)
|
|
|
|
assert response.headers["content-disposition"] == "inline; filename*=UTF-8''opaque-image-id.png"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_public_wiki_index_preserves_weknora_group_and_item_order(monkeypatch) -> None:
|
|
store = PublicStore(_published_row())
|
|
calls: list[tuple[tuple[str, ...], str]] = []
|
|
|
|
class Client:
|
|
async def get_wiki_index(
|
|
self,
|
|
knowledge_base_id: str,
|
|
*,
|
|
types: list[str],
|
|
limit: int,
|
|
cursor: str,
|
|
):
|
|
assert knowledge_base_id == "remote-one"
|
|
assert limit == 500
|
|
page_type = types[0]
|
|
calls.append((tuple(types), cursor))
|
|
if page_type == "concept" and not cursor:
|
|
return {
|
|
"intro": "# Wiki Index\n\n公开目录",
|
|
"version": 3,
|
|
"groups": [
|
|
{
|
|
"type": "concept",
|
|
"total": 2,
|
|
"next_cursor": "concept-next",
|
|
"items": [
|
|
{
|
|
"slug": "concept/first",
|
|
"title": "第一项",
|
|
"summary": "第一项摘要",
|
|
"category_path": ["流程图", "规范"],
|
|
}
|
|
],
|
|
}
|
|
],
|
|
}
|
|
if page_type == "concept":
|
|
return {
|
|
"intro": "# Wiki Index\n\n公开目录",
|
|
"version": 3,
|
|
"groups": [
|
|
{
|
|
"type": "concept",
|
|
"total": 2,
|
|
"items": [
|
|
{
|
|
"slug": "concept/second",
|
|
"title": "第二项",
|
|
"summary": "第二项摘要",
|
|
"category_path": ["流程图", "规范"],
|
|
}
|
|
],
|
|
}
|
|
],
|
|
}
|
|
return {
|
|
"intro": "# Wiki Index\n\n公开目录",
|
|
"version": 3,
|
|
"groups": [{"type": page_type, "total": 0, "items": []}],
|
|
}
|
|
|
|
monkeypatch.setattr(llmwiki, "_client_or_503", lambda: Client())
|
|
|
|
payload = await llmwiki.get_public_wiki_index(_request(store), "kb-one")
|
|
|
|
concept = next(group for group in payload["groups"] if group["type"] == "concept")
|
|
assert payload["intro"] == "# Wiki Index\n\n公开目录"
|
|
assert payload["version"] == 3
|
|
assert [item["slug"] for item in concept["items"]] == [
|
|
"concept/first",
|
|
"concept/second",
|
|
]
|
|
assert concept["total"] == 2
|
|
assert (("concept",), "concept-next") in calls
|