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