461 lines
16 KiB
Python
461 lines
16 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_live_safe_metadata_and_generated_index(monkeypatch) -> 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": "对话沉淀",
|
|
},
|
|
{
|
|
**_published_row(),
|
|
"id": "stale-kb",
|
|
"weknora_id": "remote-missing",
|
|
},
|
|
]
|
|
)
|
|
|
|
class Client:
|
|
async def list_knowledge_bases(self):
|
|
return [
|
|
{
|
|
"id": "remote-one",
|
|
"name": "远端公开知识库",
|
|
"description": "远端公开说明",
|
|
"type": "wiki",
|
|
"knowledge_count": 12,
|
|
"chunk_count": 34,
|
|
"processing_count": 1,
|
|
"share_count": 3,
|
|
"created_at": "2026-07-01T08:00:00+08:00",
|
|
"updated_at": "2026-08-01T08:00:00+08:00",
|
|
"tenant_id": 10001,
|
|
"creator_id": "secret-creator",
|
|
"storage_config": {"secret_key": "must-not-leak"},
|
|
}
|
|
]
|
|
|
|
async def get_knowledge_base(self, knowledge_base_id: str):
|
|
assert knowledge_base_id == "remote-one"
|
|
return {
|
|
"id": knowledge_base_id,
|
|
"vector_store_name": "System default",
|
|
"vector_store_source": "env",
|
|
"vector_store_engine_type": "elasticsearch",
|
|
"vector_store_status": "available",
|
|
"wiki_config": {"enabled": True},
|
|
}
|
|
|
|
async def list_wiki_pages(self, knowledge_base_id: str, **kwargs):
|
|
assert knowledge_base_id == "remote-one"
|
|
assert kwargs == {"page": 1, "page_size": 500, "query": ""}
|
|
return {
|
|
"pages": [
|
|
{
|
|
"id": "page-one",
|
|
"slug": "concept/start",
|
|
"title": "开始使用",
|
|
"page_type": "concept",
|
|
"status": "completed",
|
|
"created_at": "2026-07-02T08:00:00+08:00",
|
|
"updated_at": "2026-08-02T08:00:00+08:00",
|
|
},
|
|
{
|
|
"id": "page-two",
|
|
"slug": "summary/overview",
|
|
"title": "知识概览",
|
|
"page_type": "summary",
|
|
"status": "completed",
|
|
"created_at": "2026-07-03T08:00:00+08:00",
|
|
"updated_at": "2026-08-03T08:00:00+08:00",
|
|
},
|
|
],
|
|
"total": 2,
|
|
"page": 1,
|
|
"page_size": 500,
|
|
"total_pages": 1,
|
|
}
|
|
|
|
monkeypatch.setattr(llmwiki, "_client_or_503", lambda: Client())
|
|
|
|
payload = await llmwiki.list_public_knowledge_bases(_request(store))
|
|
|
|
assert store.calls == [("__public_wiki__", "public", False)]
|
|
assert payload["total"] == 1
|
|
item = payload["knowledge_bases"][0]
|
|
assert item["id"] == "1063edee-332a-43b4-bbd9-73bf0a708b94"
|
|
assert item["name"] == "远端公开知识库"
|
|
assert item["description"] == "远端公开说明"
|
|
assert item["type"] == "wiki"
|
|
assert item["knowledge_count"] == 12
|
|
assert item["chunk_count"] == 34
|
|
assert item["processing_count"] == 1
|
|
assert item["share_count"] == 3
|
|
assert item["is_processing"] is True
|
|
assert item["remote_created_at"] == "2026-07-01T08:00:00+08:00"
|
|
assert item["remote_updated_at"] == "2026-08-01T08:00:00+08:00"
|
|
assert item["vector_store_engine_type"] == "elasticsearch"
|
|
assert item["wiki"] == {
|
|
"status": "ready",
|
|
"page_count": 2,
|
|
"loaded_page_count": 2,
|
|
"stats_complete": True,
|
|
"page_type_counts": {"concept": 1, "summary": 1},
|
|
"page_status_counts": {"completed": 2},
|
|
"index_mode": "generated",
|
|
"has_explicit_index": False,
|
|
"index_slug": "__index__",
|
|
"default_slug": "__index__",
|
|
"created_at": "2026-07-02T08:00:00+08:00",
|
|
"updated_at": "2026-08-03T08:00:00+08:00",
|
|
"embed_route": "/embed/knowledge/1063edee-332a-43b4-bbd9-73bf0a708b94?wikiSlug=__index__",
|
|
}
|
|
assert item["local_index"]["state"] == "not_synced"
|
|
assert item["local_index"]["fully_vectorized"] is False
|
|
assert item["owner_user_id"] is None
|
|
assert "weknora_id" not in item
|
|
assert "tenant_id" not in item
|
|
assert "creator_id" not in item
|
|
assert "storage_config" not in item
|
|
|
|
|
|
@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"] == []
|
|
|
|
|
|
def test_public_wiki_summary_prefers_an_explicit_index_page() -> None:
|
|
payload = llmwiki._public_wiki_summary(
|
|
"mapping-one",
|
|
[
|
|
{
|
|
"slug": "guide/start",
|
|
"title": "开始使用",
|
|
"page_type": "article",
|
|
"status": "completed",
|
|
},
|
|
{
|
|
"slug": "home",
|
|
"title": "知识首页",
|
|
"page_type": "index",
|
|
"status": "completed",
|
|
},
|
|
],
|
|
total=2,
|
|
complete=True,
|
|
)
|
|
|
|
assert payload["status"] == "ready"
|
|
assert payload["index_mode"] == "explicit"
|
|
assert payload["has_explicit_index"] is True
|
|
assert payload["index_slug"] == "home"
|
|
assert payload["default_slug"] == "home"
|
|
assert payload["embed_route"] == "/embed/knowledge/mapping-one?wikiSlug=home"
|
|
|
|
|
|
@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
|