1010 lines
35 KiB
Python
1010 lines
35 KiB
Python
from __future__ import annotations
|
|
|
|
import asyncio
|
|
from types import SimpleNamespace
|
|
from typing import Any
|
|
|
|
import pytest
|
|
from fastapi import HTTPException
|
|
|
|
from app.gateway import llmwiki_index_scheduler
|
|
from app.gateway.routers import llmwiki as llmwiki_router
|
|
from app.gateway.wiki_generation_config import WIKI_CONFIG_SCHEMA_VERSION
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_prompt_schema_upgrade_recompiles_reference_directories_idempotently() -> None:
|
|
rows = [
|
|
{
|
|
"id": "mapping-current-paths",
|
|
"weknora_id": "remote-current-paths",
|
|
"wiki_generation_managed": True,
|
|
"wiki_config_schema_version": 2,
|
|
"wiki_config_revision": 4,
|
|
"wiki_directory_paths": [["政策"], ["政策", "量化目标"]],
|
|
"wiki_parent_directories": ["政策"],
|
|
"wiki_focus_instructions": "只提取明确年份和量化目标",
|
|
"wiki_content_instructions_raw": "保留原文数字",
|
|
"wiki_extraction_instructions_raw": "提取责任主体",
|
|
},
|
|
{
|
|
"id": "mapping-legacy-roots",
|
|
"weknora_id": "remote-legacy-roots",
|
|
"wiki_generation_managed": True,
|
|
"wiki_config_schema_version": 2,
|
|
"wiki_config_revision": 1,
|
|
"wiki_directory_paths": None,
|
|
"wiki_parent_directories": ["历史", "现状"],
|
|
"wiki_focus_instructions": "",
|
|
"wiki_content_instructions_raw": "",
|
|
"wiki_extraction_instructions_raw": "",
|
|
},
|
|
{
|
|
"id": "mapping-unknown-schema",
|
|
"weknora_id": "remote-unknown-schema",
|
|
"wiki_generation_managed": True,
|
|
"wiki_config_schema_version": None,
|
|
"wiki_content_instructions_raw": None,
|
|
"wiki_extraction_instructions_raw": None,
|
|
},
|
|
{
|
|
"id": "mapping-already-current",
|
|
"weknora_id": "remote-already-current",
|
|
"wiki_generation_managed": True,
|
|
"wiki_config_schema_version": WIKI_CONFIG_SCHEMA_VERSION,
|
|
},
|
|
{
|
|
"id": "mapping-unmanaged",
|
|
"weknora_id": "remote-unmanaged",
|
|
"wiki_generation_managed": False,
|
|
"wiki_config_schema_version": 1,
|
|
},
|
|
]
|
|
remote_updates: list[tuple[str, dict[str, Any]]] = []
|
|
|
|
class Store:
|
|
async def list_visible(self, *_: object, **__: object) -> list[dict[str, Any]]:
|
|
return rows
|
|
|
|
async def update_mapping(
|
|
self,
|
|
mapping_id: str,
|
|
**changes: Any,
|
|
) -> dict[str, Any]:
|
|
row = next(item for item in rows if item["id"] == mapping_id)
|
|
row.update(changes)
|
|
return dict(row)
|
|
|
|
class Client:
|
|
async def update_knowledge_base(
|
|
self,
|
|
remote_id: str,
|
|
changes: dict[str, Any],
|
|
) -> dict[str, Any]:
|
|
remote_updates.append((remote_id, changes))
|
|
return {"id": remote_id, **changes}
|
|
|
|
app = SimpleNamespace(state=SimpleNamespace(llmwiki_store=Store()))
|
|
result = await llmwiki_index_scheduler._upgrade_wiki_generation_configs_once(
|
|
app,
|
|
client=Client(),
|
|
)
|
|
|
|
assert result == {"eligible": 2, "upgraded": 2, "failed": 0}
|
|
assert {remote_id for remote_id, _ in remote_updates} == {
|
|
"remote-current-paths",
|
|
"remote-legacy-roots",
|
|
}
|
|
for _, changes in remote_updates:
|
|
content = changes["wiki_config"]["content_instructions"]
|
|
assert "【参考目录层级】" in content
|
|
assert "不是必须逐项套用的目录模板" in content
|
|
assert "唯一允许使用的目录白名单" not in content
|
|
current_prompt = next(
|
|
changes["wiki_config"]
|
|
for remote_id, changes in remote_updates
|
|
if remote_id == "remote-current-paths"
|
|
)
|
|
assert "只提取明确年份和量化目标" in current_prompt["content_instructions"]
|
|
assert "只提取明确年份和量化目标" in current_prompt["extraction_instructions"]
|
|
assert rows[0]["wiki_config_schema_version"] == WIKI_CONFIG_SCHEMA_VERSION
|
|
assert rows[0]["wiki_config_revision"] == 5
|
|
assert rows[1]["wiki_config_schema_version"] == WIKI_CONFIG_SCHEMA_VERSION
|
|
assert rows[1]["wiki_config_apply_status"] == "active"
|
|
assert rows[1]["wiki_config_apply_error"] is None
|
|
|
|
second = await llmwiki_index_scheduler._upgrade_wiki_generation_configs_once(
|
|
app,
|
|
client=Client(),
|
|
)
|
|
assert second == {"eligible": 0, "upgraded": 0, "failed": 0}
|
|
assert len(remote_updates) == 2
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_prompt_schema_upgrade_records_failure_and_retries() -> None:
|
|
row: dict[str, Any] = {
|
|
"id": "mapping-retry",
|
|
"weknora_id": "remote-retry",
|
|
"wiki_generation_managed": True,
|
|
"wiki_config_schema_version": 2,
|
|
"wiki_config_revision": 7,
|
|
"wiki_directory_paths": [["参考目录"]],
|
|
"wiki_focus_instructions": "",
|
|
"wiki_content_instructions_raw": "",
|
|
"wiki_extraction_instructions_raw": "",
|
|
}
|
|
|
|
class Store:
|
|
async def list_visible(self, *_: object, **__: object) -> list[dict[str, Any]]:
|
|
return [row]
|
|
|
|
async def update_mapping(
|
|
self,
|
|
_: str,
|
|
**changes: Any,
|
|
) -> dict[str, Any]:
|
|
row.update(changes)
|
|
return dict(row)
|
|
|
|
class FailingClient:
|
|
async def update_knowledge_base(self, *_: object, **__: object) -> dict[str, Any]:
|
|
raise RuntimeError("provider unavailable")
|
|
|
|
class WorkingClient:
|
|
async def update_knowledge_base(
|
|
self,
|
|
remote_id: str,
|
|
changes: dict[str, Any],
|
|
) -> dict[str, Any]:
|
|
return {"id": remote_id, **changes}
|
|
|
|
app = SimpleNamespace(state=SimpleNamespace(llmwiki_store=Store()))
|
|
failed = await llmwiki_index_scheduler._upgrade_wiki_generation_configs_once(
|
|
app,
|
|
client=FailingClient(),
|
|
)
|
|
assert failed == {"eligible": 1, "upgraded": 0, "failed": 1}
|
|
assert row["wiki_config_schema_version"] == 2
|
|
assert row["wiki_config_apply_status"] == "error"
|
|
assert "provider unavailable" in row["wiki_config_apply_error"]
|
|
|
|
retried = await llmwiki_index_scheduler._upgrade_wiki_generation_configs_once(
|
|
app,
|
|
client=WorkingClient(),
|
|
)
|
|
assert retried == {"eligible": 1, "upgraded": 1, "failed": 0}
|
|
assert row["wiki_config_schema_version"] == WIKI_CONFIG_SCHEMA_VERSION
|
|
assert row["wiki_config_revision"] == 8
|
|
assert row["wiki_config_apply_status"] == "active"
|
|
assert row["wiki_config_apply_error"] is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_persists_legacy_reference_roots_without_precreating_them(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
persisted: list[dict[str, Any]] = []
|
|
create_calls: list[dict[str, Any]] = []
|
|
|
|
class Client:
|
|
async def create_knowledge_base(self, **kwargs: Any) -> dict[str, Any]:
|
|
create_calls.append(kwargs)
|
|
return {
|
|
"id": "remote-kb",
|
|
"name": kwargs["name"],
|
|
"description": kwargs["description"],
|
|
"type": kwargs["kb_type"],
|
|
}
|
|
|
|
class Store:
|
|
async def create_mapping(self, **kwargs: Any) -> dict[str, Any]:
|
|
persisted.append(kwargs)
|
|
return {
|
|
"id": "mapping-kb",
|
|
"publication_status": "private",
|
|
"wiki_index_enabled": True,
|
|
**kwargs,
|
|
}
|
|
|
|
async def actor(_: object) -> tuple[str, bool]:
|
|
return "user-a", False
|
|
|
|
async def embedding_models(_: object) -> list[dict[str, Any]]:
|
|
return [{"id": "embed-one", "type": "Embedding", "available": True, "is_default": True}]
|
|
|
|
monkeypatch.setattr(llmwiki_router, "_actor", actor)
|
|
monkeypatch.setattr(llmwiki_router, "_embedding_models", embedding_models)
|
|
monkeypatch.setattr(llmwiki_router, "_client_or_503", lambda: Client())
|
|
monkeypatch.setattr(llmwiki_router, "_store", lambda _: Store())
|
|
monkeypatch.setattr(llmwiki_router, "_schedule_local_wiki_sync", lambda *_: None)
|
|
|
|
created = await llmwiki_router.create_knowledge_base(
|
|
object(),
|
|
llmwiki_router.KnowledgeBaseCreate(
|
|
name="专题资料",
|
|
wiki_enabled=True,
|
|
wiki_config=llmwiki_router.WikiGenerationInstructions(
|
|
content_instructions="保留证据",
|
|
parent_directories=[" 第一章 政治 ", "第二章 军事", "第一章 政治"],
|
|
),
|
|
),
|
|
)
|
|
|
|
assert persisted[0]["wiki_parent_directories"] == ["第一章 政治", "第二章 军事"]
|
|
assert persisted[0]["wiki_directory_paths"] == [
|
|
["第一章 政治"],
|
|
["第二章 军事"],
|
|
]
|
|
assert "parent_directories" not in create_calls[0]
|
|
assert "【参考目录层级】" in create_calls[0]["wiki_content_instructions"]
|
|
assert created["wiki_config"]["parent_directories"] == ["第一章 政治", "第二章 军事"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_compiles_and_persists_hierarchy_and_focus(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
create_calls: list[dict[str, Any]] = []
|
|
persisted: list[dict[str, Any]] = []
|
|
|
|
class Client:
|
|
async def create_knowledge_base(self, **kwargs: Any) -> dict[str, Any]:
|
|
create_calls.append(kwargs)
|
|
return {"id": "remote-kb", "name": kwargs["name"], "type": "document"}
|
|
|
|
class Store:
|
|
async def create_mapping(self, **kwargs: Any) -> dict[str, Any]:
|
|
persisted.append(kwargs)
|
|
return {"id": "mapping-kb", "publication_status": "private", **kwargs}
|
|
|
|
async def actor(_: object) -> tuple[str, bool]:
|
|
return "user-a", False
|
|
|
|
async def embedding_models(_: object) -> list[dict[str, Any]]:
|
|
return [
|
|
{
|
|
"id": "embed-one",
|
|
"type": "Embedding",
|
|
"available": True,
|
|
"is_default": True,
|
|
}
|
|
]
|
|
|
|
monkeypatch.setattr(llmwiki_router, "_actor", actor)
|
|
monkeypatch.setattr(llmwiki_router, "_embedding_models", embedding_models)
|
|
monkeypatch.setattr(llmwiki_router, "_client_or_503", lambda: Client())
|
|
monkeypatch.setattr(llmwiki_router, "_store", lambda _: Store())
|
|
monkeypatch.setattr(llmwiki_router, "_schedule_local_wiki_sync", lambda *_: None)
|
|
|
|
created = await llmwiki_router.create_knowledge_base(
|
|
object(),
|
|
llmwiki_router.KnowledgeBaseCreate(
|
|
name="政策知识库",
|
|
wiki_enabled=True,
|
|
wiki_config=llmwiki_router.WikiGenerationInstructions(
|
|
content_instructions="保留原文数字",
|
|
extraction_instructions="提取主体",
|
|
directory_paths=[
|
|
["政策法规", "国内政策", "申报流程"],
|
|
["政策法规", "国外政策"],
|
|
],
|
|
focus_instructions="重点关注适用对象与期限",
|
|
),
|
|
),
|
|
)
|
|
|
|
expected_paths = [
|
|
["政策法规"],
|
|
["政策法规", "国内政策"],
|
|
["政策法规", "国内政策", "申报流程"],
|
|
["政策法规", "国外政策"],
|
|
]
|
|
assert "【参考目录层级】" in create_calls[0]["wiki_content_instructions"]
|
|
assert "不是必须逐项套用的目录模板" in create_calls[0]["wiki_content_instructions"]
|
|
assert "重点关注适用对象与期限" in create_calls[0]["wiki_content_instructions"]
|
|
assert "重点关注适用对象与期限" in create_calls[0]["wiki_extraction_instructions"]
|
|
assert persisted[0]["wiki_directory_paths"] == expected_paths
|
|
assert persisted[0]["wiki_parent_directories"] == ["政策法规"]
|
|
assert persisted[0]["wiki_focus_instructions"] == "重点关注适用对象与期限"
|
|
assert persisted[0]["wiki_content_instructions_raw"] == "保留原文数字"
|
|
assert persisted[0]["wiki_config_schema_version"] == WIKI_CONFIG_SCHEMA_VERSION
|
|
assert persisted[0]["wiki_config_revision"] == 1
|
|
assert persisted[0]["wiki_config_apply_status"] == "active"
|
|
assert created["wiki_config"]["directory_paths"] == expected_paths
|
|
assert created["wiki_config"]["directory_prompt_coverage"] == "full"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_without_fixed_roots_keeps_legacy_path(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
persisted: dict[str, Any] = {}
|
|
|
|
class Client:
|
|
async def create_knowledge_base(self, **kwargs: Any) -> dict[str, Any]:
|
|
return {"id": "remote-kb", "name": kwargs["name"], "type": kwargs["kb_type"]}
|
|
|
|
class Store:
|
|
async def create_mapping(self, **kwargs: Any) -> dict[str, Any]:
|
|
persisted.update(kwargs)
|
|
return {"id": "mapping-kb", "publication_status": "private", **kwargs}
|
|
|
|
async def actor(_: object) -> tuple[str, bool]:
|
|
return "user-a", False
|
|
|
|
async def embedding_models(_: object) -> list[dict[str, Any]]:
|
|
return [{"id": "embed-one", "type": "Embedding", "available": True, "is_default": True}]
|
|
|
|
monkeypatch.setattr(llmwiki_router, "_actor", actor)
|
|
monkeypatch.setattr(llmwiki_router, "_embedding_models", embedding_models)
|
|
monkeypatch.setattr(llmwiki_router, "_client_or_503", lambda: Client())
|
|
monkeypatch.setattr(llmwiki_router, "_store", lambda _: Store())
|
|
monkeypatch.setattr(llmwiki_router, "_schedule_local_wiki_sync", lambda *_: None)
|
|
|
|
await llmwiki_router.create_knowledge_base(
|
|
object(),
|
|
llmwiki_router.KnowledgeBaseCreate(
|
|
name="普通资料",
|
|
wiki_enabled=False,
|
|
),
|
|
)
|
|
|
|
assert persisted["wiki_parent_directories"] is None
|
|
assert persisted["wiki_generation_managed"] is False
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_reference_hierarchy_does_not_call_remote_folder_apis(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
persisted: list[dict[str, Any]] = []
|
|
|
|
class Client:
|
|
async def create_knowledge_base(self, **kwargs: Any) -> dict[str, Any]:
|
|
return {"id": "remote-kb", "name": kwargs["name"], "type": "document"}
|
|
|
|
class Store:
|
|
async def create_mapping(self, **kwargs: Any) -> dict[str, Any]:
|
|
persisted.append(kwargs)
|
|
return {"id": "mapping-kb", "publication_status": "private", **kwargs}
|
|
|
|
async def actor(_: object) -> tuple[str, bool]:
|
|
return "user-a", False
|
|
|
|
async def embedding_models(_: object) -> list[dict[str, Any]]:
|
|
return [
|
|
{
|
|
"id": "embed-one",
|
|
"type": "Embedding",
|
|
"available": True,
|
|
"is_default": True,
|
|
}
|
|
]
|
|
|
|
monkeypatch.setattr(llmwiki_router, "_actor", actor)
|
|
monkeypatch.setattr(llmwiki_router, "_embedding_models", embedding_models)
|
|
monkeypatch.setattr(llmwiki_router, "_client_or_503", lambda: Client())
|
|
monkeypatch.setattr(llmwiki_router, "_store", lambda _: Store())
|
|
monkeypatch.setattr(llmwiki_router, "_schedule_local_wiki_sync", lambda *_: None)
|
|
|
|
await llmwiki_router.create_knowledge_base(
|
|
object(),
|
|
llmwiki_router.KnowledgeBaseCreate(
|
|
name="层级知识库",
|
|
wiki_enabled=True,
|
|
wiki_config=llmwiki_router.WikiGenerationInstructions(
|
|
directory_paths=[["一级"], ["一级", "二级"]],
|
|
),
|
|
),
|
|
)
|
|
|
|
assert persisted[0]["wiki_directory_paths"] == [["一级"], ["一级", "二级"]]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_edit_updates_directory_reference_without_precreate_or_settle(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
row = {
|
|
"id": "mapping-kb",
|
|
"weknora_id": "remote-kb",
|
|
"owner_user_id": "user-a",
|
|
"name": "专题资料",
|
|
"description": "",
|
|
"kb_type": "document",
|
|
"publication_status": "private",
|
|
"wiki_parent_directories": ["旧目录"],
|
|
}
|
|
remote_updates: list[dict[str, Any]] = []
|
|
local_updates: list[dict[str, Any]] = []
|
|
scheduled: list[dict[str, Any]] = []
|
|
|
|
class Client:
|
|
async def update_knowledge_base(self, _: str, changes: dict[str, Any]) -> dict[str, Any]:
|
|
remote_updates.append(changes)
|
|
return {
|
|
"id": "remote-kb",
|
|
"name": row["name"],
|
|
"type": "document",
|
|
"wiki_config": changes.get("wiki_config", {}),
|
|
}
|
|
|
|
async def get_knowledge_base(self, _: str) -> dict[str, Any]:
|
|
return {"id": "remote-kb", "name": row["name"], "type": "document"}
|
|
|
|
class Store:
|
|
async def update_mapping(self, _: str, **kwargs: Any) -> dict[str, Any]:
|
|
local_updates.append(kwargs)
|
|
return {
|
|
**row,
|
|
"wiki_parent_directories": kwargs.get(
|
|
"wiki_parent_directories",
|
|
row["wiki_parent_directories"],
|
|
),
|
|
}
|
|
|
|
async def authorized(*_: object, **__: object) -> tuple[dict[str, Any], str, bool]:
|
|
return row, "user-a", False
|
|
|
|
monkeypatch.setattr(llmwiki_router, "_authorized_mapping", authorized)
|
|
monkeypatch.setattr(llmwiki_router, "_client_or_503", lambda: Client())
|
|
monkeypatch.setattr(llmwiki_router, "_store", lambda _: Store())
|
|
monkeypatch.setattr(
|
|
llmwiki_router,
|
|
"_schedule_local_wiki_sync",
|
|
lambda _, mapping, **__: scheduled.append(dict(mapping)),
|
|
)
|
|
|
|
updated = await llmwiki_router.update_knowledge_base(
|
|
object(),
|
|
"mapping-kb",
|
|
llmwiki_router.KnowledgeBaseUpdate(
|
|
wiki_config=llmwiki_router.WikiGenerationInstructions(
|
|
content_instructions="按证据整理",
|
|
parent_directories=[" 政治环境 ", "军事环境"],
|
|
)
|
|
),
|
|
)
|
|
|
|
assert len(remote_updates) == 1
|
|
assert remote_updates[0]["wiki_config"]["content_instructions"].startswith(
|
|
"按证据整理\n\n【参考目录层级】"
|
|
)
|
|
assert "- 政治环境" in remote_updates[0]["wiki_config"]["content_instructions"]
|
|
assert remote_updates[0]["wiki_config"]["extraction_instructions"] == ""
|
|
assert local_updates[-1]["wiki_parent_directories"] == ["政治环境", "军事环境"]
|
|
assert scheduled == []
|
|
assert updated["wiki_config"]["parent_directories"] == ["政治环境", "军事环境"]
|
|
|
|
cleared = await llmwiki_router.update_knowledge_base(
|
|
object(),
|
|
"mapping-kb",
|
|
llmwiki_router.KnowledgeBaseUpdate(
|
|
wiki_config=llmwiki_router.WikiGenerationInstructions(
|
|
parent_directories=[],
|
|
)
|
|
),
|
|
)
|
|
|
|
assert local_updates[-1]["wiki_parent_directories"] == []
|
|
assert scheduled == []
|
|
assert cleared["wiki_config"]["parent_directories"] == []
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_focus_only_patch_schedules_reconcile_and_preserves_unset_semantics(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
row: dict[str, Any] = {
|
|
"id": "mapping-focus",
|
|
"weknora_id": "remote-focus",
|
|
"owner_user_id": "user-a",
|
|
"name": "重点知识库",
|
|
"description": "",
|
|
"kb_type": "document",
|
|
"publication_status": "private",
|
|
"wiki_parent_directories": [],
|
|
"wiki_directory_paths": [],
|
|
"wiki_focus_instructions": "旧重点",
|
|
"wiki_content_instructions_raw": "",
|
|
"wiki_extraction_instructions_raw": "",
|
|
"wiki_generation_managed": True,
|
|
"wiki_config_schema_version": 2,
|
|
"wiki_config_revision": 1,
|
|
"wiki_config_apply_status": "active",
|
|
}
|
|
remote: dict[str, Any] = {
|
|
"id": "remote-focus",
|
|
"name": row["name"],
|
|
"description": "",
|
|
"type": "document",
|
|
"indexing_strategy": {"wiki_enabled": True},
|
|
"wiki_config": {
|
|
"content_instructions": "旧重点",
|
|
"extraction_instructions": "旧重点",
|
|
},
|
|
}
|
|
scheduled: list[tuple[str, bool]] = []
|
|
|
|
class Client:
|
|
async def get_knowledge_base(self, _: str) -> dict[str, Any]:
|
|
return {
|
|
**remote,
|
|
"wiki_config": dict(remote["wiki_config"]),
|
|
"indexing_strategy": dict(remote["indexing_strategy"]),
|
|
}
|
|
|
|
async def update_knowledge_base(
|
|
self,
|
|
_: str,
|
|
changes: dict[str, Any],
|
|
) -> dict[str, Any]:
|
|
if "wiki_config" in changes:
|
|
remote["wiki_config"] = dict(changes["wiki_config"])
|
|
remote.update(
|
|
{
|
|
key: value
|
|
for key, value in changes.items()
|
|
if key != "wiki_config"
|
|
}
|
|
)
|
|
return await self.get_knowledge_base("remote-focus")
|
|
|
|
class Store:
|
|
async def update_mapping(
|
|
self,
|
|
_: str,
|
|
**changes: Any,
|
|
) -> dict[str, Any]:
|
|
for key in (
|
|
"wiki_parent_directories",
|
|
"wiki_directory_paths",
|
|
"wiki_focus_instructions",
|
|
"wiki_content_instructions_raw",
|
|
"wiki_extraction_instructions_raw",
|
|
"wiki_generation_managed",
|
|
"wiki_config_schema_version",
|
|
"wiki_config_revision",
|
|
"wiki_config_apply_status",
|
|
"wiki_config_apply_error",
|
|
):
|
|
if key in changes:
|
|
row[key] = changes[key]
|
|
return dict(row)
|
|
|
|
async def authorized(*_: object, **__: object) -> tuple[dict[str, Any], str, bool]:
|
|
return dict(row), "user-a", False
|
|
|
|
monkeypatch.setattr(llmwiki_router, "_authorized_mapping", authorized)
|
|
monkeypatch.setattr(llmwiki_router, "_client_or_503", lambda: Client())
|
|
monkeypatch.setattr(llmwiki_router, "_store", lambda _: Store())
|
|
monkeypatch.setattr(
|
|
llmwiki_router,
|
|
"_schedule_local_wiki_sync",
|
|
lambda _, mapping, *, constraints_changed=False: scheduled.append(
|
|
(
|
|
str(mapping.get("wiki_focus_instructions") or ""),
|
|
constraints_changed,
|
|
)
|
|
),
|
|
)
|
|
|
|
await llmwiki_router.update_knowledge_base(
|
|
object(),
|
|
"mapping-focus",
|
|
llmwiki_router.KnowledgeBaseUpdate(
|
|
wiki_config=llmwiki_router.WikiGenerationInstructions(
|
|
focus_instructions="新重点",
|
|
)
|
|
),
|
|
)
|
|
assert row["wiki_focus_instructions"] == "新重点"
|
|
assert scheduled == [("新重点", True)]
|
|
|
|
# Omitting focus preserves it and a content-only edit does not enqueue a
|
|
# destructive reconciliation pass.
|
|
await llmwiki_router.update_knowledge_base(
|
|
object(),
|
|
"mapping-focus",
|
|
llmwiki_router.KnowledgeBaseUpdate(
|
|
wiki_config=llmwiki_router.WikiGenerationInstructions(
|
|
content_instructions="新的内容生成要求",
|
|
)
|
|
),
|
|
)
|
|
assert row["wiki_focus_instructions"] == "新重点"
|
|
assert scheduled == [("新重点", True)]
|
|
|
|
# Explicit null/empty focus clears the persisted constraint and still
|
|
# supersedes any pending focus reconciliation plan.
|
|
await llmwiki_router.update_knowledge_base(
|
|
object(),
|
|
"mapping-focus",
|
|
llmwiki_router.KnowledgeBaseUpdate(
|
|
wiki_config=llmwiki_router.WikiGenerationInstructions(
|
|
focus_instructions=None,
|
|
)
|
|
),
|
|
)
|
|
assert row["wiki_focus_instructions"] == ""
|
|
assert scheduled == [("新重点", True), ("", True)]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_local_failure_restores_previous_remote_instructions(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
row = {
|
|
"id": "mapping-kb",
|
|
"weknora_id": "remote-kb",
|
|
"owner_user_id": "user-a",
|
|
"name": "专题资料",
|
|
"description": "",
|
|
"kb_type": "document",
|
|
"publication_status": "private",
|
|
"wiki_parent_directories": ["政策法规"],
|
|
"wiki_directory_paths": [["政策法规"], ["政策法规", "国内政策"]],
|
|
"wiki_focus_instructions": "旧重点",
|
|
"wiki_content_instructions_raw": "旧内容要求",
|
|
"wiki_extraction_instructions_raw": "旧抽取要求",
|
|
"wiki_generation_managed": True,
|
|
"wiki_config_schema_version": 2,
|
|
"wiki_config_revision": 1,
|
|
"wiki_config_apply_status": "active",
|
|
}
|
|
updates: list[dict[str, Any]] = []
|
|
|
|
class Client:
|
|
async def get_knowledge_base(self, _: str) -> dict[str, Any]:
|
|
return {
|
|
"id": "remote-kb",
|
|
"name": row["name"],
|
|
"description": "",
|
|
"type": "document",
|
|
"indexing_strategy": {"wiki_enabled": True},
|
|
"wiki_config": {
|
|
"content_instructions": "旧远端内容",
|
|
"extraction_instructions": "旧远端抽取",
|
|
},
|
|
}
|
|
|
|
async def update_knowledge_base(
|
|
self,
|
|
_: str,
|
|
changes: dict[str, Any],
|
|
) -> dict[str, Any]:
|
|
updates.append(changes)
|
|
return {
|
|
"id": "remote-kb",
|
|
"name": row["name"],
|
|
"type": "document",
|
|
"wiki_config": changes.get("wiki_config", {}),
|
|
}
|
|
|
|
class Store:
|
|
async def update_mapping(self, *_: object, **__: object) -> dict[str, Any]:
|
|
raise RuntimeError("database unavailable")
|
|
|
|
async def authorized(*_: object, **__: object) -> tuple[dict[str, Any], str, bool]:
|
|
return row, "user-a", False
|
|
|
|
monkeypatch.setattr(llmwiki_router, "_authorized_mapping", authorized)
|
|
monkeypatch.setattr(llmwiki_router, "_client_or_503", lambda: Client())
|
|
monkeypatch.setattr(llmwiki_router, "_store", lambda _: Store())
|
|
|
|
with pytest.raises(RuntimeError, match="database unavailable"):
|
|
await llmwiki_router.update_knowledge_base(
|
|
object(),
|
|
"mapping-kb",
|
|
llmwiki_router.KnowledgeBaseUpdate(
|
|
wiki_config=llmwiki_router.WikiGenerationInstructions(
|
|
focus_instructions="新重点",
|
|
)
|
|
),
|
|
)
|
|
|
|
assert len(updates) == 2
|
|
assert "新重点" in updates[0]["wiki_config"]["content_instructions"]
|
|
assert updates[1] == {
|
|
"wiki_config": {
|
|
"content_instructions": "旧远端内容",
|
|
"extraction_instructions": "旧远端抽取",
|
|
}
|
|
}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_directory_references_do_not_schedule_reconciliation(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
reconciled: list[dict[str, Any]] = []
|
|
|
|
class Client:
|
|
async def get_knowledge_base(self, _: str) -> dict[str, Any]:
|
|
return {"id": "remote-kb", "is_processing": False}
|
|
|
|
async def reconcile(_: object, mapping: dict[str, Any], **__: Any) -> dict[str, Any]:
|
|
reconciled.append(dict(mapping))
|
|
return {"status": "completed"}
|
|
|
|
import app.gateway.wiki_directory_constraints as constraints
|
|
|
|
monkeypatch.setattr(constraints, "reconcile_fixed_parent_directories", reconcile)
|
|
app = SimpleNamespace(state=SimpleNamespace(llmwiki_sync_service=None))
|
|
mapping = {
|
|
"id": "mapping-kb",
|
|
"weknora_id": "remote-kb",
|
|
"wiki_directory_paths": [
|
|
["政治环境"],
|
|
["政治环境", "国内环境"],
|
|
],
|
|
}
|
|
|
|
task = llmwiki_index_scheduler.schedule_local_wiki_mutation_sync(
|
|
app,
|
|
mapping,
|
|
delays=(0.001,),
|
|
client=Client(),
|
|
sync_local=False,
|
|
)
|
|
|
|
assert task is None
|
|
assert reconciled == []
|
|
assert (
|
|
llmwiki_index_scheduler.schedule_local_wiki_mutation_sync(
|
|
app,
|
|
{
|
|
"id": "legacy",
|
|
"weknora_id": "remote-kb",
|
|
"wiki_parent_directories": ["旧目录"],
|
|
},
|
|
delays=(0.001,),
|
|
client=Client(),
|
|
sync_local=False,
|
|
)
|
|
is None
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_focus_only_reconciliation_runs_without_local_index(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
reconciled: list[dict[str, Any]] = []
|
|
|
|
class Client:
|
|
async def get_knowledge_base(self, _: str) -> dict[str, Any]:
|
|
return {"id": "remote-kb", "is_processing": False}
|
|
|
|
async def list_documents(
|
|
self,
|
|
_: str,
|
|
*,
|
|
page: int,
|
|
page_size: int,
|
|
) -> dict[str, Any]:
|
|
return {"items": [], "total": 0, "page": page, "page_size": page_size}
|
|
|
|
async def reconcile(_: object, mapping: dict[str, Any], **__: Any) -> dict[str, Any]:
|
|
reconciled.append(dict(mapping))
|
|
return {"status": "completed"}
|
|
|
|
import app.gateway.wiki_directory_constraints as constraints
|
|
|
|
monkeypatch.setattr(constraints, "reconcile_fixed_parent_directories", reconcile)
|
|
mapping = {
|
|
"id": "mapping-focus",
|
|
"weknora_id": "remote-kb",
|
|
"wiki_directory_paths": [],
|
|
"wiki_focus_instructions": "只抽取明确年份与量化指标",
|
|
}
|
|
task = llmwiki_index_scheduler.schedule_local_wiki_mutation_sync(
|
|
SimpleNamespace(state=SimpleNamespace(llmwiki_sync_service=None)),
|
|
mapping,
|
|
delays=(0.001,),
|
|
client=Client(),
|
|
sync_local=False,
|
|
)
|
|
|
|
assert task is not None
|
|
await asyncio.wait_for(task, timeout=1)
|
|
assert reconciled == [mapping]
|
|
|
|
|
|
def test_router_passes_client_for_focus_only_settle_without_local_index(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
client = object()
|
|
scheduled: list[dict[str, Any]] = []
|
|
|
|
def schedule(
|
|
app: object,
|
|
mapping: dict[str, Any],
|
|
*,
|
|
client: object | None,
|
|
sync_local: bool,
|
|
) -> None:
|
|
scheduled.append(
|
|
{
|
|
"app": app,
|
|
"mapping": mapping,
|
|
"client": client,
|
|
"sync_local": sync_local,
|
|
}
|
|
)
|
|
|
|
monkeypatch.setattr(
|
|
llmwiki_index_scheduler,
|
|
"schedule_local_wiki_mutation_sync",
|
|
schedule,
|
|
)
|
|
monkeypatch.setattr(llmwiki_router, "_client_or_503", lambda: client)
|
|
app = SimpleNamespace(
|
|
state=SimpleNamespace(
|
|
llmwiki_sync_service=None,
|
|
llmwiki_effective_config=SimpleNamespace(
|
|
enabled=False,
|
|
auto_sync=False,
|
|
),
|
|
)
|
|
)
|
|
request = SimpleNamespace(app=app)
|
|
mapping = {
|
|
"id": "mapping-focus",
|
|
"weknora_id": "remote-kb",
|
|
"wiki_directory_paths": [],
|
|
"wiki_focus_instructions": "只抽取量化目标",
|
|
}
|
|
|
|
llmwiki_router._schedule_local_wiki_sync(request, mapping)
|
|
|
|
assert scheduled == [
|
|
{
|
|
"app": app,
|
|
"mapping": mapping,
|
|
"client": client,
|
|
"sync_local": False,
|
|
}
|
|
]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_remote_generation_status_uses_document_pending_subtasks() -> None:
|
|
class Client:
|
|
async def get_knowledge_base(self, _: str) -> dict[str, Any]:
|
|
return {"id": "remote-kb", "is_processing": False, "processing_count": 0}
|
|
|
|
async def list_documents(
|
|
self,
|
|
_: str,
|
|
*,
|
|
page: int,
|
|
page_size: int,
|
|
) -> dict[str, Any]:
|
|
return {
|
|
"items": [
|
|
{
|
|
"id": "doc-one",
|
|
"parse_status": "completed",
|
|
"processing_status": "finalizing",
|
|
"pending_subtasks_count": 1,
|
|
}
|
|
],
|
|
"total": 1,
|
|
"page": page,
|
|
"page_size": page_size,
|
|
}
|
|
|
|
assert await llmwiki_index_scheduler._remote_generation_is_processing(
|
|
Client(),
|
|
{"weknora_id": "remote-kb"},
|
|
) is True
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_remote_generation_status_reports_complete_documents() -> None:
|
|
class Client:
|
|
async def get_knowledge_base(self, _: str) -> dict[str, Any]:
|
|
return {"id": "remote-kb", "is_processing": False}
|
|
|
|
async def list_documents(
|
|
self,
|
|
_: str,
|
|
*,
|
|
page: int,
|
|
page_size: int,
|
|
) -> dict[str, Any]:
|
|
return {
|
|
"items": [
|
|
{
|
|
"id": "doc-one",
|
|
"parse_status": "completed",
|
|
"processing_status": "completed",
|
|
"pending_subtasks_count": 0,
|
|
}
|
|
],
|
|
"total": 1,
|
|
"page": page,
|
|
"page_size": page_size,
|
|
}
|
|
|
|
assert await llmwiki_index_scheduler._remote_generation_is_processing(
|
|
Client(),
|
|
{"weknora_id": "remote-kb"},
|
|
) is False
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_legacy_directory_reference_does_not_start_settle_plan(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
reconciled: list[dict[str, Any]] = []
|
|
|
|
class Client:
|
|
async def get_knowledge_base(self, _: str) -> dict[str, Any]:
|
|
return {"id": "remote-kb", "is_processing": False}
|
|
|
|
async def reconcile(_: object, mapping: dict[str, Any], **__: Any) -> dict[str, Any]:
|
|
reconciled.append(dict(mapping))
|
|
return {"status": "completed"}
|
|
|
|
import app.gateway.wiki_directory_constraints as constraints
|
|
|
|
monkeypatch.setattr(constraints, "reconcile_fixed_parent_directories", reconcile)
|
|
app = SimpleNamespace(state=SimpleNamespace(llmwiki_sync_service=None))
|
|
task = llmwiki_index_scheduler.schedule_local_wiki_mutation_sync(
|
|
app,
|
|
{
|
|
"id": "mapping-kb",
|
|
"weknora_id": "remote-kb",
|
|
"wiki_parent_directories": ["政治环境"],
|
|
},
|
|
delays=(0.1,),
|
|
client=Client(),
|
|
sync_local=False,
|
|
)
|
|
assert task is None
|
|
assert reconciled == []
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_directory_only_mapping_does_not_poll_processing_status(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
reconciliations = 0
|
|
|
|
class Client:
|
|
def __init__(self) -> None:
|
|
self.states = [True, False, False, False]
|
|
self.status_reads = 0
|
|
|
|
async def get_knowledge_base(self, _: str) -> dict[str, Any]:
|
|
self.status_reads += 1
|
|
return {"is_processing": self.states.pop(0)}
|
|
|
|
async def reconcile(*_: object, **__: Any) -> dict[str, Any]:
|
|
nonlocal reconciliations
|
|
reconciliations += 1
|
|
return {"status": "completed"}
|
|
|
|
import app.gateway.wiki_directory_constraints as constraints
|
|
|
|
monkeypatch.setattr(constraints, "reconcile_fixed_parent_directories", reconcile)
|
|
client = Client()
|
|
task = llmwiki_index_scheduler.schedule_local_wiki_mutation_sync(
|
|
SimpleNamespace(state=SimpleNamespace(llmwiki_sync_service=None)),
|
|
{
|
|
"id": "mapping-kb",
|
|
"weknora_id": "remote-kb",
|
|
"wiki_parent_directories": ["政治环境"],
|
|
},
|
|
delays=(0.001,),
|
|
client=client,
|
|
sync_local=False,
|
|
)
|
|
|
|
assert task is None
|
|
assert client.status_reads == 0
|
|
assert reconciliations == 0
|