1904 lines
64 KiB
Python
1904 lines
64 KiB
Python
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import json
|
|
import threading
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
from fastapi import FastAPI, HTTPException
|
|
from langchain_core.messages import HumanMessage
|
|
from starlette.requests import Request
|
|
|
|
from app.gateway.services import (
|
|
_with_knowledge_retrieval_progress_stream,
|
|
enforce_knowledge_retrieval_mode_run_config,
|
|
resolve_knowledge_retrieval_mode,
|
|
)
|
|
from deerflow.integrations.weknora.client import WeKnoraClient, WeKnoraError
|
|
from deerflow.tools.builtins import llmwiki_reasoning_tools as tools
|
|
|
|
|
|
class _SettingsStore:
|
|
def __init__(self, enabled: bool) -> None:
|
|
self.enabled = enabled
|
|
|
|
async def get(self) -> bool:
|
|
return self.enabled
|
|
|
|
|
|
class _MappingStore:
|
|
async def get_authorized(
|
|
self,
|
|
mapping_id: str,
|
|
user_id: str,
|
|
*,
|
|
write: bool,
|
|
is_admin: bool,
|
|
) -> dict | None:
|
|
assert user_id == "user-1"
|
|
assert write is False
|
|
if mapping_id != "mapping-1":
|
|
return None
|
|
return {
|
|
"id": "mapping-1",
|
|
"weknora_id": "remote-1",
|
|
"name": "制度库",
|
|
}
|
|
|
|
|
|
class _Client:
|
|
async def search(self, query: str, knowledge_base_ids: list[str]) -> list[dict]:
|
|
assert knowledge_base_ids == ["remote-1"]
|
|
return [
|
|
{
|
|
"chunk_id": "chunk-1",
|
|
"knowledge_id": "doc-1",
|
|
"knowledge_base_id": "remote-1",
|
|
"title": "采购制度",
|
|
"content": "采购金额超过十万元时需要集体审议。",
|
|
"score": 0.8 if query == "采购金额" else 0.9,
|
|
}
|
|
]
|
|
|
|
async def list_documents(self, knowledge_base_id: str, *, page: int, page_size: int) -> dict:
|
|
assert knowledge_base_id == "remote-1"
|
|
return {
|
|
"items": [{"id": "doc-1", "title": "采购制度", "updated_at": "2026-01-01"}],
|
|
"total": 1,
|
|
}
|
|
|
|
async def list_chunks(self, knowledge_id: str, *, page: int, page_size: int) -> dict:
|
|
assert knowledge_id == "doc-1"
|
|
return {
|
|
"items": [
|
|
{"id": "chunk-1", "content": "条款 A-17：采购金额超过十万元。", "chunk_index": 2},
|
|
{"id": "chunk-2", "content": "普通说明。", "chunk_index": 3},
|
|
],
|
|
"total": 2,
|
|
}
|
|
|
|
async def grep_chunks(self, query: str, knowledge_base_ids: list[str]) -> dict:
|
|
assert query == r"A-17|十万元"
|
|
assert knowledge_base_ids == ["remote-1"]
|
|
return {
|
|
"items": [
|
|
{
|
|
"chunk_id": "chunk-1",
|
|
"knowledge_id": "doc-1",
|
|
"knowledge_base_id": "remote-1",
|
|
"title": "采购制度",
|
|
"content": "条款 A-17：采购金额超过十万元。",
|
|
"match_count": 2,
|
|
"chunk_index": 2,
|
|
}
|
|
],
|
|
"contract": "weknora-v0.7.2-grep-v1",
|
|
"compatibility_fallback": False,
|
|
"partial": False,
|
|
"warnings": [],
|
|
"successful_knowledge_base_ids": ["remote-1"],
|
|
"failed_knowledge_base_ids": [],
|
|
"scanned_chunks": 2,
|
|
"scan_truncated": False,
|
|
}
|
|
|
|
|
|
def _request(store: _SettingsStore | None) -> Request:
|
|
app = FastAPI()
|
|
if store is not None:
|
|
app.state.knowledge_retrieval_settings_store = store
|
|
return Request(
|
|
{
|
|
"type": "http",
|
|
"method": "POST",
|
|
"path": "/api/threads/t/runs",
|
|
"headers": [],
|
|
"query_string": b"",
|
|
"app": app,
|
|
}
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_database_gate_defaults_classic_and_rejects_disabled_smart_mode() -> None:
|
|
assert await resolve_knowledge_retrieval_mode(
|
|
_request(None),
|
|
SimpleNamespace(context={}),
|
|
) == "classic"
|
|
|
|
with pytest.raises(HTTPException) as caught:
|
|
await resolve_knowledge_retrieval_mode(
|
|
_request(_SettingsStore(False)),
|
|
SimpleNamespace(context={"knowledge_retrieval_mode": "smart_reasoning"}),
|
|
)
|
|
assert caught.value.status_code == 409
|
|
assert caught.value.detail["code"] == "SMART_REASONING_DISABLED"
|
|
|
|
assert await resolve_knowledge_retrieval_mode(
|
|
_request(_SettingsStore(True)),
|
|
SimpleNamespace(context={"knowledge_retrieval_mode": "smart_reasoning"}),
|
|
) == "smart_reasoning"
|
|
|
|
|
|
def test_validated_mode_overwrites_untrusted_preseeded_config() -> None:
|
|
config = {
|
|
"configurable": {
|
|
"knowledge_retrieval_mode": "smart_reasoning",
|
|
"disable_tools": True,
|
|
"excluded_tools": ["ordinary_wiki_search"],
|
|
},
|
|
"context": {
|
|
"knowledge_retrieval_mode": "smart_reasoning",
|
|
"disable_tools": True,
|
|
"excluded_tools": ["ordinary_knowledge_search"],
|
|
},
|
|
"metadata": {"rag_mode_enabled": True, "rag_mode": True},
|
|
}
|
|
enforce_knowledge_retrieval_mode_run_config(config, "classic")
|
|
assert config["configurable"]["knowledge_retrieval_mode"] == "classic"
|
|
assert config["context"]["knowledge_retrieval_mode"] == "classic"
|
|
|
|
enforce_knowledge_retrieval_mode_run_config(config, "smart_reasoning")
|
|
assert config["configurable"]["rag_mode_enabled"] is False
|
|
assert config["context"]["rag_mode_enabled"] is False
|
|
assert config["configurable"]["knowledge_rag_enabled"] is False
|
|
assert config["context"]["knowledge_rag_enabled"] is False
|
|
assert config["metadata"]["rag_mode_enabled"] is False
|
|
assert config["metadata"]["rag_mode"] is False
|
|
assert config["configurable"]["disable_tools"] is False
|
|
assert config["context"]["disable_tools"] is False
|
|
assert set(config["configurable"]["excluded_tools"]) == set(
|
|
tools.CLASSIC_LLMWIKI_TOOL_NAMES
|
|
)
|
|
assert set(config["context"]["excluded_tools"]) == set(
|
|
tools.CLASSIC_LLMWIKI_TOOL_NAMES
|
|
)
|
|
|
|
|
|
def test_smart_mode_always_enables_custom_progress_stream() -> None:
|
|
assert _with_knowledge_retrieval_progress_stream(
|
|
["values", "messages-tuple"],
|
|
"smart_reasoning",
|
|
) == ["values", "messages-tuple", "custom"]
|
|
assert _with_knowledge_retrieval_progress_stream(
|
|
["values", "custom", "custom"],
|
|
"smart_reasoning",
|
|
) == ["values", "custom"]
|
|
assert _with_knowledge_retrieval_progress_stream(
|
|
["values"],
|
|
"classic",
|
|
) == ["values"]
|
|
|
|
|
|
def test_numbered_citation_prompt_is_classic_only() -> None:
|
|
from app.gateway import services
|
|
|
|
graph_input = {"messages": [HumanMessage(content="查制度")]}
|
|
smart = services._apply_classic_rag_citation_prompt(
|
|
graph_input,
|
|
enabled=True,
|
|
knowledge_retrieval_mode="smart_reasoning",
|
|
)
|
|
smart_extra = smart["messages"][0].additional_kwargs
|
|
assert services._RAG_CITATION_PROMPT not in str(
|
|
smart_extra.get("prompt_prefix") or ""
|
|
)
|
|
|
|
classic = services._apply_classic_rag_citation_prompt(
|
|
graph_input,
|
|
enabled=True,
|
|
knowledge_retrieval_mode="classic",
|
|
)
|
|
assert services._RAG_CITATION_PROMPT in str(
|
|
classic["messages"][0].additional_kwargs.get("prompt_prefix") or ""
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize("assistant_id", [None, "deep-research-planner", "deep-research-collector"])
|
|
async def test_smart_scope_binds_this_library_to_authorized_database_metadata(
|
|
monkeypatch: pytest.MonkeyPatch, assistant_id: str | None,
|
|
) -> None:
|
|
from app.gateway import llmwiki_rag
|
|
|
|
class ScopeStore:
|
|
async def get_authorized(
|
|
self,
|
|
mapping_id: str,
|
|
user_id: str,
|
|
*,
|
|
write: bool,
|
|
is_admin: bool,
|
|
) -> dict | None:
|
|
assert (mapping_id, user_id, write, is_admin) == (
|
|
"mapping-news",
|
|
"user-1",
|
|
False,
|
|
False,
|
|
)
|
|
return {
|
|
"id": mapping_id,
|
|
"weknora_id": "remote-news",
|
|
"name": "国内新闻资料库",
|
|
"description": "收录国内重要新闻和政策发布。",
|
|
"wiki_focus_instructions": "突出时间、地点和关键数据",
|
|
"wiki_directory_paths": [["年度", "重大活动"], ["政策", "发布会"]],
|
|
}
|
|
|
|
app = FastAPI()
|
|
app.state.config = SimpleNamespace()
|
|
app.state.llmwiki_store = ScopeStore()
|
|
request = Request(
|
|
{
|
|
"type": "http",
|
|
"method": "POST",
|
|
"path": "/api/threads/t/runs",
|
|
"headers": [],
|
|
"query_string": b"",
|
|
"app": app,
|
|
}
|
|
)
|
|
monkeypatch.setattr(
|
|
llmwiki_rag,
|
|
"get_resolved_llmwiki_runtime",
|
|
lambda _config: SimpleNamespace(weknora_enabled=True),
|
|
)
|
|
|
|
async def actor(_request: Request) -> tuple[str, bool]:
|
|
return "user-1", False
|
|
|
|
monkeypatch.setattr(llmwiki_rag, "_actor", actor)
|
|
rag = await llmwiki_rag.build_llmwiki_smart_reasoning_context(
|
|
request=request,
|
|
body=SimpleNamespace(
|
|
context={"llmwiki_knowledge_base_ids": ["mapping-news"]},
|
|
metadata={},
|
|
assistant_id=assistant_id,
|
|
),
|
|
graph_input={"messages": [HumanMessage(content="这个库讲了什么")]},
|
|
)
|
|
|
|
assert rag.knowledge_base_context == [
|
|
{
|
|
"id": "mapping-news",
|
|
"name": "国内新闻资料库",
|
|
"description": "收录国内重要新闻和政策发布。",
|
|
"focus": "突出时间、地点和关键数据",
|
|
"directory_hints": ["年度 / 重大活动", "政策 / 发布会"],
|
|
}
|
|
]
|
|
assert "这个库" in rag.prompt
|
|
assert "国内新闻资料库" in rag.prompt
|
|
assert "ordinary_knowledge_search" in rag.prompt
|
|
assert "/mnt/knowledge" in rag.prompt
|
|
assert "不要追问" in rag.prompt
|
|
|
|
attached = llmwiki_rag.attach_llmwiki_rag_to_graph_input(
|
|
{"messages": [HumanMessage(content="这个库讲了什么")]},
|
|
rag,
|
|
)
|
|
extra = attached["messages"][0].additional_kwargs
|
|
assert extra["llmwiki_knowledge_base_context"] == rag.knowledge_base_context
|
|
assert "这个库" in extra["smart_knowledge_scope_prompt"]
|
|
assert "prompt_prefix" not in extra
|
|
|
|
|
|
def test_smart_scope_keeps_all_identities_under_one_global_prompt_budget() -> None:
|
|
from app.gateway import llmwiki_rag
|
|
|
|
rows = [
|
|
{
|
|
"id": f"mapping-{index:03d}",
|
|
"name": f"第{index:03d}个知识库" + ("长名称" * 40),
|
|
"description": "详细简介" * 300,
|
|
"wiki_focus_instructions": "抽取重点" * 300,
|
|
"wiki_directory_paths": [
|
|
[f"一级目录{index}", "二级目录" * 80]
|
|
for _ in range(12)
|
|
],
|
|
}
|
|
for index in range(100)
|
|
]
|
|
|
|
contexts = llmwiki_rag._smart_knowledge_base_context(rows)
|
|
serialized = json.dumps(contexts, ensure_ascii=False, separators=(",", ":"))
|
|
prompt = llmwiki_rag._build_smart_scope_prompt(contexts)
|
|
|
|
assert len(contexts) == 100
|
|
assert [item["id"] for item in contexts] == [
|
|
f"mapping-{index:03d}" for index in range(100)
|
|
]
|
|
assert all(item["name"] for item in contexts)
|
|
assert len(serialized) <= llmwiki_rag._MAX_SCOPE_CONTEXT_JSON_CHARS
|
|
assert len(prompt) <= llmwiki_rag._MAX_SMART_SCOPE_PROMPT_CHARS
|
|
assert any(item.get("description") for item in contexts)
|
|
assert any("description" not in item for item in contexts)
|
|
|
|
|
|
def test_smart_metadata_disables_reference_mode_while_classic_keeps_it() -> None:
|
|
from app.gateway import services
|
|
|
|
smart_run = {"rag_mode_enabled": True, "rag_mode": True}
|
|
smart_record = {"rag_mode_enabled": True, "rag_mode": True}
|
|
smart_thread = {"rag_mode_enabled": True, "rag_mode": True}
|
|
smart = services._sync_retrieval_rag_metadata(
|
|
run_metadata=smart_run,
|
|
record_metadata=smart_record,
|
|
thread_metadata_update=smart_thread,
|
|
knowledge_retrieval_mode="smart_reasoning",
|
|
knowledge_scope_active=True,
|
|
selection_declared=True,
|
|
explicit_rag_requested=True,
|
|
)
|
|
assert smart == {"rag_mode_enabled": False, "rag_mode": False}
|
|
assert smart_run == smart
|
|
assert smart_record == smart
|
|
assert smart_thread == smart
|
|
|
|
classic_run: dict = {}
|
|
classic_record: dict = {}
|
|
classic_thread: dict = {}
|
|
classic = services._sync_retrieval_rag_metadata(
|
|
run_metadata=classic_run,
|
|
record_metadata=classic_record,
|
|
thread_metadata_update=classic_thread,
|
|
knowledge_retrieval_mode="classic",
|
|
knowledge_scope_active=True,
|
|
selection_declared=True,
|
|
explicit_rag_requested=False,
|
|
)
|
|
assert classic == {"rag_mode_enabled": True, "rag_mode": True}
|
|
assert classic_run == classic
|
|
assert classic_record == classic
|
|
assert classic_thread == classic
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_smart_run_stops_before_launch_when_mode_metadata_cannot_persist(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
from app.gateway import analysis_projects, services
|
|
from app.gateway.llmwiki_rag import LlmWikiRagContext
|
|
|
|
class ThreadStore:
|
|
def __init__(self) -> None:
|
|
self.statuses: list[str] = []
|
|
self.metadata_writes = 0
|
|
|
|
async def get(self, _thread_id: str) -> dict:
|
|
return {"metadata": {}, "assistant_id": None}
|
|
|
|
async def update_status(self, _thread_id: str, status: str) -> None:
|
|
self.statuses.append(status)
|
|
|
|
async def update_metadata(self, _thread_id: str, _metadata: dict) -> None:
|
|
self.metadata_writes += 1
|
|
raise RuntimeError("database unavailable")
|
|
|
|
record = SimpleNamespace(run_id="run-1", metadata={}, task=None)
|
|
|
|
class RunManager:
|
|
def __init__(self) -> None:
|
|
self.terminal_status: object | None = None
|
|
|
|
async def create_or_reject(self, *_args, **kwargs):
|
|
record.metadata = dict(kwargs["metadata"])
|
|
return record
|
|
|
|
async def set_status(self, _run_id, status, **_kwargs) -> None:
|
|
self.terminal_status = status
|
|
|
|
thread_store = ThreadStore()
|
|
run_manager = RunManager()
|
|
run_context = SimpleNamespace(app_config=object(), thread_store=thread_store)
|
|
|
|
async def prepare_project_run(body, _thread_id, _request):
|
|
return body, None
|
|
|
|
async def unexpected_fixed_answer_lookup(**_kwargs):
|
|
raise AssertionError("smart retrieval must bypass fixed answers")
|
|
|
|
async def smart_context(**_kwargs):
|
|
return LlmWikiRagContext(["kb-1"], [], 0, "", [], "")
|
|
|
|
launched = False
|
|
|
|
async def launch(**_kwargs):
|
|
nonlocal launched
|
|
launched = True
|
|
|
|
monkeypatch.setattr(analysis_projects, "prepare_project_run", prepare_project_run)
|
|
monkeypatch.setattr(services, "get_stream_bridge", lambda _request: object())
|
|
monkeypatch.setattr(services, "get_run_manager", lambda _request: run_manager)
|
|
monkeypatch.setattr(services, "get_run_context", lambda _request: run_context)
|
|
monkeypatch.setattr(services, "get_concurrency_gate", lambda _request: None)
|
|
monkeypatch.setattr(services, "get_effective_user_id", lambda: "user-1")
|
|
monkeypatch.setattr(
|
|
services,
|
|
"_find_fixed_answer",
|
|
unexpected_fixed_answer_lookup,
|
|
)
|
|
monkeypatch.setattr(services, "_resolve_effective_model_name", lambda *_args, **_kwargs: "model-1")
|
|
monkeypatch.setattr(services, "_get_model_concurrency_limit", lambda _name: None)
|
|
monkeypatch.setattr(services, "_get_user_model_concurrency_limit", lambda _name: None)
|
|
monkeypatch.setattr(services, "resolve_agent_factory", lambda _assistant_id: object())
|
|
monkeypatch.setattr(services, "build_llmwiki_smart_reasoning_context", smart_context)
|
|
monkeypatch.setattr(services, "_run_agent_with_concurrency_lease", launch)
|
|
|
|
body = SimpleNamespace(
|
|
input={"messages": [{"role": "user", "content": "查制度"}]},
|
|
config={},
|
|
context={"knowledge_retrieval_mode": "smart_reasoning"},
|
|
metadata={},
|
|
assistant_id=None,
|
|
on_disconnect="continue",
|
|
multitask_strategy="reject",
|
|
)
|
|
request = SimpleNamespace(
|
|
app=SimpleNamespace(
|
|
state=SimpleNamespace(
|
|
knowledge_retrieval_settings_store=_SettingsStore(True),
|
|
)
|
|
),
|
|
client=None,
|
|
headers={},
|
|
)
|
|
|
|
with pytest.raises(RuntimeError, match="database unavailable"):
|
|
await services.start_run(body, "thread-1", request, enforce_access=False)
|
|
|
|
assert thread_store.metadata_writes == 1
|
|
assert thread_store.statuses == ["running", "error"]
|
|
assert str(run_manager.terminal_status).endswith("error")
|
|
assert launched is False
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_multi_query_search_merges_same_chunk_and_preserves_query_evidence(monkeypatch) -> None:
|
|
monkeypatch.setattr(tools, "get_llmwiki_store", lambda: _MappingStore())
|
|
monkeypatch.setattr(tools, "get_effective_user_id", lambda: "user-1")
|
|
monkeypatch.setattr(tools, "get_current_user", lambda: SimpleNamespace(system_role="user"))
|
|
monkeypatch.setattr(tools, "_client", lambda: _Client())
|
|
|
|
payload = json.loads(
|
|
await tools._knowledge_search_impl(
|
|
["采购金额", "十万元审议"],
|
|
{
|
|
"knowledge_retrieval_mode": "smart_reasoning",
|
|
"llmwiki_knowledge_base_ids": ["mapping-1"],
|
|
},
|
|
)
|
|
)
|
|
|
|
assert payload["ok"] is True
|
|
assert payload["result_count"] == 1
|
|
assert payload["results"][0]["score"] == pytest.approx(0.9)
|
|
assert payload["results"][0]["matched_queries"] == ["采购金额", "十万元审议"]
|
|
assert payload["results"][0]["knowledge_base_id"] == "mapping-1"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_vector_tool_streams_public_query_and_aggregate_result_progress(
|
|
monkeypatch,
|
|
) -> None:
|
|
events: list[dict] = []
|
|
context = {
|
|
"knowledge_retrieval_mode": "smart_reasoning",
|
|
"llmwiki_knowledge_base_ids": ["mapping-1", "mapping-2"],
|
|
}
|
|
runtime = SimpleNamespace(
|
|
context=context,
|
|
stream_writer=events.append,
|
|
tool_call_id="call-vector-1",
|
|
)
|
|
|
|
async def fake_search(queries, received_context):
|
|
assert queries == ["国内新闻", "政策发布"]
|
|
assert received_context is context
|
|
return tools._result(
|
|
operation="knowledge_search",
|
|
results=[
|
|
{
|
|
"chunk_id": "chunk-secret",
|
|
"knowledge_id": "doc-secret",
|
|
"knowledge_base_id": "mapping-1",
|
|
"knowledge_base": "国内新闻库",
|
|
"title": "新闻原文",
|
|
"content": "抽屉中显示的安全新闻正文预览",
|
|
}
|
|
],
|
|
queries=queries,
|
|
partial=True,
|
|
warnings=[{"code": "PROVIDER_QUERY_FAILED", "query_index": 2}],
|
|
)
|
|
|
|
monkeypatch.setattr(tools, "_knowledge_search_impl", fake_search)
|
|
clock = iter([10.0, 10.456])
|
|
monkeypatch.setattr(tools.time, "perf_counter", lambda: next(clock))
|
|
output = await tools.ordinary_knowledge_search_tool.coroutine(
|
|
queries=["国内新闻", "政策发布"],
|
|
runtime=runtime,
|
|
)
|
|
|
|
output_payload = json.loads(output)
|
|
assert output_payload["result_count"] == 1
|
|
assert output_payload["elapsed_ms"] == 456
|
|
assert [(event["stage"], event["status"]) for event in events] == [
|
|
("query_rewrite", "completed"),
|
|
("vector_search", "started"),
|
|
("vector_search", "completed"),
|
|
]
|
|
assert all(
|
|
event["type"] == "knowledge_retrieval_progress"
|
|
and event["version"] == 1
|
|
and event["retrieval_id"] == "call-vector-1"
|
|
for event in events
|
|
)
|
|
assert events[0]["queries"] == ["国内新闻", "政策发布"]
|
|
assert events[1]["knowledge_base_count"] == 2
|
|
assert events[-1]["result_count"] == 1
|
|
assert events[-1]["partial"] is True
|
|
assert events[-1]["elapsed_ms"] == 456
|
|
assert events[-1]["warning_codes"] == ["PROVIDER_QUERY_FAILED"]
|
|
assert events[-1]["result_items"] == [
|
|
{
|
|
"kind": "document_chunk",
|
|
"knowledge_base_id": "mapping-1",
|
|
"knowledge_base": "国内新闻库",
|
|
"title": "新闻原文",
|
|
"chunk_id": "chunk-secret",
|
|
"knowledge_id": "doc-secret",
|
|
"content": "抽屉中显示的安全新闻正文预览",
|
|
"content_truncated": False,
|
|
}
|
|
]
|
|
assert events[-1]["result_items_truncated"] is False
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_wiki_tool_streams_channel_coverage_and_scan_count(monkeypatch) -> None:
|
|
events: list[dict] = []
|
|
context = {
|
|
"knowledge_retrieval_mode": "smart_reasoning",
|
|
"llmwiki_knowledge_base_ids": ["mapping-1"],
|
|
}
|
|
runtime = SimpleNamespace(
|
|
context=context,
|
|
stream_writer=events.append,
|
|
tool_call_id="call-wiki-1",
|
|
)
|
|
|
|
async def fake_wiki(pattern, received_context):
|
|
assert pattern == "新闻|发布会"
|
|
assert received_context is context
|
|
return tools._result(
|
|
operation="wiki_search",
|
|
results=[
|
|
{
|
|
"page_ref": tools._page_ref("mapping-1", "news/release"),
|
|
"knowledge_base_id": "mapping-1",
|
|
"knowledge_base": "国内新闻库",
|
|
"title": "发布会",
|
|
"slug": "news/release",
|
|
"summary": "可在抽屉中展示的 Wiki 摘要",
|
|
}
|
|
],
|
|
pattern=pattern,
|
|
channel="provider_native_regex",
|
|
partial=False,
|
|
native_regex_supported=True,
|
|
unsupported_knowledge_base_count=0,
|
|
coverage=[
|
|
{
|
|
"knowledge_base": "国内新闻库",
|
|
"channel": "provider_native_regex",
|
|
"status": "searched",
|
|
}
|
|
],
|
|
scanned_pages=17,
|
|
)
|
|
|
|
monkeypatch.setattr(tools, "_wiki_search_impl", fake_wiki)
|
|
await tools.ordinary_wiki_search_tool.coroutine(
|
|
pattern="新闻|发布会",
|
|
runtime=runtime,
|
|
)
|
|
|
|
assert [(event["stage"], event["status"]) for event in events] == [
|
|
("query_rewrite", "completed"),
|
|
("wiki_regex_search", "started"),
|
|
("wiki_regex_search", "completed"),
|
|
]
|
|
completed = events[-1]
|
|
assert completed["pattern"] == "新闻|发布会"
|
|
assert completed["channel"] == "provider_native_regex"
|
|
assert completed["scanned_pages"] == 17
|
|
assert completed["result_count"] == 1
|
|
assert completed["native_regex_supported"] is True
|
|
assert completed["unsupported_knowledge_base_count"] == 0
|
|
assert completed["coverage"] == [
|
|
{
|
|
"knowledge_base": "国内新闻库",
|
|
"channel": "provider_native_regex",
|
|
"status": "searched",
|
|
}
|
|
]
|
|
assert completed["result_items"] == [
|
|
{
|
|
"kind": "wiki_page",
|
|
"knowledge_base_id": "mapping-1",
|
|
"knowledge_base": "国内新闻库",
|
|
"title": "发布会",
|
|
"page_ref": tools._page_ref("mapping-1", "news/release"),
|
|
"slug": "news/release",
|
|
"summary": "可在抽屉中展示的 Wiki 摘要",
|
|
}
|
|
]
|
|
|
|
|
|
def test_public_result_items_are_scope_filtered_and_bounded() -> None:
|
|
payload = {
|
|
"operation": "knowledge_search",
|
|
"results": [
|
|
{
|
|
"knowledge_base_id": "mapping-1",
|
|
"provider_knowledge_base_id": "remote-must-not-leak",
|
|
"knowledge_base": "制度库",
|
|
"title": f"文档 {index}",
|
|
"chunk_id": f"chunk-{index}",
|
|
"knowledge_id": f"doc-{index}",
|
|
"content": "甲" * 2_000,
|
|
}
|
|
for index in range(22)
|
|
]
|
|
+ [
|
|
{
|
|
"knowledge_base_id": "mapping-foreign",
|
|
"title": "越权结果",
|
|
"chunk_id": "foreign-chunk",
|
|
"content": "不得进入事件",
|
|
}
|
|
],
|
|
}
|
|
|
|
items, truncated = tools._public_result_items(
|
|
payload,
|
|
authorized_mapping_ids={"mapping-1"},
|
|
)
|
|
|
|
assert len(items) == tools._MAX_PROGRESS_RESULT_ITEMS
|
|
assert truncated is True
|
|
assert all(item["knowledge_base_id"] == "mapping-1" for item in items)
|
|
assert all(len(item["content"]) <= tools._MAX_PROGRESS_RAW_CONTENT_CHARS for item in items)
|
|
rendered = json.dumps(items, ensure_ascii=False)
|
|
assert "remote-must-not-leak" not in rendered
|
|
assert "mapping-foreign" not in rendered
|
|
assert "foreign-chunk" not in rendered
|
|
|
|
|
|
def test_oversized_nested_tool_metadata_still_returns_valid_json() -> None:
|
|
output = tools._result(
|
|
operation="wiki_browse_index",
|
|
results=[{"title": "入口"}],
|
|
groups=[{"type": "summary", "items": [{"summary": "甲" * 60_000}]}],
|
|
)
|
|
|
|
payload = json.loads(output)
|
|
assert payload == {
|
|
"ok": True,
|
|
"operation": "wiki_browse_index",
|
|
"result_count": 0,
|
|
"results": [],
|
|
"output_truncated": True,
|
|
}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_progress_failure_hides_raw_exception_and_keeps_tool_contract(
|
|
monkeypatch,
|
|
) -> None:
|
|
events: list[dict] = []
|
|
runtime = SimpleNamespace(
|
|
context={
|
|
"knowledge_retrieval_mode": "smart_reasoning",
|
|
"llmwiki_knowledge_base_ids": ["mapping-1"],
|
|
},
|
|
stream_writer=events.append,
|
|
tool_call_id="call-failed-1",
|
|
)
|
|
|
|
async def failed_search(_queries, _context):
|
|
raise RuntimeError("provider-token-must-never-leak")
|
|
|
|
monkeypatch.setattr(tools, "_knowledge_search_impl", failed_search)
|
|
clock = iter([20.0, 20.125])
|
|
monkeypatch.setattr(tools.time, "perf_counter", lambda: next(clock))
|
|
output = await tools.ordinary_knowledge_search_tool.coroutine(
|
|
queries=["国内新闻"],
|
|
runtime=runtime,
|
|
)
|
|
|
|
payload = json.loads(output)
|
|
assert payload["ok"] is False
|
|
assert payload["error"]["code"] == "SMART_RETRIEVAL_FAILED"
|
|
assert payload["elapsed_ms"] == 125
|
|
assert events[-1]["status"] == "failed"
|
|
assert events[-1]["elapsed_ms"] == 125
|
|
assert events[-1]["error"] == payload["error"]
|
|
assert "provider-token-must-never-leak" not in json.dumps(
|
|
events,
|
|
ensure_ascii=False,
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_chunk_regex_search_is_case_insensitive_and_bounded(monkeypatch) -> None:
|
|
monkeypatch.setattr(tools, "get_llmwiki_store", lambda: _MappingStore())
|
|
monkeypatch.setattr(tools, "get_effective_user_id", lambda: "user-1")
|
|
monkeypatch.setattr(tools, "get_current_user", lambda: SimpleNamespace(system_role="user"))
|
|
monkeypatch.setattr(tools, "_client", lambda: _Client())
|
|
|
|
payload = json.loads(
|
|
await tools._chunk_regex_impl(
|
|
r"A-17|十万元",
|
|
{
|
|
"knowledge_retrieval_mode": "smart_reasoning",
|
|
"llmwiki_knowledge_base_ids": ["mapping-1"],
|
|
},
|
|
)
|
|
)
|
|
|
|
assert payload["result_count"] == 1
|
|
assert payload["results"][0]["chunk_id"] == "chunk-1"
|
|
assert payload["scanned_chunks"] == 2
|
|
|
|
|
|
def test_wiki_page_reference_round_trip_and_regex_validation() -> None:
|
|
reference = tools._page_ref("mapping-1", "制度/采购")
|
|
assert tools._decode_page_ref(reference) == ("mapping-1", "制度/采购")
|
|
assert tools._safe_regex("采购|A-17").search("a-17")
|
|
with pytest.raises(tools.SmartKnowledgeToolError):
|
|
tools._safe_regex("(a+)+")
|
|
|
|
|
|
class _MultiMappingStore:
|
|
async def get_authorized(
|
|
self,
|
|
mapping_id: str,
|
|
user_id: str,
|
|
*,
|
|
write: bool,
|
|
is_admin: bool,
|
|
) -> dict | None:
|
|
assert user_id == "user-1"
|
|
assert write is False
|
|
if mapping_id not in {"mapping-1", "mapping-2"}:
|
|
return None
|
|
suffix = mapping_id.rsplit("-", 1)[-1]
|
|
return {
|
|
"id": mapping_id,
|
|
"weknora_id": f"remote-{suffix}",
|
|
"name": f"制度库{suffix}",
|
|
}
|
|
|
|
|
|
def _patch_scope(monkeypatch, *, multi: bool = False) -> None:
|
|
monkeypatch.setattr(
|
|
tools,
|
|
"get_llmwiki_store",
|
|
lambda: _MultiMappingStore() if multi else _MappingStore(),
|
|
)
|
|
monkeypatch.setattr(tools, "get_effective_user_id", lambda: "user-1")
|
|
monkeypatch.setattr(
|
|
tools,
|
|
"get_current_user",
|
|
lambda: SimpleNamespace(system_role="user"),
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_classic_wiki_index_is_scope_bound_drawer_ready_and_bounded(
|
|
monkeypatch,
|
|
) -> None:
|
|
_patch_scope(monkeypatch)
|
|
|
|
class Client:
|
|
async def get_wiki_index(
|
|
self,
|
|
knowledge_base_id: str,
|
|
*,
|
|
types: list[str],
|
|
limit: int,
|
|
cursor: str,
|
|
) -> dict:
|
|
assert knowledge_base_id == "remote-1"
|
|
assert types == ["summary", "entity"]
|
|
assert limit == 100
|
|
assert cursor == ""
|
|
return {
|
|
"intro": "入口见 [[news/timeline|新闻时间线]]。",
|
|
"groups": [
|
|
{
|
|
"type": "summary",
|
|
"total": 60,
|
|
"next_cursor": "next-summary",
|
|
"items": [
|
|
{
|
|
"id": f"page-{index}",
|
|
"slug": f"news/page-{index}",
|
|
"title": f"新闻页面 {index}",
|
|
"summary": "摘要" * 1_000,
|
|
"category_path": ["新闻", "国内"],
|
|
}
|
|
for index in range(60)
|
|
],
|
|
}
|
|
],
|
|
}
|
|
|
|
monkeypatch.setattr(tools, "_client", Client)
|
|
payload = json.loads(
|
|
await tools._wiki_index_impl(
|
|
"mapping-1",
|
|
{
|
|
"knowledge_retrieval_mode": "classic",
|
|
"llmwiki_knowledge_base_ids": ["mapping-1"],
|
|
},
|
|
types=["summary", "entity"],
|
|
limit=100,
|
|
)
|
|
)
|
|
|
|
assert payload["ok"] is True
|
|
assert payload["operation"] == "wiki_browse_index"
|
|
assert payload["results_truncated"] is True
|
|
assert len(payload["groups"][0]["items"]) == tools._MAX_CLASSIC_WIKI_INDEX_ITEMS
|
|
assert payload["groups"][0]["items_truncated"] is True
|
|
assert payload["groups"][0]["next_cursor"] == "next-summary"
|
|
assert "/weknora-source-preview?" in payload["results"][0]["content"]
|
|
assert "wiki_slug=news%2Ftimeline" in payload["results"][0]["content"]
|
|
assert len(json.dumps(payload, ensure_ascii=False)) < tools._MAX_TOOL_OUTPUT_CHARS
|
|
|
|
with pytest.raises(tools.SmartKnowledgeToolError) as caught:
|
|
await tools._wiki_index_impl(
|
|
"mapping-1",
|
|
{
|
|
"knowledge_retrieval_mode": "smart_reasoning",
|
|
"llmwiki_knowledge_base_ids": ["mapping-1"],
|
|
},
|
|
)
|
|
assert caught.value.code == "CLASSIC_MODE_REQUIRED"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_classic_wiki_page_read_returns_source_drawer_metadata(monkeypatch) -> None:
|
|
_patch_scope(monkeypatch)
|
|
|
|
class MissingLocalMirror:
|
|
async def list_index_status(self, _mapping_ids: list[str]) -> list[dict]:
|
|
raise RuntimeError("local index disabled")
|
|
|
|
class Client:
|
|
async def get_wiki_page(self, knowledge_base_id: str, slug: str) -> dict:
|
|
assert (knowledge_base_id, slug) == ("remote-1", "news/timeline")
|
|
return {
|
|
"slug": slug,
|
|
"title": "新闻时间线",
|
|
"description": "按日期整理",
|
|
"markdown": "火炬传递、开幕与闭幕时间线。",
|
|
"status": "published",
|
|
}
|
|
|
|
monkeypatch.setattr(tools, "get_local_wiki_index_store", MissingLocalMirror)
|
|
monkeypatch.setattr(tools, "_client", Client)
|
|
payload = json.loads(
|
|
await tools._wiki_read_impl(
|
|
tools._page_ref("mapping-1", "news/timeline"),
|
|
{
|
|
"knowledge_retrieval_mode": "classic",
|
|
"llmwiki_knowledge_base_ids": ["mapping-1"],
|
|
},
|
|
required_mode="classic",
|
|
)
|
|
)
|
|
|
|
page = payload["results"][0]
|
|
assert page["knowledge_base_id"] == "mapping-1"
|
|
assert page["knowledge_base_name"] == "制度库"
|
|
assert page["wiki_slug"] == "news/timeline"
|
|
assert page["type"] == "llmwiki_wiki_page"
|
|
assert page["content"] == "火炬传递、开幕与闭幕时间线。"
|
|
assert page["url"].endswith("knowledge_base_id=mapping-1&wiki_slug=news%2Ftimeline")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_authorization_read_is_routed_to_the_engine_event_loop(
|
|
monkeypatch,
|
|
) -> None:
|
|
home_loop = asyncio.new_event_loop()
|
|
ready = threading.Event()
|
|
|
|
def run_home_loop() -> None:
|
|
asyncio.set_event_loop(home_loop)
|
|
ready.set()
|
|
home_loop.run_forever()
|
|
home_loop.close()
|
|
|
|
thread = threading.Thread(target=run_home_loop, daemon=True)
|
|
thread.start()
|
|
assert ready.wait(timeout=2)
|
|
observed_loops: list[asyncio.AbstractEventLoop] = []
|
|
|
|
class Store:
|
|
async def get_authorized(
|
|
self,
|
|
mapping_id: str,
|
|
user_id: str,
|
|
*,
|
|
write: bool,
|
|
is_admin: bool,
|
|
) -> dict:
|
|
observed_loops.append(asyncio.get_running_loop())
|
|
return {"id": mapping_id, "owner_user_id": user_id}
|
|
|
|
monkeypatch.setattr(tools, "get_engine_loop", lambda: home_loop)
|
|
try:
|
|
row = await tools._get_authorized_on_engine_loop(
|
|
Store(),
|
|
"mapping-1",
|
|
"user-1",
|
|
is_admin=False,
|
|
)
|
|
finally:
|
|
home_loop.call_soon_threadsafe(home_loop.stop)
|
|
thread.join(timeout=2)
|
|
|
|
assert row == {"id": "mapping-1", "owner_user_id": "user-1"}
|
|
assert observed_loops == [home_loop]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_search_all_failures_are_errors_and_partial_failure_is_structured(
|
|
monkeypatch,
|
|
) -> None:
|
|
_patch_scope(monkeypatch)
|
|
|
|
class Client:
|
|
async def search(self, query: str, knowledge_base_ids: list[str]) -> list[dict]:
|
|
if query == "失败查询":
|
|
raise WeKnoraError("down")
|
|
return [
|
|
{
|
|
"chunk_id": "chunk-ok",
|
|
"knowledge_id": "doc-ok",
|
|
"knowledge_base_id": "remote-1",
|
|
"content": "可用证据",
|
|
"score": 1,
|
|
}
|
|
]
|
|
|
|
monkeypatch.setattr(tools, "_client", Client)
|
|
failed = json.loads(
|
|
await tools._knowledge_search_impl(
|
|
["失败查询"],
|
|
{
|
|
"knowledge_retrieval_mode": "smart_reasoning",
|
|
"llmwiki_knowledge_base_ids": ["mapping-1"],
|
|
},
|
|
)
|
|
)
|
|
assert failed["ok"] is False
|
|
assert failed["error"]["code"] == "PROVIDER_UNAVAILABLE"
|
|
|
|
partial = json.loads(
|
|
await tools._knowledge_search_impl(
|
|
["成功查询", "失败查询"],
|
|
{
|
|
"knowledge_retrieval_mode": "smart_reasoning",
|
|
"llmwiki_knowledge_base_ids": ["mapping-1"],
|
|
},
|
|
)
|
|
)
|
|
assert partial["ok"] is True
|
|
assert partial["partial"] is True
|
|
assert partial["warnings"] == [
|
|
{"code": "PROVIDER_QUERY_FAILED", "query_index": 2}
|
|
]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_wiki_search_uses_native_provider_for_each_mapping_concurrently(
|
|
monkeypatch,
|
|
) -> None:
|
|
_patch_scope(monkeypatch, multi=True)
|
|
|
|
class LocalIndexMustNotRun:
|
|
def __init__(self) -> None:
|
|
raise AssertionError("smart Wiki regex must use the native provider API")
|
|
|
|
class Client:
|
|
calls: list[tuple[str, str, int]] = []
|
|
active = 0
|
|
peak = 0
|
|
|
|
async def search_wiki_pages(
|
|
self,
|
|
knowledge_base_id: str,
|
|
query: str,
|
|
*,
|
|
limit: int,
|
|
) -> list[dict]:
|
|
self.calls.append((knowledge_base_id, query, limit))
|
|
self.active += 1
|
|
self.peak = max(self.peak, self.active)
|
|
await asyncio.sleep(0.01)
|
|
self.active -= 1
|
|
suffix = knowledge_base_id.rsplit("-", 1)[-1]
|
|
return [
|
|
{
|
|
"slug": f"news/{suffix}",
|
|
"title": f"needle 页面 {suffix}",
|
|
"content": "needle 正文",
|
|
"status": "published",
|
|
}
|
|
]
|
|
|
|
async def list_wiki_pages(self, *args, **kwargs) -> dict:
|
|
raise AssertionError("Wiki directory enumeration is forbidden")
|
|
|
|
async def get_wiki_page(self, *args, **kwargs) -> dict:
|
|
raise AssertionError("search must not fetch each page body")
|
|
|
|
client = Client()
|
|
monkeypatch.setattr(tools, "get_local_wiki_index_store", LocalIndexMustNotRun)
|
|
monkeypatch.setattr(tools, "_client", lambda: client)
|
|
payload = json.loads(
|
|
await tools._wiki_search_impl(
|
|
"needle",
|
|
{
|
|
"knowledge_retrieval_mode": "smart_reasoning",
|
|
"llmwiki_knowledge_base_ids": ["mapping-1", "mapping-2"],
|
|
},
|
|
)
|
|
)
|
|
|
|
assert payload["ok"] is True
|
|
assert payload["native_regex_supported"] is True
|
|
assert payload["result_count"] == 2
|
|
assert {call[:2] for call in client.calls} == {
|
|
("remote-1", "needle"),
|
|
("remote-2", "needle"),
|
|
}
|
|
assert client.peak == 2
|
|
assert payload["channel"] == "provider_native_regex"
|
|
assert [item["status"] for item in payload["coverage"]] == [
|
|
"searched",
|
|
"searched",
|
|
]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_wiki_search_marks_provider_limit_as_truncated(monkeypatch) -> None:
|
|
_patch_scope(monkeypatch)
|
|
|
|
class Client:
|
|
async def search_wiki_pages(
|
|
self,
|
|
knowledge_base_id: str,
|
|
query: str,
|
|
*,
|
|
limit: int,
|
|
) -> list[dict]:
|
|
assert (knowledge_base_id, query, limit) == ("remote-1", "needle", 50)
|
|
return [
|
|
{
|
|
"slug": f"news/{index}",
|
|
"title": f"needle 页面 {index}",
|
|
"content": "needle 正文",
|
|
}
|
|
for index in range(limit)
|
|
]
|
|
|
|
monkeypatch.setattr(tools, "_client", Client)
|
|
payload = json.loads(
|
|
await tools._wiki_search_impl(
|
|
"needle",
|
|
{
|
|
"knowledge_retrieval_mode": "smart_reasoning",
|
|
"llmwiki_knowledge_base_ids": ["mapping-1"],
|
|
},
|
|
)
|
|
)
|
|
|
|
assert payload["scanned_pages"] == 50
|
|
assert payload["scan_truncated"] is True
|
|
assert payload["result_count"] == tools._MAX_RESULTS
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_wiki_search_handles_supported_and_unsupported_mappings(
|
|
monkeypatch,
|
|
) -> None:
|
|
_patch_scope(monkeypatch, multi=True)
|
|
|
|
class Client:
|
|
async def search_wiki_pages(
|
|
self,
|
|
knowledge_base_id: str,
|
|
query: str,
|
|
*,
|
|
limit: int,
|
|
) -> list[dict]:
|
|
if knowledge_base_id == "remote-2":
|
|
raise WeKnoraError("route unavailable", status_code=404)
|
|
return [{"slug": "news/one", "title": f"{query} 页面"}]
|
|
|
|
monkeypatch.setattr(tools, "_client", Client)
|
|
payload = json.loads(
|
|
await tools._wiki_search_impl(
|
|
"needle",
|
|
{
|
|
"knowledge_retrieval_mode": "smart_reasoning",
|
|
"llmwiki_knowledge_base_ids": ["mapping-1", "mapping-2"],
|
|
},
|
|
)
|
|
)
|
|
|
|
assert payload["ok"] is True
|
|
assert payload["native_regex_supported"] is True
|
|
assert payload["unsupported_knowledge_base_count"] == 1
|
|
assert payload["partial"] is False
|
|
assert payload["warnings"] == []
|
|
assert [item["status"] for item in payload["coverage"]] == [
|
|
"searched",
|
|
"unsupported",
|
|
]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_wiki_search_without_native_client_method_is_unsupported(
|
|
monkeypatch,
|
|
) -> None:
|
|
_patch_scope(monkeypatch)
|
|
|
|
class LegacyClient:
|
|
async def list_wiki_pages(self, *args, **kwargs) -> dict:
|
|
raise AssertionError("legacy directory fallback must not run")
|
|
|
|
async def get_wiki_page(self, *args, **kwargs) -> dict:
|
|
raise AssertionError("legacy page fallback must not run")
|
|
|
|
monkeypatch.setattr(tools, "_client", LegacyClient)
|
|
payload = json.loads(
|
|
await tools._wiki_search_impl(
|
|
"needle",
|
|
{
|
|
"knowledge_retrieval_mode": "smart_reasoning",
|
|
"llmwiki_knowledge_base_ids": ["mapping-1"],
|
|
},
|
|
)
|
|
)
|
|
|
|
assert payload["ok"] is True
|
|
assert payload["native_regex_supported"] is False
|
|
assert payload["unsupported_knowledge_base_count"] == 1
|
|
assert payload["results"] == []
|
|
assert payload["warnings"] == []
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize("status_code", [404, 405])
|
|
async def test_wiki_search_skips_unsupported_native_regex_without_fallback(
|
|
monkeypatch,
|
|
status_code: int,
|
|
) -> None:
|
|
_patch_scope(monkeypatch)
|
|
|
|
class Client:
|
|
async def search_wiki_pages(self, *args, **kwargs) -> list[dict]:
|
|
raise WeKnoraError("route unavailable", status_code=status_code)
|
|
|
|
async def list_wiki_pages(self, *args, **kwargs) -> dict:
|
|
raise AssertionError("unsupported native search must not enumerate Wiki pages")
|
|
|
|
async def get_wiki_page(self, *args, **kwargs) -> dict:
|
|
raise AssertionError("unsupported native search must not read page bodies")
|
|
|
|
monkeypatch.setattr(tools, "_client", Client)
|
|
payload = json.loads(
|
|
await tools._wiki_search_impl(
|
|
"星河工程",
|
|
{
|
|
"knowledge_retrieval_mode": "smart_reasoning",
|
|
"llmwiki_knowledge_base_ids": ["mapping-1"],
|
|
},
|
|
)
|
|
)
|
|
|
|
assert payload["ok"] is True
|
|
assert payload["result_count"] == 0
|
|
assert payload["native_regex_supported"] is False
|
|
assert payload["unsupported_knowledge_base_count"] == 1
|
|
assert payload["partial"] is False
|
|
assert payload["warnings"] == []
|
|
assert payload["coverage"][0]["status"] == "unsupported"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_native_wiki_search_result_can_be_read_with_markdown_shape(
|
|
monkeypatch,
|
|
) -> None:
|
|
_patch_scope(monkeypatch)
|
|
|
|
class EmptyMirror:
|
|
async def list_index_status(self, mapping_ids: list[str]) -> list[dict]:
|
|
return []
|
|
|
|
class Client:
|
|
search_calls: list[tuple[str, str, int]] = []
|
|
|
|
async def search_wiki_pages(
|
|
self,
|
|
knowledge_base_id: str,
|
|
query: str,
|
|
*,
|
|
limit: int,
|
|
) -> list[dict]:
|
|
self.search_calls.append((knowledge_base_id, query, limit))
|
|
return [
|
|
{
|
|
"slug": "news/xinghe",
|
|
"title": "国内科技新闻",
|
|
"description": "科技报道汇总",
|
|
"markdown": "星河工程完成关键节点验收。",
|
|
"status": "published",
|
|
}
|
|
]
|
|
|
|
async def list_wiki_pages(self, *args, **kwargs) -> dict:
|
|
raise AssertionError("native search must not enumerate the directory")
|
|
|
|
async def get_wiki_page(self, knowledge_base_id: str, slug: str) -> dict:
|
|
assert (knowledge_base_id, slug) == ("remote-1", "news/xinghe")
|
|
return {
|
|
"slug": slug,
|
|
"title": "国内科技新闻",
|
|
"description": "科技报道汇总",
|
|
"markdown": "星河工程完成关键节点验收。参见 [[news/验收公告]]。",
|
|
"out_links": ["news/承建单位"],
|
|
"status": "published",
|
|
}
|
|
|
|
client = Client()
|
|
monkeypatch.setattr(tools, "get_local_wiki_index_store", EmptyMirror)
|
|
monkeypatch.setattr(tools, "_client", lambda: client)
|
|
|
|
payload = json.loads(
|
|
await tools._wiki_search_impl(
|
|
"星河工程",
|
|
{
|
|
"knowledge_retrieval_mode": "smart_reasoning",
|
|
"llmwiki_knowledge_base_ids": ["mapping-1"],
|
|
},
|
|
)
|
|
)
|
|
assert payload["results"][0]["matched_fields"] == ["content"]
|
|
assert client.search_calls == [("remote-1", "星河工程", 50)]
|
|
|
|
read = json.loads(
|
|
await tools._wiki_read_impl(
|
|
payload["results"][0]["page_ref"],
|
|
{
|
|
"knowledge_retrieval_mode": "smart_reasoning",
|
|
"llmwiki_knowledge_base_ids": ["mapping-1"],
|
|
},
|
|
)
|
|
)
|
|
assert read["results"][0]["content_markdown"] == "星河工程完成关键节点验收。参见 [[news/验收公告]]。"
|
|
assert read["results"][0]["summary"] == "科技报道汇总"
|
|
assert read["results"][0]["links"] == ["news/承建单位", "news/验收公告"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_local_wiki_read_exposes_bounded_graph_links_for_adaptive_expansion(
|
|
monkeypatch,
|
|
) -> None:
|
|
_patch_scope(monkeypatch)
|
|
|
|
class LocalIndex:
|
|
async def list_index_status(self, mapping_ids: list[str]) -> list[dict]:
|
|
assert mapping_ids == ["mapping-1"]
|
|
return [
|
|
{
|
|
"knowledge_base_mapping_id": "mapping-1",
|
|
"state": "idle",
|
|
"last_completed_at": "2026-10-14T00:00:00Z",
|
|
"remote_page_count": 2,
|
|
"local_page_count": 2,
|
|
"failed_page_count": 0,
|
|
"last_error": None,
|
|
"last_sync_result": {"status": "completed"},
|
|
}
|
|
]
|
|
|
|
async def get_page(self, mapping_id: str, slug: str) -> dict:
|
|
assert (mapping_id, slug) == ("mapping-1", "news/xinghe")
|
|
return {
|
|
"slug": slug,
|
|
"title": "星河工程建设进展",
|
|
"aliases": ["星河计划"],
|
|
"content_md": "已完成节点验收。",
|
|
"status": "published",
|
|
}
|
|
|
|
async def get_adjacent_pages(
|
|
self,
|
|
mapping_ids: list[str],
|
|
seed_slugs: dict[str, list[str]],
|
|
*,
|
|
direction: str,
|
|
limit: int,
|
|
include_drafts: bool,
|
|
) -> list[dict]:
|
|
assert mapping_ids == ["mapping-1"]
|
|
assert seed_slugs == {"mapping-1": ["news/xinghe"]}
|
|
assert direction == "both"
|
|
assert limit == 6
|
|
assert include_drafts is False
|
|
return [
|
|
{
|
|
"neighbor_slug": "entities/tianheng",
|
|
"direction": "outgoing",
|
|
"page": {
|
|
"slug": "entities/tianheng",
|
|
"title": "天衡建设集团",
|
|
},
|
|
}
|
|
]
|
|
|
|
class Client:
|
|
async def get_wiki_page(self, *_args, **_kwargs):
|
|
raise AssertionError("complete local mirror must serve the page")
|
|
|
|
monkeypatch.setattr(tools, "get_local_wiki_index_store", LocalIndex)
|
|
monkeypatch.setattr(tools, "_client", Client)
|
|
|
|
payload = json.loads(
|
|
await tools._wiki_read_impl(
|
|
tools._page_ref("mapping-1", "news/xinghe"),
|
|
{
|
|
"knowledge_retrieval_mode": "smart_reasoning",
|
|
"llmwiki_knowledge_base_ids": ["mapping-1"],
|
|
},
|
|
)
|
|
)
|
|
|
|
assert payload["channel"] == "local_wiki_mirror"
|
|
assert payload["results"][0]["aliases"] == ["星河计划"]
|
|
assert payload["results"][0]["links"] == [
|
|
{
|
|
"slug": "entities/tianheng",
|
|
"title": "天衡建设集团",
|
|
"direction": "outgoing",
|
|
}
|
|
]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_wiki_read_hides_expansion_links_when_database_setting_is_disabled(
|
|
monkeypatch,
|
|
) -> None:
|
|
_patch_scope(monkeypatch)
|
|
|
|
class DisabledMappingStore(_MappingStore):
|
|
async def get_authorized(self, *args, **kwargs) -> dict | None:
|
|
row = await super().get_authorized(*args, **kwargs)
|
|
return {
|
|
**(row or {}),
|
|
"retrieval_settings": {
|
|
"entity_expansion_enabled": False,
|
|
"max_related_terms": 6,
|
|
"statistics_enabled": True,
|
|
},
|
|
}
|
|
|
|
class LocalIndex:
|
|
async def list_index_status(self, _mapping_ids: list[str]) -> list[dict]:
|
|
return [
|
|
{
|
|
"knowledge_base_mapping_id": "mapping-1",
|
|
"state": "idle",
|
|
"last_completed_at": "2026-10-14T00:00:00Z",
|
|
"remote_page_count": 1,
|
|
"local_page_count": 1,
|
|
"failed_page_count": 0,
|
|
"last_error": None,
|
|
"last_sync_result": {"status": "completed"},
|
|
}
|
|
]
|
|
|
|
async def get_page(self, _mapping_id: str, slug: str) -> dict:
|
|
return {
|
|
"slug": slug,
|
|
"title": "星河工程建设进展",
|
|
"content_md": "参见 [[entities/tianheng]]。",
|
|
"out_links": ["entities/tianheng"],
|
|
"status": "published",
|
|
}
|
|
|
|
async def get_adjacent_pages(self, *_args, **_kwargs) -> list[dict]:
|
|
raise AssertionError("disabled expansion must not query the Wiki graph")
|
|
|
|
monkeypatch.setattr(tools, "get_llmwiki_store", DisabledMappingStore)
|
|
monkeypatch.setattr(tools, "get_local_wiki_index_store", LocalIndex)
|
|
|
|
payload = json.loads(
|
|
await tools._wiki_read_impl(
|
|
tools._page_ref("mapping-1", "news/xinghe"),
|
|
{
|
|
"knowledge_retrieval_mode": "smart_reasoning",
|
|
"llmwiki_knowledge_base_ids": ["mapping-1"],
|
|
},
|
|
)
|
|
)
|
|
|
|
assert payload["results"][0]["links"] == []
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_wiki_regex_does_not_use_a_stale_local_mirror(
|
|
monkeypatch,
|
|
) -> None:
|
|
_patch_scope(monkeypatch)
|
|
|
|
class StaleMirror:
|
|
def __init__(self) -> None:
|
|
raise AssertionError("smart Wiki regex must not inspect the local mirror")
|
|
|
|
class Client:
|
|
async def search_wiki_pages(
|
|
self,
|
|
knowledge_base_id: str,
|
|
query: str,
|
|
*,
|
|
limit: int,
|
|
) -> list[dict]:
|
|
assert (knowledge_base_id, query, limit) == (
|
|
"remote-1",
|
|
"星河工程",
|
|
50,
|
|
)
|
|
return [
|
|
{
|
|
"slug": "news/new",
|
|
"title": "新报道",
|
|
"content": "星河工程发布最新进展。",
|
|
"status": "published",
|
|
}
|
|
]
|
|
|
|
monkeypatch.setattr(tools, "get_local_wiki_index_store", StaleMirror)
|
|
monkeypatch.setattr(tools, "_client", Client)
|
|
payload = json.loads(
|
|
await tools._wiki_search_impl(
|
|
"星河工程",
|
|
{
|
|
"knowledge_retrieval_mode": "smart_reasoning",
|
|
"llmwiki_knowledge_base_ids": ["mapping-1"],
|
|
},
|
|
)
|
|
)
|
|
|
|
assert payload["result_count"] == 1
|
|
assert payload["results"][0]["slug"] == "news/new"
|
|
assert payload["channel"] == "provider_native_regex"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_wiki_provider_all_failures_return_error_and_partial_is_warning(
|
|
monkeypatch,
|
|
) -> None:
|
|
_patch_scope(monkeypatch, multi=True)
|
|
|
|
class AllFailedClient:
|
|
async def search_wiki_pages(self, *args, **kwargs) -> list[dict]:
|
|
raise WeKnoraError("down", status_code=503)
|
|
|
|
monkeypatch.setattr(tools, "_client", AllFailedClient)
|
|
failed = json.loads(
|
|
await tools._wiki_search_impl(
|
|
"needle",
|
|
{
|
|
"knowledge_retrieval_mode": "smart_reasoning",
|
|
"llmwiki_knowledge_base_ids": ["mapping-1", "mapping-2"],
|
|
},
|
|
)
|
|
)
|
|
assert failed["ok"] is False
|
|
assert failed["error"]["code"] == "PROVIDER_UNAVAILABLE"
|
|
|
|
class PartiallyFailedClient:
|
|
async def search_wiki_pages(
|
|
self,
|
|
knowledge_base_id: str,
|
|
query: str,
|
|
*,
|
|
limit: int,
|
|
) -> list[dict]:
|
|
if knowledge_base_id == "remote-2":
|
|
raise WeKnoraError("down", status_code=503)
|
|
return [{"slug": "ok", "title": "needle", "content": "正文"}]
|
|
|
|
monkeypatch.setattr(tools, "_client", PartiallyFailedClient)
|
|
partial = json.loads(
|
|
await tools._wiki_search_impl(
|
|
"needle",
|
|
{
|
|
"knowledge_retrieval_mode": "smart_reasoning",
|
|
"llmwiki_knowledge_base_ids": ["mapping-1", "mapping-2"],
|
|
},
|
|
)
|
|
)
|
|
assert partial["ok"] is True
|
|
assert partial["partial"] is True
|
|
assert partial["warnings"] == [
|
|
{"code": "PROVIDER_WIKI_SEARCH_FAILED", "knowledge_base": "制度库2"}
|
|
]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_client_grep_compatibility_fallback_is_marked_and_fair() -> None:
|
|
class Client(WeKnoraClient):
|
|
async def _request(self, *args, **kwargs):
|
|
raise WeKnoraError("missing", status_code=404)
|
|
|
|
async def list_documents(
|
|
self,
|
|
knowledge_base_id: str,
|
|
*,
|
|
page: int,
|
|
page_size: int,
|
|
) -> dict:
|
|
count = 30 if knowledge_base_id == "remote-1" else 1
|
|
rows = [
|
|
{"id": f"{knowledge_base_id}-doc-{index}", "title": "needle title"}
|
|
for index in range(count)
|
|
]
|
|
return {"items": rows, "total": count}
|
|
|
|
async def list_chunks(
|
|
self,
|
|
knowledge_id: str,
|
|
*,
|
|
page: int,
|
|
page_size: int,
|
|
) -> dict:
|
|
return {
|
|
"items": [
|
|
{
|
|
"id": f"{knowledge_id}-chunk",
|
|
"content": f"needle from {knowledge_id}",
|
|
"chunk_index": 0,
|
|
}
|
|
],
|
|
"total": 1,
|
|
}
|
|
|
|
client = Client(base_url="http://weknora.test")
|
|
payload = await client.grep_chunks("needle", ["remote-1", "remote-2"])
|
|
|
|
assert payload["compatibility_fallback"] is True
|
|
assert payload["contract"] == "weknora-existing-api-compat-v1"
|
|
assert payload["successful_knowledge_base_ids"] == ["remote-1", "remote-2"]
|
|
assert {item["knowledge_base_id"] for item in payload["items"]} == {
|
|
"remote-1",
|
|
"remote-2",
|
|
}
|
|
assert payload["warnings"][0]["code"] == "WEKNORA_GREP_COMPATIBILITY_FALLBACK"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_client_grep_prefers_single_cross_scope_provider_request() -> None:
|
|
class Client(WeKnoraClient):
|
|
calls: list[tuple[str, str, dict]] = []
|
|
|
|
async def _request(self, method: str, path: str, *, json=None, **kwargs):
|
|
self.calls.append((method, path, json))
|
|
return {
|
|
"success": True,
|
|
"data": {
|
|
"contract": "weknora-v0.7.2-grep-v1",
|
|
"chunk_results": [
|
|
{
|
|
"id": "chunk-2",
|
|
"knowledge_id": "doc-2",
|
|
"knowledge_base_id": "remote-2",
|
|
"knowledge_title": "第二库",
|
|
"content": "needle",
|
|
}
|
|
],
|
|
},
|
|
}
|
|
|
|
client = Client(base_url="http://weknora.test")
|
|
payload = await client.grep_chunks("needle", ["remote-1", "remote-2"])
|
|
|
|
assert client.calls == [
|
|
(
|
|
"POST",
|
|
"/api/v1/chunks/grep",
|
|
{
|
|
"query": "needle",
|
|
"knowledge_base_ids": ["remote-1", "remote-2"],
|
|
},
|
|
)
|
|
]
|
|
assert payload["compatibility_fallback"] is False
|
|
assert payload["items"][0]["knowledge_base_id"] == "remote-2"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_client_grep_compatibility_all_scope_failure_raises() -> None:
|
|
class Client(WeKnoraClient):
|
|
async def _request(self, *args, **kwargs):
|
|
raise WeKnoraError("missing", status_code=405)
|
|
|
|
async def list_documents(self, *args, **kwargs) -> dict:
|
|
raise WeKnoraError("down")
|
|
|
|
client = Client(base_url="http://weknora.test")
|
|
with pytest.raises(WeKnoraError):
|
|
await client.grep_chunks("needle", ["remote-1", "remote-2"])
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_chunk_grep_all_provider_failure_returns_error(monkeypatch) -> None:
|
|
_patch_scope(monkeypatch)
|
|
|
|
class Client:
|
|
async def grep_chunks(self, query: str, knowledge_base_ids: list[str]) -> dict:
|
|
raise WeKnoraError("down")
|
|
|
|
monkeypatch.setattr(tools, "_client", Client)
|
|
payload = json.loads(
|
|
await tools._chunk_regex_impl(
|
|
"needle",
|
|
{
|
|
"knowledge_retrieval_mode": "smart_reasoning",
|
|
"llmwiki_knowledge_base_ids": ["mapping-1"],
|
|
},
|
|
)
|
|
)
|
|
assert payload["ok"] is False
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_chunk_grep_partial_failure_is_structured_and_sanitized(monkeypatch) -> None:
|
|
_patch_scope(monkeypatch, multi=True)
|
|
|
|
class Client:
|
|
async def grep_chunks(self, query: str, knowledge_base_ids: list[str]) -> dict:
|
|
return {
|
|
"items": [
|
|
{
|
|
"chunk_id": "chunk-1",
|
|
"knowledge_id": "doc-1",
|
|
"knowledge_base_id": "remote-1",
|
|
"content": "needle",
|
|
}
|
|
],
|
|
"contract": "weknora-existing-api-compat-v1",
|
|
"compatibility_fallback": True,
|
|
"partial": True,
|
|
"warnings": [
|
|
{"code": "PROVIDER_SCOPE_FAILED", "knowledge_base_id": "remote-2"}
|
|
],
|
|
"successful_knowledge_base_ids": ["remote-1"],
|
|
"failed_knowledge_base_ids": ["remote-2"],
|
|
}
|
|
|
|
monkeypatch.setattr(tools, "_client", Client)
|
|
payload = json.loads(
|
|
await tools._chunk_regex_impl(
|
|
"needle",
|
|
{
|
|
"knowledge_retrieval_mode": "smart_reasoning",
|
|
"llmwiki_knowledge_base_ids": ["mapping-1", "mapping-2"],
|
|
},
|
|
)
|
|
)
|
|
assert payload["ok"] is True
|
|
assert payload["partial"] is True
|
|
assert payload["compatibility_fallback"] is True
|
|
assert payload["warnings"] == [
|
|
{"code": "PROVIDER_SCOPE_FAILED", "knowledge_base": "制度库2"}
|
|
]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_wiki_read_continuation_reconstructs_long_page_and_binds_cursor(
|
|
monkeypatch,
|
|
) -> None:
|
|
_patch_scope(monkeypatch)
|
|
content = "# 第一章\n\n" + "甲" * 35_000 + "\n\n# 第二章\n\n" + "乙" * 35_000
|
|
|
|
class LocalIndex:
|
|
async def list_index_status(self, mapping_ids: list[str]) -> list[dict]:
|
|
return [
|
|
{
|
|
"knowledge_base_mapping_id": "mapping-1",
|
|
"state": "idle",
|
|
"last_completed_at": "2026-10-14T00:00:00Z",
|
|
"remote_page_count": 1,
|
|
"local_page_count": 1,
|
|
"failed_page_count": 0,
|
|
"last_error": None,
|
|
"last_sync_result": {"status": "completed"},
|
|
}
|
|
]
|
|
|
|
async def get_page(self, mapping_id: str, slug: str) -> dict:
|
|
assert (mapping_id, slug) == ("mapping-1", "long-page")
|
|
return {"title": "长文", "slug": slug, "content_md": content}
|
|
|
|
monkeypatch.setattr(tools, "get_local_wiki_index_store", LocalIndex)
|
|
page_ref = tools._page_ref("mapping-1", "long-page")
|
|
cursor = ""
|
|
parts: list[str] = []
|
|
for _ in range(10):
|
|
payload = json.loads(
|
|
await tools._wiki_read_impl(
|
|
page_ref,
|
|
{
|
|
"knowledge_retrieval_mode": "smart_reasoning",
|
|
"llmwiki_knowledge_base_ids": ["mapping-1"],
|
|
},
|
|
cursor=cursor,
|
|
)
|
|
)
|
|
page = payload["results"][0]
|
|
parts.append(page["content_markdown"])
|
|
if page["complete"]:
|
|
assert page["next_cursor"] is None
|
|
break
|
|
assert page["content_truncated"] is True
|
|
cursor = page["next_cursor"]
|
|
else:
|
|
pytest.fail("long Wiki page did not complete")
|
|
|
|
assert "".join(parts) == content
|
|
other_page_ref = tools._page_ref("mapping-1", "other-page")
|
|
with pytest.raises(tools.SmartKnowledgeToolError) as caught:
|
|
tools._decode_continuation_cursor(cursor, other_page_ref)
|
|
assert caught.value.code == "INVALID_CONTINUATION_CURSOR"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_deep_read_rejects_cross_scope_ids_and_allows_second_mapping(monkeypatch) -> None:
|
|
_patch_scope(monkeypatch)
|
|
|
|
class UnauthorizedClient:
|
|
async def get_chunk(self, chunk_id: str) -> dict:
|
|
return {"id": chunk_id, "knowledge_base_id": "remote-2"}
|
|
|
|
async def get_document(self, knowledge_id: str) -> dict:
|
|
return {"id": knowledge_id, "knowledge_base_id": "remote-2"}
|
|
|
|
monkeypatch.setattr(tools, "_client", UnauthorizedClient)
|
|
context = {
|
|
"knowledge_retrieval_mode": "smart_reasoning",
|
|
"llmwiki_knowledge_base_ids": ["mapping-1"],
|
|
}
|
|
with pytest.raises(tools.SmartKnowledgeToolError) as chunk_error:
|
|
await tools._document_chunks_impl(
|
|
"foreign-chunk",
|
|
"chunk_id",
|
|
context,
|
|
page=1,
|
|
page_size=20,
|
|
)
|
|
assert chunk_error.value.code == "CHUNK_NOT_FOUND"
|
|
with pytest.raises(tools.SmartKnowledgeToolError) as document_error:
|
|
await tools._document_chunks_impl(
|
|
"foreign-doc",
|
|
"knowledge_id",
|
|
context,
|
|
page=1,
|
|
page_size=20,
|
|
)
|
|
assert document_error.value.code == "DOCUMENT_NOT_FOUND"
|
|
|
|
_patch_scope(monkeypatch, multi=True)
|
|
|
|
class AuthorizedSecondClient:
|
|
async def get_document(self, knowledge_id: str) -> dict:
|
|
return {"id": knowledge_id, "knowledge_base_id": "remote-2"}
|
|
|
|
async def list_chunks(
|
|
self,
|
|
knowledge_id: str,
|
|
*,
|
|
page: int,
|
|
page_size: int,
|
|
) -> dict:
|
|
return {
|
|
"items": [{"id": "chunk-2", "content": "第二库正文"}],
|
|
"total": 1,
|
|
}
|
|
|
|
monkeypatch.setattr(tools, "_client", AuthorizedSecondClient)
|
|
allowed = json.loads(
|
|
await tools._document_chunks_impl(
|
|
"doc-2",
|
|
"knowledge_id",
|
|
{
|
|
"knowledge_retrieval_mode": "smart_reasoning",
|
|
"llmwiki_knowledge_base_ids": ["mapping-1", "mapping-2"],
|
|
},
|
|
page=1,
|
|
page_size=20,
|
|
)
|
|
)
|
|
assert allowed["ok"] is True
|
|
assert allowed["results"][0]["content"] == "第二库正文"
|
|
assert allowed["results"][0]["knowledge_base_id"] == "mapping-2"
|
|
|
|
|
|
def test_regex_execution_has_a_hard_timeout() -> None:
|
|
expression = tools._safe_regex("(a|aa)+$")
|
|
with pytest.raises(tools.SmartKnowledgeToolError) as caught:
|
|
tools._regex_search(expression, "a" * 10_000 + "!")
|
|
assert caught.value.code == "REGEX_TIMEOUT"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize("assistant_id", ["deep-research-planner", "deep-research-collector", "custom-agent"])
|
|
async def test_research_smart_scope_cannot_access_unauthorized_library(monkeypatch, assistant_id):
|
|
from app.gateway import llmwiki_rag
|
|
|
|
request = _request(None)
|
|
request.app.state.llmwiki_store = _MappingStore()
|
|
monkeypatch.setattr(llmwiki_rag, "get_resolved_llmwiki_runtime", lambda _: SimpleNamespace(weknora_enabled=True))
|
|
monkeypatch.setattr(llmwiki_rag, "_agent_bound_ids", lambda _: [])
|
|
|
|
async def actor(_request):
|
|
return "user-1", False
|
|
|
|
monkeypatch.setattr(llmwiki_rag, "_actor", actor)
|
|
with pytest.raises(HTTPException) as caught:
|
|
await llmwiki_rag.build_llmwiki_smart_reasoning_context(
|
|
request=request,
|
|
body=SimpleNamespace(context={"llmwiki_knowledge_base_ids": ["not-authorized"]}, metadata={}, assistant_id=assistant_id),
|
|
graph_input={"messages": [HumanMessage(content="test")]},
|
|
)
|
|
assert caught.value.status_code == 404
|
|
|
|
|
|
@pytest.mark.parametrize("agent_id", ["deep-research-planner", "deep-research-collector"])
|
|
def test_builtin_research_tools_use_caller_authorized_scope(agent_id):
|
|
assert tools._allowed_agent_ids({"agent_id": agent_id}) is None
|
|
from deerflow.tools.builtins.llmwiki_search_tool import _agent_allowed_ids
|
|
|
|
assert _agent_allowed_ids({"agent_id": agent_id}) is None
|