1022 lines
39 KiB
Python
1022 lines
39 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
|
|
import httpx
|
|
import pytest
|
|
from fastapi import HTTPException
|
|
from starlette.datastructures import Headers, QueryParams
|
|
from starlette.requests import Request
|
|
|
|
from deerflow.config.agents_config import AgentConfig
|
|
from deerflow.config.llmwiki_config import (
|
|
LlmWikiConfig,
|
|
LlmWikiRuntimeOverride,
|
|
resolve_llmwiki_runtime,
|
|
)
|
|
from deerflow.integrations.weknora.client import WeKnoraClient, WeKnoraError
|
|
from deerflow.persistence.llmwiki import MemoryLlmWikiStore
|
|
from deerflow.tools.builtins.llmwiki_search_tool import (
|
|
_sanitize_conversation_deposit_value as sanitize_tool_deposit_value,
|
|
selected_knowledge_base_ids,
|
|
)
|
|
|
|
|
|
def _request(
|
|
path: str,
|
|
*,
|
|
query: str = "",
|
|
cookies: dict[str, str] | None = None,
|
|
request_headers: dict[str, str] | None = None,
|
|
) -> Request:
|
|
headers: list[tuple[bytes, bytes]] = [
|
|
(key.lower().encode(), value.encode()) for key, value in (request_headers or {}).items()
|
|
]
|
|
if cookies:
|
|
headers.append(("cookie".encode(), "; ".join(f"{key}={value}" for key, value in cookies.items()).encode()))
|
|
return Request(
|
|
{
|
|
"type": "http",
|
|
"method": "GET",
|
|
"path": path,
|
|
"query_string": query.encode(),
|
|
"headers": headers,
|
|
"scheme": "http",
|
|
"server": ("testserver", 80),
|
|
"client": ("testclient", 50000),
|
|
}
|
|
)
|
|
|
|
|
|
def _first_gateway_match(method: str, path: str) -> str:
|
|
from app.gateway.app import app as gateway_app
|
|
|
|
for route in gateway_app.router.routes:
|
|
route_contexts = getattr(route, "effective_route_contexts", None)
|
|
candidates = route_contexts() if callable(route_contexts) else [route]
|
|
for candidate in candidates:
|
|
methods = getattr(candidate, "methods", None)
|
|
path_regex = getattr(candidate, "path_regex", None)
|
|
if methods is not None and method in methods and path_regex is not None and path_regex.match(path):
|
|
return str(getattr(candidate, "path", ""))
|
|
raise AssertionError(f"No route matched {method} {path}")
|
|
|
|
|
|
def test_empty_weknora_address_keeps_legacy_mode() -> None:
|
|
runtime = resolve_llmwiki_runtime(LlmWikiConfig())
|
|
|
|
assert runtime.provider == "legacy"
|
|
assert runtime.api_base_url == ""
|
|
|
|
|
|
def test_configured_weknora_address_enables_weknora_mode() -> None:
|
|
config = LlmWikiConfig.model_validate(
|
|
{
|
|
"weknora": {
|
|
"api_base_url": "http://weknora-app:8080/",
|
|
"web_base_url": "https://weknora.example.test/",
|
|
"admin_email": "admin@example.test",
|
|
"admin_password": "secret",
|
|
}
|
|
}
|
|
)
|
|
|
|
runtime = resolve_llmwiki_runtime(config)
|
|
|
|
assert runtime.provider == "weknora"
|
|
assert runtime.api_base_url == "http://weknora-app:8080"
|
|
assert runtime.web_base_url == "https://weknora.example.test"
|
|
assert runtime.admin_email == "admin@example.test"
|
|
assert runtime.admin_password == "secret"
|
|
|
|
|
|
def test_runtime_override_can_force_legacy_or_reset_to_file() -> None:
|
|
config = LlmWikiConfig.model_validate({"weknora": {"api_base_url": "http://weknora:8080"}})
|
|
|
|
forced_legacy = resolve_llmwiki_runtime(
|
|
config,
|
|
LlmWikiRuntimeOverride(enabled=True, api_base_url=""),
|
|
)
|
|
reset_to_file = resolve_llmwiki_runtime(
|
|
config,
|
|
LlmWikiRuntimeOverride(enabled=False, api_base_url=""),
|
|
)
|
|
|
|
assert forced_legacy.provider == "legacy"
|
|
assert reset_to_file.provider == "weknora"
|
|
|
|
|
|
def test_weknora_url_rejects_credentials_and_non_http_schemes() -> None:
|
|
with pytest.raises(ValueError):
|
|
LlmWikiConfig.model_validate({"weknora": {"api_base_url": "ftp://weknora.local"}})
|
|
with pytest.raises(ValueError):
|
|
LlmWikiConfig.model_validate({"weknora": {"api_base_url": "http://user:pass@weknora.local"}})
|
|
|
|
|
|
def test_agent_config_tracks_allowed_llmwiki_knowledge_bases() -> None:
|
|
config = AgentConfig(
|
|
name="Research agent",
|
|
llmwiki_knowledge_base_ids=["kb-one", "kb-two", "kb-one"],
|
|
)
|
|
|
|
assert config.llmwiki_knowledge_base_ids == ["kb-one", "kb-two"]
|
|
|
|
|
|
def test_conversation_selection_is_deduplicated_and_absence_is_distinct() -> None:
|
|
assert selected_knowledge_base_ids({}) is None
|
|
assert selected_knowledge_base_ids({"llmwiki_knowledge_base_ids": []}) == []
|
|
assert selected_knowledge_base_ids({"llmwiki_knowledge_base_ids": ["kb-one", "kb-one", "", "kb-two"]}) == ["kb-one", "kb-two"]
|
|
|
|
|
|
def test_weknora_frame_bootstrap_accepts_query_token_only_for_frame_route() -> None:
|
|
from app.gateway.deps import get_request_access_token
|
|
|
|
frame_request = _request(
|
|
"/api/llmwiki/knowledge-bases/local-kb/weknora/frame",
|
|
query="access_token=deerflow-token",
|
|
)
|
|
normal_request = _request("/api/models", query="access_token=deerflow-token")
|
|
|
|
assert get_request_access_token(frame_request) == "deerflow-token"
|
|
assert get_request_access_token(normal_request) is None
|
|
|
|
|
|
def test_weknora_embed_cookie_bypasses_only_proxy_paths() -> None:
|
|
from app.gateway import auth_middleware, csrf_middleware
|
|
from app.gateway.weknora_embed import sign_weknora_embed_payload
|
|
|
|
raw_cookie = sign_weknora_embed_payload(
|
|
{
|
|
"mapping_id": "local-kb",
|
|
"weknora_id": "remote-kb",
|
|
"user_id": "user-a",
|
|
"is_admin": False,
|
|
"can_write": False,
|
|
"exp": 4_102_444_800,
|
|
}
|
|
)
|
|
cookies = {"deerflow_weknora_embed": raw_cookie}
|
|
|
|
assert auth_middleware._is_weknora_embed_proxy_request(_request("/platform/knowledge-bases/kb", cookies=cookies))
|
|
assert auth_middleware._is_weknora_embed_proxy_request(_request("/assets/index.js", cookies=cookies))
|
|
assert auth_middleware._is_weknora_embed_proxy_request(_request("/config.js", cookies=cookies))
|
|
assert auth_middleware._is_weknora_embed_proxy_request(
|
|
_request("/tdesign-icons/0.4.1/fonts/index.js", cookies=cookies)
|
|
)
|
|
assert auth_middleware._is_weknora_embed_proxy_request(_request("/api/v1/knowledge-search", cookies=cookies))
|
|
assert auth_middleware._is_weknora_embed_proxy_request(
|
|
_request("/api/llmwiki/weknora-embed/api/v1/auth/me", cookies=cookies)
|
|
)
|
|
assert csrf_middleware.is_weknora_embed_proxy_request(_request("/api/v1/knowledge-search", cookies=cookies))
|
|
assert not auth_middleware._is_weknora_embed_proxy_request(
|
|
_request("/api/v1/knowledge-search", cookies={"deerflow_weknora_embed": "signed"})
|
|
)
|
|
assert not csrf_middleware.is_weknora_embed_proxy_request(
|
|
_request("/api/v1/knowledge-search", cookies={"deerflow_weknora_embed": "signed"})
|
|
)
|
|
assert not auth_middleware._is_weknora_embed_proxy_request(_request("/api/v1/auth/me", cookies=cookies))
|
|
assert not auth_middleware._is_weknora_embed_proxy_request(_request("/api/models", cookies=cookies))
|
|
assert not auth_middleware._is_weknora_embed_proxy_request(_request("/platform/knowledge-bases/kb"))
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_weknora_session_endpoint_sets_cookie_and_returns_platform_path(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
from app.gateway.routers import llmwiki as llmwiki_router
|
|
|
|
async def authorized_mapping(*_: object, **__: object) -> tuple[dict[str, str], str, bool]:
|
|
return {"id": "local-kb", "weknora_id": "remote-kb", "name": "KB"}, "user-a", False
|
|
|
|
class Store:
|
|
async def get_authorized(self, *_: object, **__: object) -> dict[str, str]:
|
|
return {"id": "local-kb", "weknora_id": "remote-kb"}
|
|
|
|
monkeypatch.setattr(llmwiki_router, "_runtime_for_iframe_proxy", lambda _: object())
|
|
monkeypatch.setattr(llmwiki_router, "_authorized_mapping", authorized_mapping)
|
|
monkeypatch.setattr(llmwiki_router, "_store", lambda _: Store())
|
|
|
|
response = await llmwiki_router.create_weknora_detail_session(
|
|
_request(
|
|
"/api/llmwiki/knowledge-bases/local-kb/weknora/session",
|
|
request_headers={"X-Forwarded-Proto": "https"},
|
|
),
|
|
"local-kb",
|
|
tab="wiki",
|
|
)
|
|
|
|
payload = json.loads(response.body)
|
|
assert payload["frame_path"] == "/platform/knowledge-bases/remote-kb?tab=wiki"
|
|
cookie = response.headers["set-cookie"]
|
|
assert "deerflow_weknora_embed=" in cookie
|
|
assert "Secure" in cookie
|
|
assert "SameSite=none" in cookie
|
|
raw = cookie.split("deerflow_weknora_embed=", 1)[1].split(";", 1)[0]
|
|
ctx = llmwiki_router._verify_embed_cookie(raw)
|
|
assert ctx.mapping_id == "local-kb"
|
|
assert ctx.weknora_id == "remote-kb"
|
|
assert ctx.can_write is True
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_weknora_client_sends_no_credentials_and_normalizes_search() -> None:
|
|
requests: list[httpx.Request] = []
|
|
|
|
async def handler(request: httpx.Request) -> httpx.Response:
|
|
requests.append(request)
|
|
assert "X-API-Key" not in request.headers
|
|
assert "Authorization" not in request.headers
|
|
assert json.loads(request.content) == {
|
|
"query": "deployment guide",
|
|
"knowledge_base_ids": ["kb-one", "kb-two"],
|
|
}
|
|
return httpx.Response(
|
|
200,
|
|
json={
|
|
"success": True,
|
|
"data": [
|
|
{
|
|
"id": "chunk-one",
|
|
"content": "matched passage",
|
|
"knowledge_id": "doc-one",
|
|
"knowledge_base_id": "kb-one",
|
|
"knowledge_title": "Guide",
|
|
"score": 0.93,
|
|
}
|
|
],
|
|
},
|
|
)
|
|
|
|
async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as http_client:
|
|
client = WeKnoraClient(
|
|
base_url="http://weknora.local:8080",
|
|
http_client=http_client,
|
|
)
|
|
results = await client.search("deployment guide", ["kb-one", "kb-two"])
|
|
|
|
assert requests[0].url.path == "/api/v1/knowledge-search"
|
|
assert results[0]["chunk_id"] == "chunk-one"
|
|
assert results[0]["title"] == "Guide"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_weknora_client_merges_required_name_for_description_update() -> None:
|
|
requests: list[httpx.Request] = []
|
|
|
|
async def handler(request: httpx.Request) -> httpx.Response:
|
|
requests.append(request)
|
|
if request.method == "GET":
|
|
return httpx.Response(
|
|
200,
|
|
json={"success": True, "data": {"id": "kb-one", "name": "对话沉淀"}},
|
|
)
|
|
assert request.method == "PUT"
|
|
assert json.loads(request.content) == {"name": "对话沉淀", "description": "cmzs description"}
|
|
return httpx.Response(
|
|
200,
|
|
json={"success": True, "data": {"id": "kb-one", "name": "对话沉淀", "description": "cmzs description"}},
|
|
)
|
|
|
|
async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as http_client:
|
|
client = WeKnoraClient(base_url="http://weknora.local:8080", http_client=http_client)
|
|
updated = await client.update_knowledge_base("kb-one", {"description": "cmzs description"})
|
|
|
|
assert updated["description"] == "cmzs description"
|
|
assert [request.method for request in requests] == ["GET", "PUT"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_weknora_client_resolves_default_embedding_model_inside_weknora() -> None:
|
|
async def handler(request: httpx.Request) -> httpx.Response:
|
|
if request.method == "GET" and request.url.path == "/api/v1/models":
|
|
return httpx.Response(
|
|
200,
|
|
json={
|
|
"success": True,
|
|
"data": [
|
|
{"id": "embed-default", "type": "Embedding", "is_default": True},
|
|
],
|
|
},
|
|
)
|
|
assert request.method == "POST"
|
|
assert request.url.path == "/api/v1/knowledge-bases"
|
|
assert json.loads(request.content)["embedding_model_id"] == "embed-default"
|
|
return httpx.Response(201, json={"success": True, "data": {"id": "kb-created", "name": "Docs"}})
|
|
|
|
async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as http_client:
|
|
client = WeKnoraClient(
|
|
base_url="http://weknora.local:8080",
|
|
http_client=http_client,
|
|
)
|
|
created = await client.create_knowledge_base(name="Docs")
|
|
|
|
assert created["id"] == "kb-created"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_weknora_client_creates_wiki_with_weknora_owned_models() -> None:
|
|
model_reads = 0
|
|
|
|
async def handler(request: httpx.Request) -> httpx.Response:
|
|
nonlocal model_reads
|
|
if request.method == "GET" and request.url.path == "/api/v1/models":
|
|
model_reads += 1
|
|
return httpx.Response(
|
|
200,
|
|
json={
|
|
"data": [
|
|
{"id": "embed-one", "type": "Embedding", "is_default": True},
|
|
{"id": "qa-one", "type": "KnowledgeQA", "status": "active"},
|
|
]
|
|
},
|
|
)
|
|
payload = json.loads(request.content)
|
|
assert payload["embedding_model_id"] == "embed-one"
|
|
assert payload["summary_model_id"] == "qa-one"
|
|
assert payload["indexing_strategy"] == {
|
|
"vector_enabled": True,
|
|
"keyword_enabled": True,
|
|
"wiki_enabled": True,
|
|
"graph_enabled": False,
|
|
}
|
|
assert payload["wiki_config"]["synthesis_model_id"] == "qa-one"
|
|
return httpx.Response(201, json={"data": {"id": "wiki-one", "name": "Wiki"}})
|
|
|
|
async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as http_client:
|
|
client = WeKnoraClient(base_url="http://weknora.local:8080", http_client=http_client)
|
|
created = await client.create_knowledge_base(name="Wiki", wiki_enabled=True)
|
|
|
|
assert created["id"] == "wiki-one"
|
|
assert model_reads == 2
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_weknora_client_imports_url_manual_content_and_lists_wiki_pages() -> None:
|
|
async def handler(request: httpx.Request) -> httpx.Response:
|
|
if request.url.path.endswith("/knowledge/url"):
|
|
assert json.loads(request.content) == {"url": "https://example.test/guide"}
|
|
return httpx.Response(201, json={"data": {"id": "url-one"}})
|
|
if request.url.path.endswith("/knowledge/manual"):
|
|
assert json.loads(request.content) == {
|
|
"title": "Guide",
|
|
"content": "# Guide",
|
|
"status": "publish",
|
|
}
|
|
return httpx.Response(201, json={"data": {"id": "manual-one"}})
|
|
assert request.url.path == "/api/v1/knowledgebase/kb-one/wiki/pages"
|
|
assert request.url.params["query"] == "guide"
|
|
return httpx.Response(
|
|
200,
|
|
json={"pages": [{"slug": "guide", "title": "Guide"}], "total": 1, "page": 1, "page_size": 50},
|
|
)
|
|
|
|
async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as http_client:
|
|
client = WeKnoraClient(base_url="http://weknora.local:8080", http_client=http_client)
|
|
await client.import_document_url("kb-one", url="https://example.test/guide")
|
|
await client.create_manual_document("kb-one", title="Guide", content="# Guide")
|
|
pages = await client.list_wiki_pages("kb-one", query="guide")
|
|
|
|
assert pages["total"] == 1
|
|
assert pages["pages"][0]["slug"] == "guide"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_weknora_client_reads_document_chunks_and_wiki_graph() -> None:
|
|
async def handler(request: httpx.Request) -> httpx.Response:
|
|
if request.url.path == "/api/v1/chunks/doc-one":
|
|
assert request.url.params["page"] == "2"
|
|
assert request.url.params["page_size"] == "10"
|
|
return httpx.Response(
|
|
200,
|
|
json={
|
|
"success": True,
|
|
"data": [{"id": "chunk-one", "content": "parsed content", "chunk_index": 10}],
|
|
"total": 31,
|
|
"page": 2,
|
|
"page_size": 10,
|
|
},
|
|
)
|
|
assert request.url.path == "/api/v1/knowledgebase/kb-one/wiki/graph"
|
|
assert request.url.params["mode"] == "overview"
|
|
assert request.url.params["limit"] == "50"
|
|
return httpx.Response(
|
|
200,
|
|
json={
|
|
"success": True,
|
|
"data": {
|
|
"nodes": [{"slug": "entity/one", "title": "Entity One", "page_type": "entity"}],
|
|
"edges": [],
|
|
"meta": {"mode": "overview", "total": 1, "returned": 1},
|
|
},
|
|
},
|
|
)
|
|
|
|
async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as http_client:
|
|
client = WeKnoraClient(base_url="http://weknora.local:8080", http_client=http_client)
|
|
chunks = await client.list_chunks("doc-one", page=2, page_size=10)
|
|
graph = await client.get_wiki_graph("kb-one", limit=50)
|
|
|
|
assert chunks["total"] == 31
|
|
assert chunks["items"][0]["content"] == "parsed content"
|
|
assert graph["nodes"][0]["slug"] == "entity/one"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_weknora_client_proxies_original_document_preview_without_credentials() -> None:
|
|
async def handler(request: httpx.Request) -> httpx.Response:
|
|
assert request.url.path == "/api/v1/knowledge/doc-one/preview"
|
|
assert "X-API-Key" not in request.headers
|
|
assert "Authorization" not in request.headers
|
|
return httpx.Response(200, content=b"preview", headers={"Content-Type": "text/plain; charset=utf-8"})
|
|
|
|
async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as http_client:
|
|
client = WeKnoraClient(base_url="http://weknora.local:8080", http_client=http_client)
|
|
content, media_type = await client.get_document_preview("doc-one")
|
|
|
|
assert content == b"preview"
|
|
assert media_type == "text/plain"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_memory_store_enforces_owner_and_public_visibility() -> None:
|
|
store = MemoryLlmWikiStore()
|
|
created = await store.create_mapping(
|
|
weknora_id="wk-private",
|
|
owner_user_id="user-a",
|
|
name="Private notes",
|
|
description="",
|
|
kb_type="document",
|
|
)
|
|
|
|
assert [row["id"] for row in await store.list_visible("user-a", scope="personal")] == [created["id"]]
|
|
assert await store.list_visible("user-b", scope="personal") == []
|
|
assert await store.get_authorized(created["id"], "user-b", write=False, is_admin=False) is None
|
|
|
|
published = await store.request_publish(created["id"], "user-a", is_admin=False)
|
|
assert published and published["publication_status"] == "published"
|
|
assert [row["id"] for row in await store.list_visible("user-b", scope="public")] == [created["id"]]
|
|
assert await store.get_authorized(created["id"], "user-b", write=False, is_admin=False) is not None
|
|
assert await store.get_authorized(created["id"], "user-b", write=True, is_admin=False) is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_conversation_deposit_mapping_is_system_owned_and_published(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
from types import SimpleNamespace
|
|
|
|
from app.gateway import llmwiki_deposit
|
|
|
|
class FakeClient:
|
|
async def list_knowledge_bases(self) -> list[dict[str, object]]:
|
|
return []
|
|
|
|
async def create_knowledge_base(self, **_: object) -> dict[str, object]:
|
|
return {
|
|
"id": "remote-deposit",
|
|
"name": "对话沉淀",
|
|
"description": "global",
|
|
"type": "document",
|
|
}
|
|
|
|
store = MemoryLlmWikiStore()
|
|
app = SimpleNamespace(state=SimpleNamespace(llmwiki_store=store))
|
|
monkeypatch.setattr(llmwiki_deposit, "_client", lambda _: FakeClient())
|
|
|
|
row = await llmwiki_deposit.ensure_conversation_deposit_knowledge_base(app, owner_user_id="user-a")
|
|
|
|
assert row is not None
|
|
assert row["owner_user_id"] == "system"
|
|
assert row["publication_status"] == "published"
|
|
assert row["description"] == llmwiki_deposit.CONVERSATION_DEPOSIT_KB_DESCRIPTION
|
|
assert row["id"] not in [item["id"] for item in await store.list_visible("user-a", scope="personal")]
|
|
assert [item["id"] for item in await store.list_visible("user-a", scope="public")] == [row["id"]]
|
|
|
|
markdown = llmwiki_deposit._format_markdown(
|
|
llmwiki_deposit.ConversationTurnDeposit(
|
|
thread_id="thread-one",
|
|
question="question",
|
|
answer="answer",
|
|
human_message_id="human-one",
|
|
assistant_message_id="assistant-one",
|
|
assistant_id=None,
|
|
knowledge_base_ids=[],
|
|
created_by_user_id="user-a",
|
|
),
|
|
"title",
|
|
)
|
|
assert "DeerFlow" not in markdown
|
|
assert "source: cmzs" in markdown
|
|
assert "#cmzs" in markdown
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_existing_conversation_deposit_description_is_refreshed(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
from types import SimpleNamespace
|
|
|
|
from app.gateway import llmwiki_deposit
|
|
|
|
updates: list[tuple[str, dict[str, object]]] = []
|
|
|
|
class FakeClient:
|
|
async def update_knowledge_base(self, knowledge_base_id: str, changes: dict[str, object]) -> dict[str, object]:
|
|
updates.append((knowledge_base_id, changes))
|
|
return {"id": knowledge_base_id, **changes}
|
|
|
|
store = MemoryLlmWikiStore()
|
|
created = await store.create_mapping(
|
|
weknora_id="remote-deposit",
|
|
owner_user_id="system",
|
|
name=llmwiki_deposit.CONVERSATION_DEPOSIT_KB_NAME,
|
|
description="old description",
|
|
kb_type="document",
|
|
)
|
|
await store.request_publish(created["id"], "system", is_admin=True)
|
|
app = SimpleNamespace(state=SimpleNamespace(llmwiki_store=store))
|
|
monkeypatch.setattr(llmwiki_deposit, "_client", lambda _: FakeClient())
|
|
|
|
row = await llmwiki_deposit.ensure_conversation_deposit_knowledge_base(app, owner_user_id="user-a")
|
|
|
|
assert row is not None
|
|
assert row["description"] == llmwiki_deposit.CONVERSATION_DEPOSIT_KB_DESCRIPTION
|
|
assert updates == [
|
|
(
|
|
"remote-deposit",
|
|
{"description": llmwiki_deposit.CONVERSATION_DEPOSIT_KB_DESCRIPTION},
|
|
)
|
|
]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_legacy_user_owned_conversation_deposit_is_adopted(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
from types import SimpleNamespace
|
|
|
|
from app.gateway import llmwiki_deposit
|
|
|
|
class FakeClient:
|
|
async def list_knowledge_bases(self) -> list[dict[str, object]]:
|
|
return [
|
|
{
|
|
"id": "remote-deposit",
|
|
"name": llmwiki_deposit.CONVERSATION_DEPOSIT_KB_NAME,
|
|
"description": "DeerFlow legacy description",
|
|
}
|
|
]
|
|
|
|
async def update_knowledge_base(self, knowledge_base_id: str, changes: dict[str, object]) -> dict[str, object]:
|
|
return {"id": knowledge_base_id, **changes}
|
|
|
|
store = MemoryLlmWikiStore()
|
|
legacy = await store.create_mapping(
|
|
weknora_id="remote-deposit",
|
|
owner_user_id="user-a",
|
|
name=llmwiki_deposit.CONVERSATION_DEPOSIT_KB_NAME,
|
|
description="DeerFlow legacy description",
|
|
kb_type="document",
|
|
)
|
|
await store.request_publish(legacy["id"], "user-a", is_admin=False)
|
|
app = SimpleNamespace(state=SimpleNamespace(llmwiki_store=store))
|
|
monkeypatch.setattr(llmwiki_deposit, "_client", lambda _: FakeClient())
|
|
|
|
row = await llmwiki_deposit.ensure_conversation_deposit_knowledge_base(app, owner_user_id="user-a")
|
|
|
|
assert row is not None
|
|
assert row["id"] == legacy["id"]
|
|
assert row["owner_user_id"] == "system"
|
|
assert row["publication_status"] == "published"
|
|
assert row["description"] == llmwiki_deposit.CONVERSATION_DEPOSIT_KB_DESCRIPTION
|
|
assert await store.list_visible("user-a", scope="personal") == []
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_listing_conversation_deposit_refreshes_remote_description(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
from app.gateway import llmwiki_deposit
|
|
from app.gateway.routers import llmwiki as llmwiki_router
|
|
|
|
updates: list[tuple[str, dict[str, object]]] = []
|
|
store = MemoryLlmWikiStore()
|
|
created = await store.create_mapping(
|
|
weknora_id="remote-deposit",
|
|
owner_user_id="system",
|
|
name=llmwiki_deposit.CONVERSATION_DEPOSIT_KB_NAME,
|
|
description="DeerFlow legacy description",
|
|
kb_type="document",
|
|
)
|
|
await store.request_publish(created["id"], "system", is_admin=True)
|
|
|
|
class Client:
|
|
async def list_knowledge_bases(self) -> list[dict[str, object]]:
|
|
return [
|
|
{
|
|
"id": "remote-deposit",
|
|
"name": llmwiki_deposit.CONVERSATION_DEPOSIT_KB_NAME,
|
|
"description": "DeerFlow legacy description",
|
|
}
|
|
]
|
|
|
|
async def create_knowledge_base(self, **_: object) -> dict[str, object]:
|
|
raise AssertionError("The existing deposit should be reused")
|
|
|
|
async def update_knowledge_base(self, knowledge_base_id: str, changes: dict[str, object]) -> dict[str, object]:
|
|
updates.append((knowledge_base_id, changes))
|
|
return {"id": knowledge_base_id, **changes}
|
|
|
|
async def actor(_: object) -> tuple[str, bool]:
|
|
return "user-a", False
|
|
|
|
monkeypatch.setattr(llmwiki_router, "_actor", actor)
|
|
monkeypatch.setattr(llmwiki_router, "_store", lambda _: store)
|
|
monkeypatch.setattr(llmwiki_router, "_client_or_503", lambda: Client())
|
|
|
|
payload = await llmwiki_router.list_knowledge_bases(object(), scope="public")
|
|
|
|
assert payload["knowledge_bases"][0]["description"] == llmwiki_deposit.CONVERSATION_DEPOSIT_KB_DESCRIPTION
|
|
assert "DeerFlow" not in json.dumps(payload, ensure_ascii=False)
|
|
assert updates == [
|
|
(
|
|
"remote-deposit",
|
|
{"description": llmwiki_deposit.CONVERSATION_DEPOSIT_KB_DESCRIPTION},
|
|
)
|
|
]
|
|
|
|
|
|
def test_conversation_deposit_history_hides_legacy_brand() -> None:
|
|
from app.gateway.routers import llmwiki as llmwiki_router
|
|
|
|
value = {
|
|
"content": "This page was generated by DeerFlow.",
|
|
"metadata": {"source": "deer-flow", "tags": ["DEER_FLOW", "keep"]},
|
|
}
|
|
|
|
gateway_value = llmwiki_router._sanitize_conversation_deposit_value(value)
|
|
tool_value = sanitize_tool_deposit_value(value)
|
|
|
|
assert gateway_value == tool_value
|
|
assert gateway_value == {
|
|
"content": "This page was generated by cmzs.",
|
|
"metadata": {"source": "cmzs", "tags": ["cmzs", "keep"]},
|
|
}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_knowledge_bases_personal_scope_filters_to_actor_for_admin(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
from app.gateway.routers import llmwiki as llmwiki_router
|
|
|
|
calls: list[dict[str, object]] = []
|
|
|
|
class Store:
|
|
async def list_visible(self, user_id: str, *, scope: str, is_admin: bool) -> list[dict[str, object]]:
|
|
calls.append({"user_id": user_id, "scope": scope, "is_admin": is_admin})
|
|
return [
|
|
{
|
|
"id": "local-admin",
|
|
"weknora_id": "remote-admin",
|
|
"owner_user_id": user_id,
|
|
"name": "Admin private",
|
|
"description": "",
|
|
"kb_type": "document",
|
|
"publication_status": "private",
|
|
}
|
|
]
|
|
|
|
class Client:
|
|
async def list_knowledge_bases(self) -> list[dict[str, object]]:
|
|
return [{"id": "remote-admin", "name": "Admin private"}]
|
|
|
|
async def actor(_: object) -> tuple[str, bool]:
|
|
return "admin-id", True
|
|
|
|
monkeypatch.setattr(llmwiki_router, "_actor", actor)
|
|
monkeypatch.setattr(llmwiki_router, "_store", lambda _: Store())
|
|
monkeypatch.setattr(llmwiki_router, "_client_or_503", lambda: Client())
|
|
|
|
payload = await llmwiki_router.list_knowledge_bases(object(), scope="personal")
|
|
|
|
assert calls == [{"user_id": "admin-id", "scope": "personal", "is_admin": False}]
|
|
assert payload["knowledge_bases"][0]["is_owner"] is True
|
|
|
|
|
|
def test_mapping_view_distinguishes_owner_from_admin_write_power() -> None:
|
|
from app.gateway.routers import llmwiki as llmwiki_router
|
|
|
|
view = llmwiki_router._mapping_view(
|
|
{
|
|
"id": "local-other",
|
|
"weknora_id": "remote-other",
|
|
"owner_user_id": "user-b",
|
|
"name": "Other private",
|
|
"description": "",
|
|
"kb_type": "document",
|
|
"publication_status": "private",
|
|
},
|
|
{"id": "remote-other", "name": "Other private"},
|
|
actor_user_id="admin-id",
|
|
is_admin=True,
|
|
)
|
|
|
|
assert view["is_owner"] is False
|
|
assert view["can_write"] is True
|
|
|
|
|
|
def test_only_system_owned_named_base_is_treated_as_conversation_deposit() -> None:
|
|
from app.gateway.routers import llmwiki as llmwiki_router
|
|
|
|
personal = {
|
|
"id": "personal-named-deposit",
|
|
"owner_user_id": "user-a",
|
|
"name": "对话沉淀",
|
|
"publication_status": "private",
|
|
}
|
|
system = {
|
|
"id": "system-deposit",
|
|
"owner_user_id": "system",
|
|
"name": "对话沉淀",
|
|
"publication_status": "published",
|
|
}
|
|
|
|
assert not llmwiki_router._is_conversation_deposit_mapping(personal)
|
|
assert llmwiki_router._is_conversation_deposit_mapping(system)
|
|
system_view = llmwiki_router._mapping_view(
|
|
system,
|
|
{
|
|
"id": "remote-system",
|
|
"name": "对话沉淀",
|
|
"description": "DeerFlow legacy description",
|
|
},
|
|
actor_user_id="admin-id",
|
|
is_admin=True,
|
|
)
|
|
assert system_view["can_write"] is False
|
|
assert system_view["description"] == "cmzs 全局问答沉淀知识库。所有用户可见，不默认参与问答检索。"
|
|
assert "DeerFlow" not in system_view["description"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_conversation_deposit_requires_thread_access(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
from types import SimpleNamespace
|
|
|
|
from app.gateway.routers import llmwiki as llmwiki_router
|
|
|
|
class ThreadStore:
|
|
async def check_access(self, thread_id: str, user_id: str, *, require_existing: bool) -> bool:
|
|
assert thread_id == "thread-other"
|
|
assert user_id == "user-a"
|
|
assert require_existing is True
|
|
return False
|
|
|
|
async def actor(_: object) -> tuple[str, bool]:
|
|
return "user-a", False
|
|
|
|
monkeypatch.setattr(llmwiki_router, "_actor", actor)
|
|
monkeypatch.setattr(llmwiki_router, "get_thread_store", lambda _: ThreadStore())
|
|
monkeypatch.setattr(
|
|
llmwiki_router,
|
|
"get_resolved_llmwiki_runtime",
|
|
lambda _: SimpleNamespace(weknora_enabled=True, provider="weknora"),
|
|
)
|
|
request = SimpleNamespace(app=SimpleNamespace(state=SimpleNamespace(config=object())))
|
|
body = llmwiki_router.ConversationDepositCreate(
|
|
thread_id="thread-other",
|
|
question="question",
|
|
answer="answer",
|
|
assistant_message_id="assistant-one",
|
|
)
|
|
|
|
with pytest.raises(HTTPException) as exc:
|
|
await llmwiki_router.create_conversation_deposit(
|
|
request,
|
|
SimpleNamespace(add_task=lambda *_: None),
|
|
body,
|
|
)
|
|
|
|
assert exc.value.status_code == 404
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_source_context_prefers_wiki_page_over_raw_chunks(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
from app.gateway.routers import llmwiki as llmwiki_router
|
|
|
|
class Client:
|
|
async def get_chunk_context(self, chunk_id: str, *, expected_knowledge_base_id: str, radius: int) -> list[dict[str, object]]:
|
|
assert chunk_id == "chunk-one"
|
|
assert expected_knowledge_base_id == "remote-kb"
|
|
return [
|
|
{
|
|
"id": "chunk-one",
|
|
"chunk_index": 0,
|
|
"content": "raw split chunk",
|
|
"metadata": {"wiki_page_slug": "people/alice"},
|
|
}
|
|
]
|
|
|
|
async def get_wiki_page(self, knowledge_base_id: str, slug: str) -> dict[str, object]:
|
|
assert knowledge_base_id == "remote-kb"
|
|
assert slug == "people/alice"
|
|
return {
|
|
"id": "wiki-one",
|
|
"slug": "people/alice",
|
|
"title": "Alice",
|
|
"summary": "clean summary",
|
|
"content": "clean llmwiki page",
|
|
"source_refs": ["chunk-one"],
|
|
}
|
|
|
|
async def authorized_mapping(*_: object, **__: object) -> tuple[dict[str, str], str, bool]:
|
|
return {"id": "local-kb", "weknora_id": "remote-kb", "name": "Public LLMWiki"}, "user-a", False
|
|
|
|
monkeypatch.setattr(llmwiki_router, "_authorized_mapping", authorized_mapping)
|
|
monkeypatch.setattr(llmwiki_router, "_client_or_503", lambda: Client())
|
|
|
|
payload = await llmwiki_router.get_source_context(object(), knowledge_base_id="local-kb", chunk_id="chunk-one")
|
|
|
|
assert payload["data"]["display_mode"] == "wiki"
|
|
assert payload["data"]["wiki_page"]["content"] == "clean llmwiki page"
|
|
assert payload["data"]["chunks"][0]["content"] == "raw split chunk"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_delete_cleans_owner_mapping_when_remote_is_already_missing(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
from app.gateway.routers import llmwiki as llmwiki_router
|
|
|
|
deleted: list[str] = []
|
|
|
|
class MissingClient:
|
|
async def delete_knowledge_base(self, _: str) -> None:
|
|
raise WeKnoraError("WeKnora resource was not found", status_code=404)
|
|
|
|
class Store:
|
|
async def delete_mapping(self, mapping_id: str) -> None:
|
|
deleted.append(mapping_id)
|
|
|
|
async def authorized_mapping(*_: object, **__: object) -> tuple[dict[str, str], str, bool]:
|
|
return {"weknora_id": "remote-missing"}, "owner", False
|
|
|
|
monkeypatch.setattr(llmwiki_router, "_authorized_mapping", authorized_mapping)
|
|
monkeypatch.setattr(llmwiki_router, "_client_or_503", lambda: MissingClient())
|
|
monkeypatch.setattr(llmwiki_router, "_store", lambda _: Store())
|
|
|
|
await llmwiki_router.delete_knowledge_base(object(), "mapping-one")
|
|
|
|
assert deleted == ["mapping-one"]
|
|
|
|
|
|
def test_weknora_iframe_cookie_is_signed_and_short_lived() -> None:
|
|
from app.gateway.routers import llmwiki as llmwiki_router
|
|
|
|
raw = llmwiki_router._sign_embed_payload(
|
|
{
|
|
"mapping_id": "local-kb",
|
|
"weknora_id": "remote-kb",
|
|
"user_id": "user-a",
|
|
"is_admin": False,
|
|
"can_write": True,
|
|
"exp": 4_102_444_800,
|
|
}
|
|
)
|
|
|
|
ctx = llmwiki_router._verify_embed_cookie(raw)
|
|
|
|
assert ctx.mapping_id == "local-kb"
|
|
assert ctx.weknora_id == "remote-kb"
|
|
assert ctx.can_write is True
|
|
|
|
with pytest.raises(HTTPException):
|
|
llmwiki_router._verify_embed_cookie(raw + "tampered")
|
|
|
|
|
|
def test_weknora_iframe_proxy_is_fallback_after_deerflow_api_routes() -> None:
|
|
assert _first_gateway_match("POST", "/api/v1/auth/login/username") == "/api/v1/auth/login/username"
|
|
assert _first_gateway_match("POST", "/api/v1/knowledge-search") == "/api/v1/{proxied_path:path}"
|
|
assert (
|
|
_first_gateway_match("GET", "/api/llmwiki/weknora-embed/api/v1/auth/me")
|
|
== "/api/llmwiki/weknora-embed/api/v1/{proxied_path:path}"
|
|
)
|
|
assert _first_gateway_match("GET", "/platform/knowledge-bases/remote-kb") == "/platform/{proxied_path:path}"
|
|
|
|
|
|
def test_weknora_iframe_filters_kb_list_to_current_mapping() -> None:
|
|
from app.gateway.routers import llmwiki as llmwiki_router
|
|
|
|
ctx = llmwiki_router.WeKnoraEmbedContext(
|
|
mapping_id="local-kb",
|
|
weknora_id="remote-kb",
|
|
user_id="user-a",
|
|
is_admin=False,
|
|
can_write=False,
|
|
exp=4_102_444_800,
|
|
)
|
|
response = httpx.Response(
|
|
200,
|
|
json={
|
|
"success": True,
|
|
"data": [
|
|
{"id": "remote-kb", "name": "Allowed"},
|
|
{"id": "other-kb", "name": "Blocked"},
|
|
],
|
|
},
|
|
)
|
|
|
|
filtered = llmwiki_router._filter_knowledge_base_list_response(response, ctx)
|
|
|
|
assert filtered is not None
|
|
payload = json.loads(filtered)
|
|
assert payload["data"] == [{"id": "remote-kb", "name": "Allowed"}]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_conversation_deposit_iframe_sanitizes_legacy_brand(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
from types import SimpleNamespace
|
|
|
|
from app.gateway.routers import llmwiki as llmwiki_router
|
|
|
|
async def fake_session(_: object, *, force_refresh: bool = False) -> dict[str, object]:
|
|
return {"token": "token", "tenant": {}}
|
|
|
|
async def authorize(*_: object) -> None:
|
|
return None
|
|
|
|
class FakeAsyncClient:
|
|
def __init__(self, **_: object) -> None:
|
|
pass
|
|
|
|
async def __aenter__(self) -> "FakeAsyncClient":
|
|
return self
|
|
|
|
async def __aexit__(self, *_: object) -> None:
|
|
return None
|
|
|
|
async def request(self, *_: object, **__: object) -> httpx.Response:
|
|
return httpx.Response(
|
|
200,
|
|
json={"success": True, "data": {"summary": "Generated by DeerFlow", "source": "deer-flow"}},
|
|
)
|
|
|
|
class FakeRequest:
|
|
method = "GET"
|
|
headers = Headers({})
|
|
query_params = QueryParams("")
|
|
|
|
async def body(self) -> bytes:
|
|
return b""
|
|
|
|
ctx = llmwiki_router.WeKnoraEmbedContext(
|
|
mapping_id="local-deposit",
|
|
weknora_id="remote-deposit",
|
|
user_id="user-a",
|
|
is_admin=False,
|
|
can_write=False,
|
|
exp=4_102_444_800,
|
|
is_conversation_deposit=True,
|
|
)
|
|
monkeypatch.setattr(llmwiki_router, "_runtime_for_iframe_proxy", lambda _: SimpleNamespace())
|
|
monkeypatch.setattr(llmwiki_router, "_get_weknora_admin_session", fake_session)
|
|
monkeypatch.setattr(llmwiki_router, "_authorize_weknora_api_proxy", authorize)
|
|
monkeypatch.setattr(llmwiki_router.httpx, "AsyncClient", FakeAsyncClient)
|
|
|
|
response = await llmwiki_router._proxy_weknora_request(
|
|
FakeRequest(),
|
|
target_base_url="http://weknora.local",
|
|
upstream_path="/api/v1/knowledge-bases/remote-deposit",
|
|
ctx=ctx,
|
|
is_api=True,
|
|
)
|
|
|
|
payload = json.loads(response.body)
|
|
assert payload["data"] == {"summary": "Generated by cmzs", "source": "cmzs"}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_weknora_iframe_proxy_blocks_other_knowledge_bases() -> None:
|
|
from app.gateway.routers import llmwiki as llmwiki_router
|
|
|
|
class FakeRequest:
|
|
method = "GET"
|
|
headers = Headers({})
|
|
query_params = QueryParams("")
|
|
|
|
ctx = llmwiki_router.WeKnoraEmbedContext(
|
|
mapping_id="local-kb",
|
|
weknora_id="remote-kb",
|
|
user_id="user-a",
|
|
is_admin=False,
|
|
can_write=False,
|
|
exp=4_102_444_800,
|
|
)
|
|
|
|
await llmwiki_router._authorize_weknora_api_proxy(
|
|
FakeRequest(),
|
|
ctx,
|
|
object(),
|
|
"/api/v1/knowledge-bases/remote-kb",
|
|
b"",
|
|
"token",
|
|
)
|
|
|
|
with pytest.raises(HTTPException) as exc_info:
|
|
await llmwiki_router._authorize_weknora_api_proxy(
|
|
FakeRequest(),
|
|
ctx,
|
|
object(),
|
|
"/api/v1/knowledge-bases/other-kb",
|
|
b"",
|
|
"token",
|
|
)
|
|
|
|
assert exc_info.value.status_code == 403
|