1212 lines
42 KiB
Python
1212 lines
42 KiB
Python
from __future__ import annotations
|
||
|
||
import asyncio
|
||
import json
|
||
from types import SimpleNamespace
|
||
from typing import Any
|
||
|
||
import httpx
|
||
import pytest
|
||
|
||
from app.gateway import wiki_directory_constraints as constraints
|
||
from app.gateway.wiki_directory_constraints import (
|
||
ensure_directory_paths,
|
||
ensure_fixed_parent_directories,
|
||
normalize_directory_paths,
|
||
normalize_parent_directories,
|
||
reconcile_fixed_parent_directories,
|
||
)
|
||
from deerflow.integrations.weknora.client import WeKnoraClient
|
||
|
||
|
||
class FakeConflict(RuntimeError):
|
||
status_code = 409
|
||
|
||
|
||
class FakeWeKnora:
|
||
def __init__(
|
||
self,
|
||
*,
|
||
folders: list[dict[str, Any]] | None = None,
|
||
pages: list[dict[str, Any]] | None = None,
|
||
folder_envelope: str = "folders",
|
||
) -> None:
|
||
self.folders = [dict(item) for item in (folders or [])]
|
||
self.pages = [dict(item) for item in (pages or [])]
|
||
self.folder_envelope = folder_envelope
|
||
self.calls: list[tuple[Any, ...]] = []
|
||
self._next_id = 1
|
||
|
||
async def list_wiki_folders(
|
||
self,
|
||
knowledge_base_id: str,
|
||
*,
|
||
parent_id: str = "",
|
||
page_types: list[str] | None = None,
|
||
) -> Any:
|
||
self.calls.append(("list_folders", knowledge_base_id, parent_id))
|
||
rows = [dict(item) for item in self.folders if str(item.get("parent_id") or "") == parent_id]
|
||
if self.folder_envelope == "data-list":
|
||
return {"data": rows}
|
||
if self.folder_envelope == "data-folders":
|
||
return {"data": {"folders": rows}}
|
||
return {"parent_id": parent_id, "folders": rows}
|
||
|
||
async def create_wiki_folder(self, knowledge_base_id: str, *, name: str, parent_id: str = "") -> dict[str, Any]:
|
||
self.calls.append(("create_folder", knowledge_base_id, name, parent_id))
|
||
if any(str(item.get("parent_id") or "") == parent_id and item.get("name") == name for item in self.folders):
|
||
raise FakeConflict("folder already exists")
|
||
folder = {"id": f"created-{self._next_id}", "name": name, "parent_id": parent_id}
|
||
self._next_id += 1
|
||
self.folders.append(folder)
|
||
return dict(folder)
|
||
|
||
async def update_wiki_folder(self, knowledge_base_id: str, folder_id: str, changes: dict[str, Any]) -> dict[str, Any]:
|
||
self.calls.append(("update_folder", knowledge_base_id, folder_id, dict(changes)))
|
||
folder = next(item for item in self.folders if item["id"] == folder_id)
|
||
target_parent = str(changes.get("parent_id") if changes.get("move_parent") else folder.get("parent_id") or "")
|
||
target_name = str(changes.get("name", folder.get("name") or ""))
|
||
if any(
|
||
item["id"] != folder_id
|
||
and str(item.get("parent_id") or "") == target_parent
|
||
and item.get("name") == target_name
|
||
for item in self.folders
|
||
):
|
||
raise FakeConflict("folder conflict")
|
||
if "name" in changes:
|
||
folder["name"] = changes["name"]
|
||
if changes.get("move_parent"):
|
||
folder["parent_id"] = str(changes.get("parent_id") or "")
|
||
return dict(folder)
|
||
|
||
async def delete_wiki_folder(self, knowledge_base_id: str, folder_id: str) -> None:
|
||
self.calls.append(("delete_folder", knowledge_base_id, folder_id))
|
||
self.folders = [item for item in self.folders if item["id"] != folder_id]
|
||
|
||
async def list_wiki_pages(
|
||
self,
|
||
knowledge_base_id: str,
|
||
*,
|
||
page: int,
|
||
page_size: int,
|
||
folder_id: str | None = None,
|
||
) -> dict[str, Any]:
|
||
self.calls.append(("list_pages", knowledge_base_id, page, page_size, folder_id))
|
||
rows = self.pages if folder_id is None else [item for item in self.pages if str(item.get("folder_id") or "") == folder_id]
|
||
start = (page - 1) * page_size
|
||
batch = [dict(item) for item in rows[start : start + page_size]]
|
||
return {"data": {"pages": batch, "total": len(rows), "page": page, "page_size": page_size}}
|
||
|
||
async def move_wiki_page(self, knowledge_base_id: str, slug: str, folder_id: str) -> dict[str, Any]:
|
||
self.calls.append(("move_page", knowledge_base_id, slug, folder_id))
|
||
page = next(item for item in self.pages if item["slug"] == slug)
|
||
page["folder_id"] = folder_id
|
||
return dict(page)
|
||
|
||
async def get_wiki_page(
|
||
self,
|
||
knowledge_base_id: str,
|
||
slug: str,
|
||
) -> dict[str, Any]:
|
||
self.calls.append(("get_page", knowledge_base_id, slug))
|
||
return dict(next(item for item in self.pages if item["slug"] == slug))
|
||
|
||
async def update_wiki_page(
|
||
self,
|
||
knowledge_base_id: str,
|
||
slug: str,
|
||
changes: dict[str, Any],
|
||
) -> dict[str, Any]:
|
||
self.calls.append(("update_page", knowledge_base_id, slug, dict(changes)))
|
||
page = next(item for item in self.pages if item["slug"] == slug)
|
||
# The production WeKnora version binds PUT to a fresh WikiPage and
|
||
# replaces these fields. Missing keys therefore become zero values;
|
||
# modelling PATCH via ``dict.update`` would hide destructive callers.
|
||
defaults: dict[str, Any] = {
|
||
"title": "",
|
||
"content": "",
|
||
"summary": "",
|
||
"page_type": "",
|
||
"status": "",
|
||
"aliases": [],
|
||
"parent_slug": "",
|
||
"folder_id": "",
|
||
"sort_order": 0,
|
||
"source_refs": [],
|
||
"chunk_refs": [],
|
||
"page_metadata": {},
|
||
}
|
||
page.update(
|
||
{
|
||
key: changes.get(key, default)
|
||
for key, default in defaults.items()
|
||
}
|
||
)
|
||
return dict(page)
|
||
|
||
async def delete_wiki_page(self, knowledge_base_id: str, slug: str) -> None:
|
||
self.calls.append(("delete_page", knowledge_base_id, slug))
|
||
self.pages = [item for item in self.pages if item["slug"] != slug]
|
||
|
||
async def rebuild_wiki_links(self, knowledge_base_id: str) -> None:
|
||
self.calls.append(("rebuild_links", knowledge_base_id))
|
||
|
||
|
||
def test_normalize_parent_directories_preserves_order_and_enforces_limits() -> None:
|
||
assert normalize_parent_directories([" 政治 ", "", "经济", "政治", " 军事 "]) == ["政治", "经济", "军事"]
|
||
assert normalize_parent_directories(None) == []
|
||
|
||
for unsafe in ("政治/制度", "军事\\体系", "经济|产业", "社会／文化", "科技｜创新", "含\t制表符"):
|
||
with pytest.raises(ValueError):
|
||
normalize_parent_directories([unsafe])
|
||
with pytest.raises(ValueError, match="100 项"):
|
||
normalize_parent_directories([f"目录{index}" for index in range(101)])
|
||
with pytest.raises(ValueError, match="100 个字符"):
|
||
normalize_parent_directories(["章" * 101])
|
||
with pytest.raises(ValueError, match="6000"):
|
||
normalize_parent_directories(["章" * 100 for _ in range(61)])
|
||
|
||
|
||
def test_normalize_directory_paths_builds_canonical_parent_first_tree() -> None:
|
||
assert normalize_directory_paths(
|
||
[
|
||
[" 政策法规 ", " 国内政策 "],
|
||
["政策法规"],
|
||
["政策法规", "国内政策", "申报流程"],
|
||
["政策法规", "国内政策"],
|
||
]
|
||
) == [
|
||
["政策法规"],
|
||
["政策法规", "国内政策"],
|
||
["政策法规", "国内政策", "申报流程"],
|
||
]
|
||
# Equivalent root spellings keep the first display form for descendants.
|
||
assert normalize_directory_paths([["Ａ"], ["a", "二级"]]) == [
|
||
["A"],
|
||
["A", "二级"],
|
||
]
|
||
|
||
with pytest.raises(ValueError, match="1～3"):
|
||
normalize_directory_paths([["一", "二", "三", "四"]])
|
||
with pytest.raises(ValueError, match="不能是"):
|
||
normalize_directory_paths([[".."]])
|
||
|
||
|
||
def test_focus_classifier_parser_ignores_in_band_reasoning_json() -> None:
|
||
text = (
|
||
'<think>Do not parse this example: {"decisions":{}}.</think>\n'
|
||
'{"decisions":{"page:noise":{"keep":false,"directory":""}}}'
|
||
)
|
||
|
||
assert constraints._parse_focus_classifier_response(
|
||
text,
|
||
{"page:noise"},
|
||
{"政策 / 目标"},
|
||
) == {"page:noise": {"keep": False, "directory": ""}}
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_default_focus_classifier_retries_only_omitted_items(
|
||
monkeypatch: pytest.MonkeyPatch,
|
||
) -> None:
|
||
prompts: list[str] = []
|
||
responses = [
|
||
{
|
||
"decisions": {
|
||
"page:one": {
|
||
"keep": True,
|
||
"directory": "政策 / 目标",
|
||
}
|
||
}
|
||
},
|
||
{
|
||
"decisions": {
|
||
"page:two": {"keep": False, "directory": ""},
|
||
"page:three": {
|
||
"keep": True,
|
||
"directory": "政策 / 目标",
|
||
},
|
||
}
|
||
},
|
||
]
|
||
|
||
class Model:
|
||
async def ainvoke(self, prompt: str) -> SimpleNamespace:
|
||
prompts.append(prompt)
|
||
return SimpleNamespace(
|
||
content=json.dumps(responses.pop(0), ensure_ascii=False)
|
||
)
|
||
|
||
import deerflow.models
|
||
|
||
monkeypatch.setattr(
|
||
deerflow.models,
|
||
"create_chat_model",
|
||
lambda **_: Model(),
|
||
)
|
||
items = [
|
||
{
|
||
"item_id": f"page:{name}",
|
||
"kind": "page",
|
||
"title": name,
|
||
}
|
||
for name in ("one", "two", "three")
|
||
]
|
||
|
||
result = await constraints._default_focus_classifier(
|
||
items,
|
||
"只保留目标",
|
||
["政策 / 目标"],
|
||
)
|
||
|
||
assert result == {
|
||
"page:one": {"keep": True, "directory": "政策 / 目标"},
|
||
"page:two": {"keep": False, "directory": ""},
|
||
"page:three": {"keep": True, "directory": "政策 / 目标"},
|
||
}
|
||
assert len(prompts) == 2
|
||
assert "page:one" in prompts[0]
|
||
assert "page:one" not in prompts[1]
|
||
assert "page:two" in prompts[1]
|
||
assert "page:three" in prompts[1]
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_default_focus_classifier_resolves_stubborn_omission_individually(
|
||
monkeypatch: pytest.MonkeyPatch,
|
||
) -> None:
|
||
prompts: list[str] = []
|
||
responses = [
|
||
{"decisions": {}},
|
||
{"decisions": {}},
|
||
{"keep": False, "directory": ""},
|
||
]
|
||
|
||
class Model:
|
||
async def ainvoke(self, prompt: str) -> SimpleNamespace:
|
||
prompts.append(prompt)
|
||
return SimpleNamespace(
|
||
content=json.dumps(responses.pop(0), ensure_ascii=False)
|
||
)
|
||
|
||
import deerflow.models
|
||
|
||
monkeypatch.setattr(
|
||
deerflow.models,
|
||
"create_chat_model",
|
||
lambda **_: Model(),
|
||
)
|
||
|
||
result = await constraints._default_focus_classifier(
|
||
[{"item_id": "page:noise", "kind": "page", "title": "无关噪声"}],
|
||
"只保留量化目标",
|
||
["政策 / 目标"],
|
||
)
|
||
|
||
assert result == {"page:noise": {"keep": False, "directory": ""}}
|
||
assert len(prompts) == 3
|
||
assert "There is exactly one page" in prompts[-1]
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_ensure_directory_paths_recursively_creates_each_level_once() -> None:
|
||
client = FakeWeKnora()
|
||
paths = [
|
||
["政策法规"],
|
||
["政策法规", "国内政策"],
|
||
["政策法规", "国内政策", "申报流程"],
|
||
]
|
||
|
||
first = await ensure_directory_paths(client, "kb-one", paths)
|
||
second = await ensure_directory_paths(client, "kb-one", paths)
|
||
|
||
assert first == second
|
||
assert list(first) == [
|
||
"政策法规",
|
||
"政策法规 / 国内政策",
|
||
"政策法规 / 国内政策 / 申报流程",
|
||
]
|
||
assert [call for call in client.calls if call[0] == "create_folder"] == [
|
||
("create_folder", "kb-one", "政策法规", ""),
|
||
("create_folder", "kb-one", "国内政策", first["政策法规"]),
|
||
(
|
||
"create_folder",
|
||
"kb-one",
|
||
"申报流程",
|
||
first["政策法规 / 国内政策"],
|
||
),
|
||
]
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_ensure_directory_paths_adopts_empty_folders_after_conflict() -> None:
|
||
"""WeKnora hides empty folders outside its merged page-type view.
|
||
|
||
The first read for each parent also simulates a concurrent-create race:
|
||
list returns no row, create reports 409, and the conflict re-read exposes
|
||
the already-created empty folder.
|
||
"""
|
||
|
||
class EmptyFolderRaceClient:
|
||
def __init__(self) -> None:
|
||
self.folders = [
|
||
{
|
||
"id": "existing-root",
|
||
"name": "国内新闻",
|
||
"parent_id": "",
|
||
},
|
||
{
|
||
"id": "existing-child",
|
||
"name": "政策发布",
|
||
"parent_id": "existing-root",
|
||
},
|
||
]
|
||
self.list_count: dict[str, int] = {}
|
||
self.create_calls: list[tuple[str, str]] = []
|
||
|
||
async def list_wiki_folders(
|
||
self,
|
||
knowledge_base_id: str,
|
||
*,
|
||
parent_id: str = "",
|
||
page_types: list[str] | None = None,
|
||
) -> dict[str, Any]:
|
||
assert knowledge_base_id == "kb-news"
|
||
assert page_types == [
|
||
"summary",
|
||
"entity",
|
||
"concept",
|
||
"index",
|
||
"synthesis",
|
||
"comparison",
|
||
"article",
|
||
]
|
||
count = self.list_count.get(parent_id, 0) + 1
|
||
self.list_count[parent_id] = count
|
||
rows = [
|
||
dict(item)
|
||
for item in self.folders
|
||
if str(item.get("parent_id") or "") == parent_id
|
||
]
|
||
# A stale first read creates the exact list/create race that the
|
||
# conflict recovery path must tolerate.
|
||
if count == 1:
|
||
rows = []
|
||
return {"parent_id": parent_id, "folders": rows}
|
||
|
||
async def create_wiki_folder(
|
||
self,
|
||
knowledge_base_id: str,
|
||
*,
|
||
name: str,
|
||
parent_id: str = "",
|
||
) -> dict[str, Any]:
|
||
assert knowledge_base_id == "kb-news"
|
||
self.create_calls.append((name, parent_id))
|
||
raise FakeConflict("wiki folder name conflict")
|
||
|
||
client = EmptyFolderRaceClient()
|
||
paths = [["国内新闻"], ["国内新闻", "政策发布"]]
|
||
|
||
first = await ensure_directory_paths(client, "kb-news", paths)
|
||
second = await ensure_directory_paths(client, "kb-news", paths)
|
||
|
||
assert first == second == {
|
||
"国内新闻": "existing-root",
|
||
"国内新闻 / 政策发布": "existing-child",
|
||
}
|
||
assert client.create_calls == [
|
||
("国内新闻", ""),
|
||
("政策发布", "existing-root"),
|
||
]
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_ensure_directory_paths_matches_existing_name_by_nfkc_casefold() -> None:
|
||
client = FakeWeKnora(
|
||
folders=[{"id": "root-a", "name": "Ａ", "parent_id": ""}],
|
||
)
|
||
|
||
result = await ensure_directory_paths(client, "kb-one", [["a"]])
|
||
|
||
assert result == {"a": "root-a"}
|
||
assert not [call for call in client.calls if call[0] == "create_folder"]
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_empty_constraints_make_no_remote_calls() -> None:
|
||
client = FakeWeKnora()
|
||
|
||
assert await ensure_fixed_parent_directories(client, "kb-one", None) == {}
|
||
result = await reconcile_fixed_parent_directories(
|
||
client,
|
||
{"weknora_id": "kb-one", "wiki_parent_directories": []},
|
||
)
|
||
|
||
assert result == {
|
||
"roots": {},
|
||
"created_roots": 0,
|
||
"moved_folders": 0,
|
||
"renamed_folders": 0,
|
||
"moved_pages": 0,
|
||
}
|
||
assert client.calls == []
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
@pytest.mark.parametrize("folder_envelope", ["folders", "data-folders", "data-list"])
|
||
async def test_ensure_fixed_roots_rebuilds_only_missing_names(folder_envelope: str) -> None:
|
||
client = FakeWeKnora(
|
||
folders=[{"id": "root-a", "name": "政治", "parent_id": ""}],
|
||
folder_envelope=folder_envelope,
|
||
)
|
||
|
||
first = await ensure_fixed_parent_directories(client, "kb-one", ["政治", "经济"])
|
||
second = await ensure_fixed_parent_directories(client, "kb-one", ["政治", "经济"])
|
||
|
||
assert list(first) == ["政治", "经济"]
|
||
assert first == second
|
||
assert [item["name"] for item in client.folders if not item.get("parent_id")] == ["政治", "经济"]
|
||
assert [call for call in client.calls if call[0] == "create_folder"] == [
|
||
("create_folder", "kb-one", "经济", "")
|
||
]
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_legacy_directory_reference_does_not_move_folders_or_pages() -> None:
|
||
client = FakeWeKnora(
|
||
folders=[
|
||
{"id": "politics", "name": "政治", "parent_id": ""},
|
||
{"id": "existing-defense", "name": "军事", "parent_id": "politics"},
|
||
{"id": "extra-defense", "name": "军事", "parent_id": ""},
|
||
{"id": "extra-defense-child", "name": "装备", "parent_id": "extra-defense"},
|
||
{"id": "extra-market", "name": "市场", "parent_id": ""},
|
||
],
|
||
pages=[
|
||
{"slug": "index", "title": "Wiki Index", "page_type": "index", "folder_id": ""},
|
||
{"slug": "directory", "title": "目录", "page_type": "summary", "folder_id": ""},
|
||
{"slug": "weekly-report", "title": "每周经济报告", "page_type": "summary", "folder_id": ""},
|
||
],
|
||
)
|
||
classified: list[str] = []
|
||
|
||
async def classifier(item: dict[str, Any], roots: list[str]) -> str:
|
||
classified.append(str(item["item_id"]))
|
||
assert roots == ["政治", "经济"]
|
||
if item.get("name") == "军事":
|
||
return "政治"
|
||
return "经济"
|
||
|
||
first = await reconcile_fixed_parent_directories(
|
||
client,
|
||
{"weknora_id": "kb-one", "wiki_parent_directories": ["政治", "经济"]},
|
||
classifier=classifier,
|
||
)
|
||
second = await reconcile_fixed_parent_directories(
|
||
client,
|
||
{"weknora_id": "kb-one", "wiki_parent_directories": ["政治", "经济"]},
|
||
classifier=classifier,
|
||
)
|
||
|
||
assert first["created_roots"] == 0
|
||
assert first["moved_folders"] == 0
|
||
assert first["renamed_folders"] == 0
|
||
assert first["moved_pages"] == 0
|
||
assert second == first
|
||
assert classified == []
|
||
|
||
top_level = {item["name"]: item["id"] for item in client.folders if not item.get("parent_id")}
|
||
assert set(top_level) == {"政治", "军事", "市场"}
|
||
moved_defense = next(item for item in client.folders if item["id"] == "extra-defense")
|
||
assert moved_defense["parent_id"] == ""
|
||
assert moved_defense["name"] == "军事"
|
||
assert next(item for item in client.folders if item["id"] == "extra-defense-child")["parent_id"] == "extra-defense"
|
||
assert next(item for item in client.folders if item["id"] == "extra-market")["parent_id"] == ""
|
||
assert next(item for item in client.pages if item["slug"] == "weekly-report")["folder_id"] == ""
|
||
assert next(item for item in client.pages if item["slug"] == "index")["folder_id"] == ""
|
||
assert next(item for item in client.pages if item["slug"] == "directory")["folder_id"] == ""
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_directory_reference_preserves_article_only_extra_folder() -> None:
|
||
class PageTypeFilteringFolders(FakeWeKnora):
|
||
async def list_wiki_folders(
|
||
self,
|
||
knowledge_base_id: str,
|
||
*,
|
||
parent_id: str = "",
|
||
page_types: list[str] | None = None,
|
||
) -> dict[str, Any]:
|
||
self.calls.append(("list_folders", knowledge_base_id, parent_id))
|
||
allowed_types = set(page_types or [])
|
||
rows = []
|
||
for folder in self.folders:
|
||
if str(folder.get("parent_id") or "") != parent_id:
|
||
continue
|
||
folder_id = str(folder.get("id") or "")
|
||
if any(
|
||
str(page.get("folder_id") or "") == folder_id
|
||
and str(page.get("page_type") or "") in allowed_types
|
||
for page in self.pages
|
||
):
|
||
rows.append(dict(folder))
|
||
return {"parent_id": parent_id, "folders": rows}
|
||
|
||
client = PageTypeFilteringFolders(
|
||
folders=[
|
||
{
|
||
"id": "manual-extra",
|
||
"name": "人工目录",
|
||
"parent_id": "",
|
||
}
|
||
],
|
||
pages=[
|
||
{
|
||
"slug": "manual-article",
|
||
"title": "人工维护页",
|
||
"content": "由管理员手工创建。",
|
||
"page_type": "article",
|
||
"folder_id": "manual-extra",
|
||
}
|
||
],
|
||
)
|
||
|
||
result = await reconcile_fixed_parent_directories(
|
||
client,
|
||
{
|
||
"weknora_id": "kb-one",
|
||
"wiki_directory_paths": [["允许目录"]],
|
||
"wiki_focus_instructions": "",
|
||
},
|
||
)
|
||
|
||
assert result["moved_pages"] == 0
|
||
assert {folder["name"] for folder in client.folders} == {"人工目录"}
|
||
assert client.pages[0]["folder_id"] == "manual-extra"
|
||
assert client.calls == []
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_directory_reference_does_not_limit_strict_focus_filter() -> None:
|
||
client = FakeWeKnora(
|
||
folders=[
|
||
{"id": "root-policy", "name": "政策", "parent_id": ""},
|
||
{"id": "leaf-target", "name": "目标", "parent_id": "root-policy"},
|
||
{"id": "extra-root", "name": "石窟寺", "parent_id": ""},
|
||
{"id": "extra-child", "name": "保护利用", "parent_id": "extra-root"},
|
||
],
|
||
pages=[
|
||
{
|
||
"slug": "concept/charging-target",
|
||
"title": "2027年充电设施目标",
|
||
"summary": "设施总量与公共充电容量",
|
||
"page_type": "concept",
|
||
"folder_id": "extra-root",
|
||
"source_refs": ["doc-a"],
|
||
},
|
||
{
|
||
"slug": "concept/cave-temple",
|
||
"title": "石窟寺保护利用",
|
||
"summary": "文物保护背景材料",
|
||
"page_type": "concept",
|
||
"folder_id": "extra-child",
|
||
"source_refs": ["doc-b"],
|
||
},
|
||
{
|
||
"slug": "extra-directory",
|
||
"title": "石窟寺目录",
|
||
"page_type": "directory",
|
||
"folder_id": "extra-root",
|
||
},
|
||
],
|
||
)
|
||
|
||
focus_calls = 0
|
||
|
||
async def focus_classifier(
|
||
items: list[dict[str, Any]],
|
||
focus: str,
|
||
allowed_paths: list[str],
|
||
) -> dict[str, dict[str, Any]]:
|
||
nonlocal focus_calls
|
||
focus_calls += 1
|
||
assert focus == "只提取明确年份和充电设施量化目标"
|
||
assert allowed_paths == []
|
||
assert {item["item_id"] for item in items} == {
|
||
"page:concept/charging-target",
|
||
"page:concept/cave-temple",
|
||
}
|
||
return {
|
||
"page:concept/charging-target": {
|
||
"keep": True,
|
||
"directory": "",
|
||
},
|
||
"page:concept/cave-temple": {"keep": False, "directory": ""},
|
||
}
|
||
|
||
result = await reconcile_fixed_parent_directories(
|
||
client,
|
||
{
|
||
"weknora_id": "kb-one",
|
||
"wiki_directory_paths": [["政策"], ["政策", "目标"]],
|
||
"wiki_focus_instructions": "只提取明确年份和充电设施量化目标",
|
||
},
|
||
focus_classifier=focus_classifier,
|
||
)
|
||
second = await reconcile_fixed_parent_directories(
|
||
client,
|
||
{
|
||
"weknora_id": "kb-one",
|
||
"wiki_directory_paths": [["政策"], ["政策", "目标"]],
|
||
"wiki_focus_instructions": "只提取明确年份和充电设施量化目标",
|
||
},
|
||
focus_classifier=focus_classifier,
|
||
)
|
||
|
||
assert result["focus_enforced"] is True
|
||
assert result["strict_hierarchy_enforced"] is False
|
||
assert result["moved_pages"] == 0
|
||
assert result["deleted_pages"] == 1
|
||
assert result["deleted_folders"] == 0
|
||
assert {folder["id"] for folder in client.folders} == {
|
||
"root-policy",
|
||
"leaf-target",
|
||
"extra-root",
|
||
"extra-child",
|
||
}
|
||
assert focus_calls == 1
|
||
assert second["moved_pages"] == 0
|
||
assert second["deleted_pages"] == 0
|
||
assert second["deleted_folders"] == 0
|
||
assert [page["slug"] for page in client.pages] == [
|
||
"concept/charging-target",
|
||
"extra-directory",
|
||
]
|
||
kept_page = client.pages[0]
|
||
assert kept_page["folder_id"] == "extra-root"
|
||
marker = kept_page["page_metadata"]["deerflow_generation_constraints"]
|
||
assert marker["schema"] == 1
|
||
assert marker["keep"] is True
|
||
assert marker["directory"] == ""
|
||
assert len([call for call in client.calls if call[0] == "update_page"]) == 1
|
||
assert ("rebuild_links", "kb-one") in client.calls
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_focus_marker_preserves_full_page_across_replacement_put() -> None:
|
||
original = {
|
||
"slug": "concept/data-infrastructure",
|
||
"title": "数据基础设施",
|
||
"content": "到2029年建成主体结构，到2035年形成完善体系。",
|
||
"summary": "建设目标与时间表",
|
||
"page_type": "concept",
|
||
"status": "published",
|
||
"aliases": ["数据底座"],
|
||
"parent_slug": "concept/digital-china",
|
||
"folder_id": "target-folder",
|
||
"sort_order": 7,
|
||
"source_refs": ["doc-one|建设指引"],
|
||
"chunk_refs": ["chunk-one"],
|
||
"page_metadata": {"language": "zh-CN"},
|
||
}
|
||
client = FakeWeKnora(
|
||
folders=[{"id": "target-folder", "name": "建设指引", "parent_id": ""}],
|
||
pages=[original],
|
||
)
|
||
|
||
async def keep_page(
|
||
items: list[dict[str, Any]],
|
||
_: str,
|
||
__: list[str],
|
||
) -> dict[str, dict[str, Any]]:
|
||
return {
|
||
str(items[0]["item_id"]): {
|
||
"keep": True,
|
||
"directory": "建设指引",
|
||
}
|
||
}
|
||
|
||
await reconcile_fixed_parent_directories(
|
||
client,
|
||
{
|
||
"weknora_id": "kb-one",
|
||
"wiki_directory_paths": [["建设指引"]],
|
||
"wiki_focus_instructions": "只保留建设目标和明确时间节点",
|
||
},
|
||
focus_classifier=keep_page,
|
||
)
|
||
|
||
page = client.pages[0]
|
||
for key in (
|
||
"title",
|
||
"content",
|
||
"summary",
|
||
"page_type",
|
||
"status",
|
||
"aliases",
|
||
"parent_slug",
|
||
"folder_id",
|
||
"sort_order",
|
||
"source_refs",
|
||
"chunk_refs",
|
||
):
|
||
assert page[key] == original[key]
|
||
assert page["page_metadata"]["language"] == "zh-CN"
|
||
assert (
|
||
page["page_metadata"]["deerflow_generation_constraints"]["keep"]
|
||
is True
|
||
)
|
||
|
||
update_calls = [call for call in client.calls if call[0] == "update_page"]
|
||
assert len(update_calls) == 1
|
||
payload = update_calls[0][3]
|
||
assert {
|
||
"title",
|
||
"content",
|
||
"summary",
|
||
"page_type",
|
||
"status",
|
||
"aliases",
|
||
"parent_slug",
|
||
"folder_id",
|
||
"sort_order",
|
||
"source_refs",
|
||
"chunk_refs",
|
||
"page_metadata",
|
||
} <= payload.keys()
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_focus_only_constraint_filters_pages_without_changing_directories() -> None:
|
||
client = FakeWeKnora(
|
||
folders=[{"id": "existing-folder", "name": "原有目录", "parent_id": ""}],
|
||
pages=[
|
||
{
|
||
"slug": "keep",
|
||
"title": "2029目标",
|
||
"folder_id": "existing-folder",
|
||
"source_refs": ["doc-one"],
|
||
},
|
||
{
|
||
"slug": "drop",
|
||
"title": "无关人物",
|
||
"folder_id": "existing-folder",
|
||
"source_refs": ["doc-one"],
|
||
},
|
||
]
|
||
)
|
||
|
||
async def focus_classifier(
|
||
items: list[dict[str, Any]],
|
||
focus: str,
|
||
allowed_paths: list[str],
|
||
) -> dict[str, dict[str, Any]]:
|
||
assert focus == "只提取2029年建设目标"
|
||
assert allowed_paths == []
|
||
return {
|
||
"page:keep": {"keep": True, "directory": ""},
|
||
"page:drop": {"keep": False, "directory": ""},
|
||
}
|
||
|
||
result = await reconcile_fixed_parent_directories(
|
||
client,
|
||
{
|
||
"weknora_id": "kb-one",
|
||
"wiki_directory_paths": [],
|
||
"wiki_focus_instructions": "只提取2029年建设目标",
|
||
},
|
||
focus_classifier=focus_classifier,
|
||
)
|
||
|
||
assert [page["slug"] for page in client.pages] == ["keep"]
|
||
assert client.pages[0]["folder_id"] == "existing-folder"
|
||
assert [folder["id"] for folder in client.folders] == ["existing-folder"]
|
||
assert result["deleted_pages"] == 1
|
||
assert result["moved_pages"] == 0
|
||
assert result["deleted_folders"] == 0
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_focus_uses_full_page_body_invalidates_marker_and_protects_manual_pages() -> None:
|
||
class SummaryOnlyListing(FakeWeKnora):
|
||
async def list_wiki_pages(
|
||
self,
|
||
knowledge_base_id: str,
|
||
*,
|
||
page: int,
|
||
page_size: int,
|
||
folder_id: str | None = None,
|
||
) -> dict[str, Any]:
|
||
result = await super().list_wiki_pages(
|
||
knowledge_base_id,
|
||
page=page,
|
||
page_size=page_size,
|
||
folder_id=folder_id,
|
||
)
|
||
for row in result["data"]["pages"]:
|
||
row.pop("content", None)
|
||
return result
|
||
|
||
client = SummaryOnlyListing(
|
||
pages=[
|
||
{
|
||
"slug": "generated",
|
||
"title": "综合说明",
|
||
"summary": "正文包含具体目标",
|
||
"content": "到2029年完成数据基础设施建设。",
|
||
"page_type": "concept",
|
||
"source_refs": ["doc-one"],
|
||
},
|
||
{
|
||
"slug": "manual",
|
||
"title": "人工维护页",
|
||
"summary": "管理员手工保存",
|
||
"content": "与重点无关，但不得由重点约束删除。",
|
||
"page_type": "article",
|
||
},
|
||
]
|
||
)
|
||
classified_content: list[str] = []
|
||
|
||
async def classify(
|
||
items: list[dict[str, Any]],
|
||
_: str,
|
||
__: list[str],
|
||
) -> dict[str, dict[str, Any]]:
|
||
assert [item["item_id"] for item in items] == ["page:generated"]
|
||
content = str(items[0]["content_excerpt"])
|
||
classified_content.append(content)
|
||
return {
|
||
"page:generated": {
|
||
"keep": "2029" in content,
|
||
"directory": "",
|
||
}
|
||
}
|
||
|
||
mapping = {
|
||
"weknora_id": "kb-one",
|
||
"wiki_directory_paths": [],
|
||
"wiki_focus_instructions": "只保留2029年建设目标",
|
||
}
|
||
first = await reconcile_fixed_parent_directories(
|
||
client,
|
||
mapping,
|
||
focus_classifier=classify,
|
||
)
|
||
assert first["deleted_pages"] == 0
|
||
assert classified_content == ["到2029年完成数据基础设施建设。"]
|
||
assert [call for call in client.calls if call[0] == "get_page"] == [
|
||
("get_page", "kb-one", "generated")
|
||
]
|
||
|
||
generated = next(page for page in client.pages if page["slug"] == "generated")
|
||
generated["content"] = "正文现已改为石窟寺保护背景材料。"
|
||
second = await reconcile_fixed_parent_directories(
|
||
client,
|
||
mapping,
|
||
focus_classifier=classify,
|
||
)
|
||
|
||
assert second["deleted_pages"] == 1
|
||
assert classified_content == [
|
||
"到2029年完成数据基础设施建设。",
|
||
"正文现已改为石窟寺保护背景材料。",
|
||
]
|
||
assert [page["slug"] for page in client.pages] == ["manual"]
|
||
assert [call for call in client.calls if call[0] == "get_page"] == [
|
||
("get_page", "kb-one", "generated"),
|
||
("get_page", "kb-one", "generated"),
|
||
]
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_focus_page_details_use_bounded_concurrency() -> None:
|
||
class SlowDetails(FakeWeKnora):
|
||
in_flight = 0
|
||
max_in_flight = 0
|
||
|
||
async def get_wiki_page(
|
||
self,
|
||
knowledge_base_id: str,
|
||
slug: str,
|
||
) -> dict[str, Any]:
|
||
self.in_flight += 1
|
||
self.max_in_flight = max(self.max_in_flight, self.in_flight)
|
||
try:
|
||
await asyncio.sleep(0.001)
|
||
return await super().get_wiki_page(knowledge_base_id, slug)
|
||
finally:
|
||
self.in_flight -= 1
|
||
|
||
client = SlowDetails(
|
||
pages=[
|
||
{
|
||
"slug": f"generated-{index:02d}",
|
||
"title": f"目标 {index}",
|
||
"content": "2029年建设目标",
|
||
"page_type": "concept",
|
||
"source_refs": [f"doc-{index}"],
|
||
}
|
||
for index in range(12)
|
||
]
|
||
)
|
||
|
||
async def keep_all(
|
||
items: list[dict[str, Any]],
|
||
_: str,
|
||
__: list[str],
|
||
) -> dict[str, dict[str, Any]]:
|
||
return {
|
||
str(item["item_id"]): {"keep": True, "directory": ""}
|
||
for item in items
|
||
}
|
||
|
||
await reconcile_fixed_parent_directories(
|
||
client,
|
||
{
|
||
"weknora_id": "kb-one",
|
||
"wiki_directory_paths": [],
|
||
"wiki_focus_instructions": "只保留建设目标",
|
||
},
|
||
focus_classifier=keep_all,
|
||
)
|
||
|
||
assert 1 < client.max_in_flight <= constraints._PAGE_DETAIL_CONCURRENCY
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_incomplete_focus_decisions_do_not_delete_or_move_pages() -> None:
|
||
client = FakeWeKnora(
|
||
folders=[{"id": "root", "name": "目标", "parent_id": ""}],
|
||
pages=[
|
||
{
|
||
"slug": "one",
|
||
"title": "一",
|
||
"folder_id": "",
|
||
"source_refs": ["doc-one"],
|
||
},
|
||
{
|
||
"slug": "two",
|
||
"title": "二",
|
||
"folder_id": "",
|
||
"source_refs": ["doc-two"],
|
||
},
|
||
],
|
||
)
|
||
|
||
async def incomplete(*_: Any) -> dict[str, dict[str, Any]]:
|
||
return {"page:one": {"keep": False, "directory": "目标"}}
|
||
|
||
with pytest.raises(RuntimeError, match="incomplete"):
|
||
await reconcile_fixed_parent_directories(
|
||
client,
|
||
{
|
||
"weknora_id": "kb-one",
|
||
"wiki_directory_paths": [["目标"]],
|
||
"wiki_focus_instructions": "硬约束",
|
||
},
|
||
focus_classifier=incomplete,
|
||
)
|
||
|
||
assert {page["slug"] for page in client.pages} == {"one", "two"}
|
||
assert not [call for call in client.calls if call[0] in {"delete_page", "move_page"}]
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_completed_focus_batches_are_cached_before_a_later_batch_fails(
|
||
monkeypatch: pytest.MonkeyPatch,
|
||
) -> None:
|
||
client = FakeWeKnora(
|
||
folders=[{"id": "leaf", "name": "目标", "parent_id": ""}],
|
||
pages=[
|
||
{
|
||
"slug": "one",
|
||
"title": "2029目标",
|
||
"folder_id": "",
|
||
"source_refs": ["doc-one"],
|
||
},
|
||
{
|
||
"slug": "two",
|
||
"title": "无关噪声",
|
||
"folder_id": "",
|
||
"source_refs": ["doc-two"],
|
||
},
|
||
],
|
||
)
|
||
monkeypatch.setattr(constraints, "_FOCUS_CLASSIFIER_BATCH_SIZE", 1)
|
||
calls: list[list[str]] = []
|
||
|
||
async def fail_second_batch(
|
||
items: list[dict[str, Any]],
|
||
*_: Any,
|
||
**__: Any,
|
||
) -> dict[str, dict[str, Any]]:
|
||
item_ids = [str(item["item_id"]) for item in items]
|
||
calls.append(item_ids)
|
||
if item_ids == ["page:one"]:
|
||
return {
|
||
"page:one": {"keep": True, "directory": "目标"}
|
||
}
|
||
raise RuntimeError("later batch failed")
|
||
|
||
monkeypatch.setattr(
|
||
constraints,
|
||
"_classify_focus_pages",
|
||
fail_second_batch,
|
||
)
|
||
mapping = {
|
||
"weknora_id": "kb-one",
|
||
"wiki_directory_paths": [["目标"]],
|
||
"wiki_focus_instructions": "只保留2029目标",
|
||
}
|
||
|
||
with pytest.raises(RuntimeError, match="later batch failed"):
|
||
await reconcile_fixed_parent_directories(client, mapping)
|
||
|
||
assert calls == [["page:one"], ["page:two"]]
|
||
assert (
|
||
client.pages[0]["page_metadata"]["deerflow_generation_constraints"][
|
||
"keep"
|
||
]
|
||
is True
|
||
)
|
||
assert not [
|
||
call for call in client.calls if call[0] in {"delete_page", "move_page"}
|
||
]
|
||
|
||
calls.clear()
|
||
|
||
async def finish_remaining(
|
||
items: list[dict[str, Any]],
|
||
*_: Any,
|
||
**__: Any,
|
||
) -> dict[str, dict[str, Any]]:
|
||
item_ids = [str(item["item_id"]) for item in items]
|
||
calls.append(item_ids)
|
||
return {"page:two": {"keep": False, "directory": ""}}
|
||
|
||
monkeypatch.setattr(
|
||
constraints,
|
||
"_classify_focus_pages",
|
||
finish_remaining,
|
||
)
|
||
result = await reconcile_fixed_parent_directories(client, mapping)
|
||
|
||
assert calls == [["page:two"]]
|
||
assert result["deleted_pages"] == 1
|
||
assert [page["slug"] for page in client.pages] == ["one"]
|
||
assert client.pages[0]["folder_id"] == ""
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_legacy_multi_root_reference_does_not_invoke_classifier(
|
||
monkeypatch: pytest.MonkeyPatch,
|
||
) -> None:
|
||
client = FakeWeKnora(
|
||
folders=[
|
||
{"id": "root-a", "name": "甲", "parent_id": ""},
|
||
{"id": "root-b", "name": "乙", "parent_id": ""},
|
||
{"id": "extra", "name": "应归乙", "parent_id": ""},
|
||
{"id": "fallback", "name": "未知", "parent_id": ""},
|
||
],
|
||
pages=[{"slug": "page-one", "title": "乙类页面", "page_type": "summary", "folder_id": ""}],
|
||
)
|
||
batches: list[list[str]] = []
|
||
|
||
async def fake_default_classifier(items: list[dict[str, Any]], roots: list[str]) -> dict[str, str]:
|
||
assert roots == ["甲", "乙"]
|
||
batches.append([str(item["item_id"]) for item in items])
|
||
return {
|
||
"folder:extra": "乙",
|
||
"folder:fallback": "不存在的根",
|
||
"page:page-one": "乙",
|
||
}
|
||
|
||
monkeypatch.setattr(constraints, "_default_classifier", fake_default_classifier)
|
||
result = await reconcile_fixed_parent_directories(
|
||
client,
|
||
{"weknora_id": "kb-one", "wiki_parent_directories": ["甲", "乙"]},
|
||
)
|
||
|
||
assert result["moved_folders"] == 0
|
||
assert result["moved_pages"] == 0
|
||
assert batches == []
|
||
assert next(item for item in client.folders if item["id"] == "extra")["parent_id"] == ""
|
||
assert next(item for item in client.folders if item["id"] == "fallback")["parent_id"] == ""
|
||
assert client.pages[0]["folder_id"] == ""
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_weknora_client_folder_and_page_placement_contracts() -> None:
|
||
seen: list[tuple[str, str]] = []
|
||
|
||
async def handler(request: httpx.Request) -> httpx.Response:
|
||
path = request.url.raw_path.decode().split("?", 1)[0]
|
||
seen.append((request.method, path))
|
||
if request.method == "GET" and path.endswith("/wiki/folders"):
|
||
assert request.url.params["parent_id"] == "parent-one"
|
||
assert request.url.params["page_types"] == "entity,concept"
|
||
return httpx.Response(200, json={"data": [{"id": "child", "name": "Child"}]})
|
||
if request.method == "PUT" and path.endswith("/wiki/folders/folder%2Fone"):
|
||
assert json.loads(request.content) == {"parent_id": "root-one", "move_parent": True}
|
||
return httpx.Response(200, json={"data": {"id": "folder/one", "parent_id": "root-one"}})
|
||
if request.method == "DELETE" and path.endswith("/wiki/folders/folder%2Fone"):
|
||
return httpx.Response(204)
|
||
if request.method == "PUT" and path.endswith("/wiki/move-page"):
|
||
assert json.loads(request.content) == {"slug": "entity/acme", "folder_id": "root-one"}
|
||
return httpx.Response(200, json={"data": {"slug": "entity/acme", "folder_id": "root-one"}})
|
||
if request.method == "GET" and path.endswith("/wiki/pages"):
|
||
if request.url.params.get("query") == "unfiltered":
|
||
assert "folder_id" not in request.url.params
|
||
else:
|
||
assert "folder_id" in request.url.params
|
||
assert request.url.params["folder_id"] == ""
|
||
return httpx.Response(200, json={"pages": [], "total": 0, "page": 1, "page_size": 50})
|
||
return httpx.Response(404)
|
||
|
||
async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as http_client:
|
||
client = WeKnoraClient(base_url="http://weknora.local:8080", http_client=http_client)
|
||
folders = await client.list_wiki_folders("kb-one", parent_id="parent-one", page_types=["entity", "concept"])
|
||
updated = await client.update_wiki_folder("kb-one", "folder/one", {"parent_id": "root-one"})
|
||
await client.delete_wiki_folder("kb-one", "folder/one")
|
||
moved = await client.move_wiki_page("kb-one", "entity/acme", "root-one")
|
||
await client.list_wiki_pages("kb-one", folder_id="")
|
||
await client.list_wiki_pages("kb-one", query="unfiltered")
|
||
|
||
assert folders["folders"] == [{"id": "child", "name": "Child"}]
|
||
assert updated["parent_id"] == "root-one"
|
||
assert moved["folder_id"] == "root-one"
|
||
assert seen == [
|
||
("GET", "/api/v1/knowledgebase/kb-one/wiki/folders"),
|
||
("PUT", "/api/v1/knowledgebase/kb-one/wiki/folders/folder%2Fone"),
|
||
("DELETE", "/api/v1/knowledgebase/kb-one/wiki/folders/folder%2Fone"),
|
||
("PUT", "/api/v1/knowledgebase/kb-one/wiki/move-page"),
|
||
("GET", "/api/v1/knowledgebase/kb-one/wiki/pages"),
|
||
("GET", "/api/v1/knowledgebase/kb-one/wiki/pages"),
|
||
]
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_weknora_client_preserves_folder_conflict_status() -> None:
|
||
async def handler(request: httpx.Request) -> httpx.Response:
|
||
return httpx.Response(
|
||
409,
|
||
json={"message": "wiki folder name conflict"},
|
||
)
|
||
|
||
async with httpx.AsyncClient(
|
||
transport=httpx.MockTransport(handler)
|
||
) as http_client:
|
||
client = WeKnoraClient(
|
||
base_url="http://weknora.local:8080",
|
||
http_client=http_client,
|
||
)
|
||
with pytest.raises(Exception) as caught:
|
||
await client.create_wiki_folder(
|
||
"kb-one",
|
||
name="国内新闻",
|
||
parent_id="",
|
||
)
|
||
|
||
assert getattr(caught.value, "status_code", None) == 409
|