from __future__ import annotations import asyncio import json from types import SimpleNamespace import httpx import pytest from app.gateway import llmwiki_index_scheduler from deerflow.config.llmwiki_config import LocalWikiEmbeddingConfig, LocalWikiIndexConfig from deerflow.integrations.weknora.local_index.embedding import StrictWikiEmbeddingClient from deerflow.integrations.weknora.local_index.sync import WikiSyncService from deerflow.integrations.weknora.local_index.vector_cache import WikiVectorCache from deerflow.persistence.llmwiki_index.memory import MemoryLlmWikiIndexStore class FakeEmbedding: dimensions = 2 fingerprint = "fp" def __init__(self) -> None: self.calls = 0 async def embed_texts(self, texts: list[str]) -> list[list[float]]: self.calls += 1 return [[1.0, 0.0] for _ in texts] class FailingEmbedding(FakeEmbedding): async def embed_texts(self, texts: list[str]) -> list[list[float]]: raise RuntimeError("embedding unavailable") class NewFingerprintEmbedding(FakeEmbedding): fingerprint = "fp-new" class FakeWeKnora: def __init__(self) -> None: self.pages = { "one": { "id": "p1", "slug": "one", "title": "One", "summary": "A long enough summary for indexing and retrieval.", "content": "# Intro\nA long enough Wiki paragraph for indexing and semantic retrieval.", "status": "published", "version": 1, }, } async def get_knowledge_base(self, knowledge_base_id: str): return {"id": knowledge_base_id, "indexing_strategy": {"wiki_enabled": True}} async def list_wiki_pages(self, knowledge_base_id: str, *, page: int, page_size: int): return { "pages": [{"slug": item["slug"], "title": item["title"]} for item in self.pages.values()], "total": len(self.pages), "total_pages": 1, } async def get_wiki_page(self, knowledge_base_id: str, slug: str): return dict(self.pages[slug]) @pytest.mark.asyncio async def test_configured_dimensions_are_validated_but_not_sent_to_bge() -> None: requests: list[dict] = [] def handler(request: httpx.Request) -> httpx.Response: payload = json.loads(request.content) requests.append(payload) if "dimensions" in payload: return httpx.Response(400, json={"detail": "dimensions is not supported"}) return httpx.Response( 200, json={"data": [{"index": 0, "embedding": [1.0, 0.0, 0.0]}]}, ) config = LocalWikiEmbeddingConfig( model="bge-embedding-m3", base_url="http://embedding.test/v1", dimensions=3, max_retries=0, ) async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as http_client: client = StrictWikiEmbeddingClient(config, http_client=http_client) vector = await client.embed_query("Wiki page") assert vector == [1.0, 0.0, 0.0] assert requests == [{"model": "bge-embedding-m3", "input": ["Wiki page"]}] @pytest.mark.asyncio async def test_embedding_error_includes_upstream_reason(caplog) -> None: def handler(request: httpx.Request) -> httpx.Response: return httpx.Response(400, json={"detail": "unsupported request parameter: dimensions"}) config = LocalWikiEmbeddingConfig( model="bge-embedding-m3", base_url="http://embedding.test/v1", max_retries=0, ) async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as http_client: client = StrictWikiEmbeddingClient(config, http_client=http_client) with pytest.raises(RuntimeError, match="unsupported request parameter: dimensions"): await client.embed_query("Wiki page") assert "unsupported request parameter: dimensions" in caplog.text @pytest.mark.asyncio async def test_omitted_dimensions_are_discovered_from_first_response() -> None: def handler(request: httpx.Request) -> httpx.Response: return httpx.Response( 200, json={"data": [{"index": 0, "embedding": [0.0, 1.0]}]}, ) config = LocalWikiEmbeddingConfig( model="bge-embedding-m3", base_url="http://embedding.test/v1", dimensions=None, max_retries=0, ) async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as http_client: client = StrictWikiEmbeddingClient(config, http_client=http_client) await client.embed_query("dimension probe") assert client.dimensions == 2 @pytest.mark.asyncio async def test_sync_is_incremental_and_full_scan_deletes_missing_pages() -> None: config = LocalWikiIndexConfig(chunk_min_chars=10, min_similarity=0) store = MemoryLlmWikiIndexStore() embedding = FakeEmbedding() client = FakeWeKnora() service = WikiSyncService(config, store, embedding, WikiVectorCache(store), client) mapping = {"id": "kb", "weknora_id": "remote", "name": "Wiki"} first = await service.sync_mapping(mapping) assert first["changed_pages"] == 1 assert embedding.calls == 1 statuses = await store.list_page_index_status("kb") assert statuses[0]["title"] == "One" assert statuses[0]["vector_count"] > 0 assert store.states["kb"]["remote_page_count"] == 1 second = await service.sync_mapping(mapping) assert second["changed_pages"] == 0 assert embedding.calls == 1 client.pages.clear() deleted = await service.sync_mapping(mapping) assert deleted["deleted_pages"] == 1 assert next(iter(store.pages.values()))["is_remote_deleted"] is True @pytest.mark.asyncio async def test_page_failure_keeps_previous_generation_and_marks_partial() -> None: config = LocalWikiIndexConfig(chunk_min_chars=10, min_similarity=0) store = MemoryLlmWikiIndexStore() client = FakeWeKnora() mapping = {"id": "kb", "weknora_id": "remote", "name": "Wiki"} await WikiSyncService(config, store, FakeEmbedding(), WikiVectorCache(store), client).sync_mapping(mapping) page = next(iter(store.pages.values())) old_hash = page["content_hash"] old_vectors = list(store.vectors[page["id"]]) client.pages["one"]["content"] += " Changed." result = await WikiSyncService(config, store, FailingEmbedding(), WikiVectorCache(store), client).sync_mapping(mapping) assert result["status"] == "partial" assert store.states["kb"]["state"] == "partial" assert store.states["kb"]["failed_page_count"] == 1 assert store.pages[page["id"]]["content_hash"] == old_hash assert store.pages[page["id"]]["index_status"] == "ready" assert store.vectors[page["id"]] == old_vectors @pytest.mark.asyncio async def test_fingerprint_change_automatically_rebuilds_entire_mapping() -> None: config = LocalWikiIndexConfig(chunk_min_chars=10, min_similarity=0) store = MemoryLlmWikiIndexStore() client = FakeWeKnora() mapping = {"id": "kb", "weknora_id": "remote", "name": "Wiki"} await WikiSyncService(config, store, FakeEmbedding(), WikiVectorCache(store), client).sync_mapping(mapping) embedding = NewFingerprintEmbedding() result = await WikiSyncService(config, store, embedding, WikiVectorCache(store), client).sync_mapping(mapping) assert result["status"] == "completed" assert result["changed_pages"] == 1 assert embedding.calls == 1 assert store.states["kb"]["state"] == "idle" assert store.states["kb"]["embedding_fingerprint"] == "fp-new" assert {vector["embedding_fingerprint"] for vectors in store.vectors.values() for vector in vectors} == {"fp-new"} @pytest.mark.asyncio async def test_auto_sync_scheduler_scans_immediately_then_waits(monkeypatch) -> None: calls: list[str] = [] async def fake_sleep(seconds: float) -> None: calls.append(f"sleep:{seconds}") raise asyncio.CancelledError async def fake_sync_once(app) -> None: calls.append("sync") if calls.count("sync") > 1: raise asyncio.CancelledError monkeypatch.setattr(llmwiki_index_scheduler.asyncio, "sleep", fake_sleep) monkeypatch.setattr(llmwiki_index_scheduler, "_sync_once", fake_sync_once) app = SimpleNamespace( state=SimpleNamespace( config=SimpleNamespace( llmwiki=SimpleNamespace( local_wiki_index=SimpleNamespace(sync_interval_seconds=120), ) ) ) ) with pytest.raises(asyncio.CancelledError): await llmwiki_index_scheduler.local_wiki_index_scheduler_loop(app) assert calls == ["sync", "sleep:120"] def test_local_wiki_index_defaults_to_automatic_sync() -> None: config = LocalWikiIndexConfig() assert config.auto_sync is True assert config.sync_interval_seconds == 30 def test_local_wiki_index_accepts_legacy_sync_flag() -> None: config = LocalWikiIndexConfig.model_validate({"sync_enabled": True}) assert config.auto_sync is True assert config.sync_enabled is True