from __future__ import annotations import asyncio import json from io import BytesIO from types import SimpleNamespace import httpx import pytest from fastapi import FastAPI, HTTPException, UploadFile from fastapi.responses import Response from starlette.datastructures import Headers, QueryParams from starlette.requests import Request from deerflow.config.agents_config import AgentConfig from deerflow.config.llmwiki_config import ( LlmWikiConfig, LlmWikiRuntimeOverride, resolve_llmwiki_runtime, ) from deerflow.integrations.weknora.client import WeKnoraClient, WeKnoraError from deerflow.integrations.weknora.knowledge_profile import ( are_test_questions_customized, default_test_questions, generated_test_questions_name, resolved_test_questions, ) from deerflow.integrations.weknora.runtime import build_weknora_client from deerflow.persistence.llmwiki import MemoryLlmWikiStore from deerflow.tools.builtins.llmwiki_search_tool import ( _sanitize_conversation_deposit_value as sanitize_tool_deposit_value, ) from deerflow.tools.builtins.llmwiki_search_tool import ( selected_knowledge_base_ids, ) def _compatible_knowledge_package_bytes(*, bom: bool = False) -> bytes: records = [ { "record_type": "source", "payload": {"package_version": 1}, }, { "record_type": "source", "payload": { "provider": "weknora", "export_format": "assistant-knowledge-jsonl", }, }, { "record_type": "wiki_page", "payload": { "slug": "wiki/cooling", "title": "散热", "content": "散热管理用于降低设备温度。", }, }, ] content = ("\n\n" + "\n".join(json.dumps(record, ensure_ascii=False) for record in records) + "\n").encode() return (b"\xef\xbb\xbf" if bom else b"") + content def _request( path: str, *, query: str = "", cookies: dict[str, str] | None = None, request_headers: dict[str, str] | None = None, root_path: str = "", ) -> Request: headers: list[tuple[bytes, bytes]] = [(key.lower().encode(), value.encode()) for key, value in (request_headers or {}).items()] if cookies: headers.append((b"cookie", "; ".join(f"{key}={value}" for key, value in cookies.items()).encode())) return Request( { "type": "http", "method": "GET", "path": path, "root_path": root_path, "query_string": query.encode(), "headers": headers, "scheme": "http", "server": ("testserver", 80), "client": ("testclient", 50000), } ) def _first_gateway_match(method: str, path: str) -> str: from app.gateway.app import app as gateway_app for route in gateway_app.router.routes: route_contexts = getattr(route, "effective_route_contexts", None) candidates = route_contexts() if callable(route_contexts) else [route] for candidate in candidates: methods = getattr(candidate, "methods", None) path_regex = getattr(candidate, "path_regex", None) if methods is not None and method in methods and path_regex is not None and path_regex.match(path): return str(getattr(candidate, "path", "")) raise AssertionError(f"No route matched {method} {path}") def _knowledge_profile_mapping_row(*, name: str, questions: list[str] | None) -> dict[str, object]: return { "id": "mapping-one", "weknora_id": "remote-one", "owner_user_id": "user-one", "name": name, "description": "", "test_questions": questions, "retrieval_settings": None, "kb_type": "document", "publication_status": "private", } def test_legacy_persisted_defaults_follow_renamed_knowledge_base() -> None: old_defaults = default_test_questions("旧名称") row = _knowledge_profile_mapping_row(name="新名称", questions=old_defaults) assert generated_test_questions_name(old_defaults) == "旧名称" assert resolved_test_questions(row) == default_test_questions("新名称") assert are_test_questions_customized(row) is False def test_genuinely_custom_questions_are_not_rewritten_after_rename() -> None: questions = ["旧名称的自定义问题保留吗?"] row = _knowledge_profile_mapping_row(name="新名称", questions=questions) assert generated_test_questions_name(questions) is None assert resolved_test_questions(row) == questions assert are_test_questions_customized(row) is True def test_mapping_view_builds_automatic_questions_from_remote_display_name() -> None: from app.gateway.routers import llmwiki as llmwiki_router row = _knowledge_profile_mapping_row( name="本地旧名称", questions=default_test_questions("更早的旧名称"), ) view = llmwiki_router._mapping_view( row, {"id": "remote-one", "name": "远程新名称"}, actor_user_id="user-one", is_admin=False, ) assert view["name"] == "远程新名称" assert view["test_questions"] == default_test_questions("远程新名称") assert view["test_questions_customized"] is False def test_empty_weknora_address_keeps_legacy_mode() -> None: runtime = resolve_llmwiki_runtime(LlmWikiConfig()) assert runtime.provider == "legacy" assert runtime.api_base_url == "" def test_configured_weknora_address_enables_weknora_mode() -> None: config = LlmWikiConfig.model_validate( { "weknora": { "api_base_url": "http://weknora-app:8080/", "web_base_url": "https://weknora.example.test/", "admin_email": "admin@example.test", "admin_password": "secret", } } ) runtime = resolve_llmwiki_runtime(config) assert runtime.provider == "weknora" assert runtime.api_base_url == "http://weknora-app:8080" assert runtime.web_base_url == "https://weknora.example.test" assert runtime.admin_email == "admin@example.test" assert runtime.admin_password == "secret" @pytest.mark.asyncio async def test_default_weknora_client_is_reused_on_the_same_event_loop() -> None: runtime = resolve_llmwiki_runtime( LlmWikiConfig.model_validate( { "weknora": { "api_base_url": "http://shared-weknora.test:8080", "admin_email": "admin@example.test", "admin_password": "secret", } } ) ) first = build_weknora_client(runtime) second = build_weknora_client(runtime) assert second is first def test_default_weknora_client_is_isolated_between_event_loops() -> None: runtime = resolve_llmwiki_runtime( LlmWikiConfig.model_validate( { "weknora": { "api_base_url": "http://loop-local-weknora.test:8080", "admin_email": "admin@example.test", "admin_password": "secret", } } ) ) async def build_pair() -> tuple[WeKnoraClient, WeKnoraClient]: return build_weknora_client(runtime), build_weknora_client(runtime) first_loop_pair = asyncio.run(build_pair()) second_loop_pair = asyncio.run(build_pair()) assert first_loop_pair[0] is first_loop_pair[1] assert second_loop_pair[0] is second_loop_pair[1] assert first_loop_pair[0] is not second_loop_pair[0] def test_runtime_override_can_force_legacy_or_reset_to_file() -> None: config = LlmWikiConfig.model_validate({"weknora": {"api_base_url": "http://weknora:8080"}}) forced_legacy = resolve_llmwiki_runtime( config, LlmWikiRuntimeOverride(enabled=True, api_base_url=""), ) reset_to_file = resolve_llmwiki_runtime( config, LlmWikiRuntimeOverride(enabled=False, api_base_url=""), ) assert forced_legacy.provider == "legacy" assert reset_to_file.provider == "weknora" def test_weknora_url_rejects_credentials_and_non_http_schemes() -> None: with pytest.raises(ValueError): LlmWikiConfig.model_validate({"weknora": {"api_base_url": "ftp://weknora.local"}}) with pytest.raises(ValueError): LlmWikiConfig.model_validate({"weknora": {"api_base_url": "http://user:pass@weknora.local"}}) def test_weknora_embed_follows_only_protected_binary_redirects() -> None: from app.gateway.routers import llmwiki as llmwiki_router assert llmwiki_router._should_follow_weknora_binary_redirect("/api/v1/knowledge/doc-one/preview") assert llmwiki_router._should_follow_weknora_binary_redirect("/api/v1/knowledge-bases/kb-one/files") assert not llmwiki_router._should_follow_weknora_binary_redirect("/api/v1/knowledge/doc-one") assert not llmwiki_router._should_follow_weknora_binary_redirect("/api/v1/knowledge-bases/kb-one") def test_agent_config_tracks_allowed_llmwiki_knowledge_bases() -> None: config = AgentConfig( name="Research agent", llmwiki_knowledge_base_ids=["kb-one", "kb-two", "kb-one"], ) assert config.llmwiki_knowledge_base_ids == ["kb-one", "kb-two"] def test_conversation_selection_is_deduplicated_and_absence_is_distinct() -> None: assert selected_knowledge_base_ids({}) is None assert selected_knowledge_base_ids({"llmwiki_knowledge_base_ids": []}) == [] assert selected_knowledge_base_ids({"llmwiki_knowledge_base_ids": ["kb-one", "kb-one", "", "kb-two"]}) == ["kb-one", "kb-two"] def test_llmwiki_rag_only_uses_explicit_conversation_selection() -> None: from app.gateway.llmwiki_rag import resolve_requested_llmwiki_knowledge_base_ids body = SimpleNamespace( assistant_id="agent-with-bound-kb", context={}, metadata={"llmwiki_knowledge_base_ids": ["kb-one", "kb-one", "", "kb-two"]}, ) assert resolve_requested_llmwiki_knowledge_base_ids(body) == ["kb-one", "kb-two"] def test_weknora_frame_bootstrap_accepts_query_token_only_for_frame_route() -> None: from app.gateway.deps import get_request_access_token frame_request = _request( "/api/llmwiki/knowledge-bases/local-kb/weknora/frame", query="access_token=deerflow-token", ) normal_request = _request("/api/models", query="access_token=deerflow-token") assert get_request_access_token(frame_request) == "deerflow-token" assert get_request_access_token(normal_request) is None def test_weknora_embed_cookie_bypasses_only_proxy_paths() -> None: from app.gateway import auth_middleware, csrf_middleware from app.gateway.weknora_embed import sign_weknora_embed_payload raw_cookie = sign_weknora_embed_payload( { "mapping_id": "local-kb", "weknora_id": "remote-kb", "user_id": "user-a", "is_admin": False, "can_write": False, "exp": 4_102_444_800, } ) cookies = {"deerflow_weknora_embed": raw_cookie} assert auth_middleware._is_weknora_embed_proxy_request(_request("/platform/knowledge-bases/kb", cookies=cookies)) assert auth_middleware._is_weknora_embed_proxy_request(_request("/deerflow/platform/knowledge-bases/kb", cookies=cookies)) assert auth_middleware._is_weknora_embed_proxy_request(_request("/assets/index.js", cookies=cookies)) assert auth_middleware._is_weknora_embed_proxy_request(_request("/deerflow/assets/index.js", cookies=cookies)) for static_path in ( "/deerflow/css/app.css", "/deerflow/fonts/app.woff2", "/deerflow/img/logo.png", "/deerflow/images/empty.svg", "/deerflow/js/app.js", "/deerflow/media/intro.mp4", "/deerflow/static/chunk.js", ): request = _request(static_path, cookies=cookies) assert auth_middleware._is_weknora_embed_proxy_request(request) assert csrf_middleware.is_weknora_embed_proxy_request(request) assert auth_middleware._is_weknora_embed_proxy_request(_request("/config.js", cookies=cookies)) assert auth_middleware._is_weknora_embed_proxy_request(_request("/tdesign-icons/0.4.1/fonts/index.js", cookies=cookies)) assert auth_middleware._is_weknora_embed_proxy_request(_request("/api/v1/knowledge-search", cookies=cookies)) assert auth_middleware._is_weknora_embed_proxy_request(_request("/deerflow/api/v1/knowledge-search", cookies=cookies)) assert auth_middleware._is_weknora_embed_proxy_request(_request("/api/llmwiki/weknora-embed/api/v1/auth/me", cookies=cookies)) assert auth_middleware._is_weknora_embed_proxy_request(_request("/deerflow/api/llmwiki/weknora-embed/api/v1/auth/me", cookies=cookies)) assert csrf_middleware.is_weknora_embed_proxy_request(_request("/api/v1/knowledge-search", cookies=cookies)) assert not auth_middleware._is_weknora_embed_proxy_request(_request("/api/v1/knowledge-search", cookies={"deerflow_weknora_embed": "signed"})) assert not csrf_middleware.is_weknora_embed_proxy_request(_request("/api/v1/knowledge-search", cookies={"deerflow_weknora_embed": "signed"})) assert not auth_middleware._is_weknora_embed_proxy_request(_request("/api/v1/auth/me", cookies=cookies)) assert not auth_middleware._is_weknora_embed_proxy_request(_request("/api/models", cookies=cookies)) assert not auth_middleware._is_weknora_embed_proxy_request(_request("/platform/knowledge-bases/kb")) @pytest.mark.asyncio async def test_weknora_session_endpoint_sets_cookie_and_returns_platform_path(monkeypatch: pytest.MonkeyPatch) -> None: from app.gateway.routers import llmwiki as llmwiki_router async def authorized_mapping(*_: object, **__: object) -> tuple[dict[str, str], str, bool]: return {"id": "local-kb", "weknora_id": "remote-kb", "name": "KB"}, "user-a", False class Store: async def get_authorized(self, *_: object, **__: object) -> dict[str, str]: return {"id": "local-kb", "weknora_id": "remote-kb"} async def list_visible(self, *_: object, **__: object) -> list[dict[str, str]]: raise AssertionError("creating an iframe session must not load every visible knowledge base") monkeypatch.setattr(llmwiki_router, "_runtime_for_iframe_proxy", lambda _: object()) monkeypatch.setattr(llmwiki_router, "_authorized_mapping", authorized_mapping) monkeypatch.setattr(llmwiki_router, "_store", lambda _: Store()) response = await llmwiki_router.create_weknora_detail_session( _request( "/api/llmwiki/knowledge-bases/local-kb/weknora/session", request_headers={"X-Forwarded-Proto": "https"}, ), "local-kb", tab="wiki", ) payload = json.loads(response.body) assert payload["frame_path"] == "/deerflow/platform/knowledge-bases/remote-kb?tab=wiki" cookie = response.headers["set-cookie"] assert "deerflow_weknora_embed=" in cookie assert "Secure" in cookie assert "SameSite=none" in cookie set_cookie_headers = [value.decode("latin-1") for key, value in response.raw_headers if key.lower() == b"set-cookie"] assert any("Path=/" in value for value in set_cookie_headers) assert any("Path=/deerflow" in value for value in set_cookie_headers) raw = cookie.split("deerflow_weknora_embed=", 1)[1].split(";", 1)[0] ctx = llmwiki_router._verify_embed_cookie(raw) assert ctx.mapping_id == "local-kb" assert ctx.weknora_id == "remote-kb" assert ctx.can_write is True assert ctx.allowed_weknora_ids == () assert len(raw) < 1024 @pytest.mark.asyncio async def test_weknora_embed_picker_hydrates_visible_bases_server_side(monkeypatch: pytest.MonkeyPatch) -> None: from app.gateway.routers import llmwiki as llmwiki_router calls: list[dict[str, object]] = [] class Store: async def list_visible(self, user_id: str, *, scope: str, is_admin: bool) -> list[dict[str, str]]: calls.append({"user_id": user_id, "scope": scope, "is_admin": is_admin}) return [ {"id": "local-kb", "weknora_id": "remote-kb"}, {"id": "public-kb", "weknora_id": "other-public-kb"}, ] monkeypatch.setattr(llmwiki_router, "_store", lambda _: Store()) ctx = llmwiki_router.WeKnoraEmbedContext( mapping_id="local-kb", weknora_id="remote-kb", user_id="user-a", is_admin=False, can_write=True, exp=4_102_444_800, ) hydrated = await llmwiki_router._with_visible_embed_knowledge_bases(_request("/platform/knowledge-bases/remote-kb"), ctx) assert calls == [{"user_id": "user-a", "scope": "all", "is_admin": False}] assert hydrated.allowed_weknora_ids == ("remote-kb", "other-public-kb") assert "allowed_weknora_ids" not in llmwiki_router._embed_context_payload(hydrated) @pytest.mark.asyncio async def test_weknora_session_endpoint_preserves_reverse_proxy_prefix(monkeypatch: pytest.MonkeyPatch) -> None: from app.gateway.routers import llmwiki as llmwiki_router async def authorized_mapping(*_: object, **__: object) -> tuple[dict[str, str], str, bool]: return {"id": "local-kb", "weknora_id": "remote-kb", "name": "KB"}, "user-a", False class Store: async def get_authorized(self, *_: object, **__: object) -> dict[str, str]: return {"id": "local-kb", "weknora_id": "remote-kb"} async def list_visible(self, *_: object, **__: object) -> list[dict[str, str]]: return [] monkeypatch.setattr(llmwiki_router, "_runtime_for_iframe_proxy", lambda _: object()) monkeypatch.setattr(llmwiki_router, "_authorized_mapping", authorized_mapping) monkeypatch.setattr(llmwiki_router, "_store", lambda _: Store()) response = await llmwiki_router.create_weknora_detail_session( _request( "/api/llmwiki/knowledge-bases/local-kb/weknora/session", request_headers={"X-Forwarded-Prefix": "/deerflow-api/"}, ), "local-kb", tab="wiki", ) payload = json.loads(response.body) assert payload["frame_path"] == "/deerflow-api/platform/knowledge-bases/remote-kb?tab=wiki" set_cookie_headers = [value.decode("latin-1") for key, value in response.raw_headers if key.lower() == b"set-cookie"] assert any("Path=/deerflow-api" in value for value in set_cookie_headers) def test_weknora_public_prefix_rejects_protocol_relative_header() -> None: from app.gateway.routers import llmwiki as llmwiki_router request = _request( "/api/llmwiki/knowledge-bases/local-kb/weknora/session", request_headers={"X-Forwarded-Prefix": "//untrusted.example"}, ) assert llmwiki_router._request_public_prefix(request) == "/deerflow" @pytest.mark.asyncio async def test_weknora_session_endpoint_uses_backend_prefix_when_frontend_has_different_prefix( monkeypatch: pytest.MonkeyPatch, ) -> None: from app.gateway.routers import llmwiki as llmwiki_router async def authorized_mapping(*_: object, **__: object) -> tuple[dict[str, str], str, bool]: return {"id": "local-kb", "weknora_id": "remote-kb", "name": "KB"}, "user-a", False class Store: async def get_authorized(self, *_: object, **__: object) -> dict[str, str]: return {"id": "local-kb", "weknora_id": "remote-kb"} async def list_visible(self, *_: object, **__: object) -> list[dict[str, str]]: return [] monkeypatch.setattr(llmwiki_router, "_runtime_for_iframe_proxy", lambda _: object()) monkeypatch.setattr(llmwiki_router, "_authorized_mapping", authorized_mapping) monkeypatch.setattr(llmwiki_router, "_store", lambda _: Store()) response = await llmwiki_router.create_weknora_detail_session( _request( "/api/llmwiki/knowledge-bases/local-kb/weknora/session", request_headers={"Referer": "https://example.test/magentweb/page/workspace/knowledge/local-kb"}, ), "local-kb", tab="wiki", ) payload = json.loads(response.body) assert payload["frame_path"] == "/deerflow/platform/knowledge-bases/remote-kb?tab=wiki" @pytest.mark.asyncio async def test_weknora_client_keeps_anonymous_mode_without_admin_credentials() -> None: requests: list[httpx.Request] = [] async def handler(request: httpx.Request) -> httpx.Response: requests.append(request) assert "X-API-Key" not in request.headers assert "Authorization" not in request.headers assert json.loads(request.content) == { "query": "deployment guide", "knowledge_base_ids": ["kb-one", "kb-two"], } return httpx.Response( 200, json={ "success": True, "data": [ { "id": "chunk-one", "content": "matched passage", "knowledge_id": "doc-one", "knowledge_base_id": "kb-one", "knowledge_title": "Guide", "score": 0.93, } ], }, ) async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as http_client: client = WeKnoraClient( base_url="http://weknora.local:8080", http_client=http_client, ) results = await client.search("deployment guide", ["kb-one", "kb-two"]) assert requests[0].url.path == "/api/v1/knowledge-search" assert results[0]["chunk_id"] == "chunk-one" assert results[0]["title"] == "Guide" @pytest.mark.asyncio async def test_weknora_client_uses_admin_token_when_credentials_are_configured() -> None: requests: list[httpx.Request] = [] async def handler(request: httpx.Request) -> httpx.Response: requests.append(request) if request.url.path == "/api/v1/auth/login": assert json.loads(request.content) == {"email": "admin@example.test", "password": "secret"} return httpx.Response( 200, json={ "success": True, "data": { "token": "token-one", "tenant": {"id": "tenant-one"}, }, }, ) assert request.headers["Authorization"] == "Bearer token-one" assert request.headers["X-Tenant-ID"] == "tenant-one" return httpx.Response(200, json={"success": True, "data": []}) async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as http_client: client = WeKnoraClient( base_url="http://weknora.local:8080", http_client=http_client, admin_email="admin@example.test", admin_password="secret", ) results = await client.search("deployment guide", ["kb-one"]) assert results == [] assert [request.url.path for request in requests] == [ "/api/v1/auth/login", "/api/v1/knowledge-search", ] @pytest.mark.asyncio async def test_weknora_client_coalesces_concurrent_admin_logins() -> None: login_count = 0 async def handler(request: httpx.Request) -> httpx.Response: nonlocal login_count if request.url.path == "/api/v1/auth/login": login_count += 1 await asyncio.sleep(0.01) return httpx.Response( 200, json={"success": True, "data": {"token": "shared-token"}}, ) assert request.headers["Authorization"] == "Bearer shared-token" return httpx.Response(200, json={"success": True, "data": []}) async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as http_client: client = WeKnoraClient( base_url="http://weknora.local:8080", http_client=http_client, admin_email="admin@example.test", admin_password="secret", ) await asyncio.gather( client.list_knowledge_bases(), client.list_knowledge_bases(), ) assert login_count == 1 @pytest.mark.asyncio async def test_weknora_client_caches_and_coalesces_knowledge_base_directory() -> None: list_count = 0 async def handler(request: httpx.Request) -> httpx.Response: nonlocal list_count assert request.url.path == "/api/v1/knowledge-bases" list_count += 1 await asyncio.sleep(0.01) return httpx.Response(200, json={"success": True, "data": [{"id": "kb-one", "name": "One"}]}) async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as http_client: client = WeKnoraClient(base_url="http://weknora.local:8080", http_client=http_client) first, second = await asyncio.gather( client.list_knowledge_bases(), client.list_knowledge_bases(), ) cached = await client.list_knowledge_bases() assert list_count == 1 assert first == second == cached == [{"id": "kb-one", "name": "One"}] assert first is not second @pytest.mark.asyncio async def test_weknora_client_invalidates_knowledge_base_directory_after_mutation() -> None: list_count = 0 async def handler(request: httpx.Request) -> httpx.Response: nonlocal list_count if request.method == "GET": list_count += 1 return httpx.Response(200, json={"success": True, "data": [{"id": "kb-one", "name": f"One-{list_count}"}]}) assert request.method == "PUT" return httpx.Response(200, json={"success": True, "data": {"id": "kb-one", "name": "Renamed"}}) async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as http_client: client = WeKnoraClient(base_url="http://weknora.local:8080", http_client=http_client) assert (await client.list_knowledge_bases())[0]["name"] == "One-1" await client.update_knowledge_base("kb-one", {"name": "Renamed"}) assert (await client.list_knowledge_bases())[0]["name"] == "One-2" assert list_count == 2 @pytest.mark.asyncio async def test_weknora_client_invalidates_knowledge_base_directory_after_create_and_delete() -> None: list_count = 0 async def handler(request: httpx.Request) -> httpx.Response: nonlocal list_count if request.method == "GET": list_count += 1 return httpx.Response(200, json={"success": True, "data": [{"id": "kb-one", "name": f"One-{list_count}"}]}) if request.method == "POST": return httpx.Response(201, json={"success": True, "data": {"id": "kb-created", "name": "Created"}}) assert request.method == "DELETE" return httpx.Response(204) async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as http_client: client = WeKnoraClient(base_url="http://weknora.local:8080", http_client=http_client) assert (await client.list_knowledge_bases())[0]["name"] == "One-1" await client.create_knowledge_base(name="Created", embedding_model_id="embed-one") assert (await client.list_knowledge_bases())[0]["name"] == "One-2" await client.delete_knowledge_base("kb-created") assert (await client.list_knowledge_bases())[0]["name"] == "One-3" assert list_count == 3 @pytest.mark.asyncio async def test_weknora_client_invalidates_knowledge_base_directory_after_document_upload() -> None: list_count = 0 async def handler(request: httpx.Request) -> httpx.Response: nonlocal list_count if request.method == "GET": list_count += 1 return httpx.Response( 200, json={ "success": True, "data": [{"id": "kb-one", "processing_count": list_count}], }, ) assert request.url.path == "/api/v1/knowledge-bases/kb-one/knowledge/file" return httpx.Response(201, json={"success": True, "data": {"id": "doc-one"}}) async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as http_client: client = WeKnoraClient(base_url="http://weknora.local:8080", http_client=http_client) assert (await client.list_knowledge_bases())[0]["processing_count"] == 1 await client.upload_document("kb-one", filename="guide.md", content=b"# Guide") assert (await client.list_knowledge_bases())[0]["processing_count"] == 2 assert list_count == 2 @pytest.mark.asyncio async def test_weknora_client_refreshes_admin_token_after_rejection() -> None: login_count = 0 search_count = 0 async def handler(request: httpx.Request) -> httpx.Response: nonlocal login_count, search_count if request.url.path == "/api/v1/auth/login": login_count += 1 return httpx.Response( 200, json={"success": True, "data": {"token": f"token-{login_count}"}}, ) search_count += 1 if search_count == 1: assert request.headers["Authorization"] == "Bearer token-1" return httpx.Response(401, json={"message": "expired"}) assert request.headers["Authorization"] == "Bearer token-2" return httpx.Response(200, json={"success": True, "data": []}) async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as http_client: client = WeKnoraClient( base_url="http://weknora.local:8080", http_client=http_client, admin_email="admin@example.test", admin_password="secret", ) await client.search("deployment guide", ["kb-one"]) assert login_count == 2 assert search_count == 2 @pytest.mark.asyncio async def test_weknora_client_merges_required_name_for_description_update() -> None: requests: list[httpx.Request] = [] async def handler(request: httpx.Request) -> httpx.Response: requests.append(request) if request.method == "GET": return httpx.Response( 200, json={"success": True, "data": {"id": "kb-one", "name": "对话沉淀"}}, ) assert request.method == "PUT" assert json.loads(request.content) == {"name": "对话沉淀", "description": "cmzs description"} return httpx.Response( 200, json={"success": True, "data": {"id": "kb-one", "name": "对话沉淀", "description": "cmzs description"}}, ) async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as http_client: client = WeKnoraClient(base_url="http://weknora.local:8080", http_client=http_client) updated = await client.update_knowledge_base("kb-one", {"description": "cmzs description"}) assert updated["description"] == "cmzs description" assert [request.method for request in requests] == ["GET", "PUT"] @pytest.mark.asyncio async def test_weknora_client_merges_wiki_instructions_without_losing_runtime_config() -> None: requests: list[httpx.Request] = [] async def handler(request: httpx.Request) -> httpx.Response: requests.append(request) if request.method == "GET": return httpx.Response( 200, json={ "success": True, "data": { "id": "kb-one", "name": "Operational Wiki", "wiki_config": { "synthesis_model_id": "qa-one", "extraction_granularity": "standard", "content_instructions": "old content guidance", "extraction_instructions": "old extraction guidance", }, }, }, ) assert request.method == "PUT" assert json.loads(request.content) == { "name": "Operational Wiki", "config": { "wiki_config": { "synthesis_model_id": "qa-one", "extraction_granularity": "standard", "content_instructions": "先结论后依据", "extraction_instructions": "old extraction guidance", } }, } return httpx.Response( 200, json={ "success": True, "data": { "id": "kb-one", "name": "Operational Wiki", "wiki_config": json.loads(request.content)["config"]["wiki_config"], }, }, ) async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as http_client: client = WeKnoraClient(base_url="http://weknora.local:8080", http_client=http_client) updated = await client.update_knowledge_base( "kb-one", { "wiki_config": { "content_instructions": " 先结论后依据 ", } }, ) assert updated["wiki_config"]["synthesis_model_id"] == "qa-one" assert [request.method for request in requests] == ["GET", "PUT"] @pytest.mark.asyncio async def test_weknora_client_resolves_default_embedding_model_inside_weknora() -> None: async def handler(request: httpx.Request) -> httpx.Response: if request.method == "GET" and request.url.path == "/api/v1/models": return httpx.Response( 200, json={ "success": True, "data": [ {"id": "embed-default", "type": "Embedding", "is_default": True}, ], }, ) assert request.method == "POST" assert request.url.path == "/api/v1/knowledge-bases" assert json.loads(request.content)["embedding_model_id"] == "embed-default" return httpx.Response(201, json={"success": True, "data": {"id": "kb-created", "name": "Docs"}}) async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as http_client: client = WeKnoraClient( base_url="http://weknora.local:8080", http_client=http_client, ) created = await client.create_knowledge_base(name="Docs") assert created["id"] == "kb-created" @pytest.mark.asyncio async def test_weknora_wiki_create_reuses_catalog_models_without_debug_probe() -> None: requests: list[tuple[str, str, dict[str, object]]] = [] async def handler(request: httpx.Request) -> httpx.Response: payload = json.loads(request.content) if request.content else {} requests.append((request.method, request.url.path, payload)) assert request.method == "POST" assert request.url.path == "/api/v1/knowledge-bases" return httpx.Response(201, json={"success": True, "data": {"id": "wiki-created", "name": "Wiki"}}) async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as http_client: client = WeKnoraClient(base_url="http://weknora.local:8080", http_client=http_client) await client.create_knowledge_base( name="Wiki", embedding_model_id="embedding-id", knowledge_qa_model_id="qa-id", wiki_enabled=True, ) assert len(requests) == 1 assert requests[0][2]["embedding_model_id"] == "embedding-id" assert requests[0][2]["summary_model_id"] == "qa-id" assert requests[0][2]["wiki_config"]["synthesis_model_id"] == "qa-id" @pytest.mark.asyncio async def test_gateway_lists_safe_embedding_models_and_uses_explicit_selection( monkeypatch: pytest.MonkeyPatch, ) -> None: from app.gateway.routers import llmwiki as llmwiki_router create_calls: list[dict[str, object]] = [] class Client: async def list_models(self) -> list[dict[str, object]]: return [ { "id": "embed-local", "type": "Embedding", "name": "Local BGE", "provider": "local", "status": "active", "is_default": True, "api_key": "must-not-leak", "config": {"secret": "must-not-leak"}, }, { "id": "embed-openai", "type": "embedding", "display_name": "OpenAI text-embedding", "source": "remote", "parameters": { "model": "text-embedding-3-large", "provider": "openai", "api_key": "must-not-leak", }, "status": "active", }, { "id": "embed-disabled", "type": "Embedding", "name": "Disabled", "status": "inactive", }, {"id": "chat-one", "type": "LLM", "name": "chat"}, {"id": "qa-default", "type": "KnowledgeQA", "name": "wiki-answer", "status": "active", "is_default": True}, ] async def create_knowledge_base(self, **kwargs: object) -> dict[str, object]: create_calls.append(kwargs) return { "id": "remote-openai", "name": kwargs["name"], "description": kwargs["description"], "type": kwargs["kb_type"], "embedding_model_id": kwargs["embedding_model_id"], } async def actor(_: object) -> tuple[str, bool]: return "user-a", False store = MemoryLlmWikiStore() client = Client() monkeypatch.setattr(llmwiki_router, "_actor", actor) 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) listed = await llmwiki_router.list_embedding_models(object()) assert listed["default_model_id"] == "embed-local" assert [item["id"] for item in listed["models"]] == [ "embed-local", "embed-openai", "embed-disabled", ] assert listed["models"][1]["provider"] == "openai" assert listed["models"][1]["model_name"] == "text-embedding-3-large" assert listed["models"][2]["available"] is False assert "api_key" not in listed["models"][0] assert "config" not in listed["models"][0] assert "parameters" not in listed["models"][1] created = await llmwiki_router.create_knowledge_base( object(), llmwiki_router.KnowledgeBaseCreate( name="OpenAI docs", embedding_model_id="embed-openai", wiki_config=llmwiki_router.WikiGenerationInstructions( content_instructions="按事件时间线生成页面", extraction_instructions="重点提取人物和组织关系", ), ), ) assert create_calls[0]["embedding_model_id"] == "embed-openai" assert create_calls[0]["knowledge_qa_model_id"] == "qa-default" assert create_calls[0]["wiki_enabled"] is True assert create_calls[0]["wiki_content_instructions"] == "按事件时间线生成页面" assert create_calls[0]["wiki_extraction_instructions"] == "重点提取人物和组织关系" assert created["embedding_model_id"] == "embed-openai" assert created["retrieval_settings"] == { "entity_expansion_enabled": True, "max_related_terms": 6, "statistics_enabled": True, } for rejected_model_id in ("embed-disabled", "chat-one", "missing"): with pytest.raises(HTTPException) as exc_info: await llmwiki_router.create_knowledge_base( object(), llmwiki_router.KnowledgeBaseCreate( name="Rejected docs", embedding_model_id=rejected_model_id, ), ) assert exc_info.value.status_code == 422 assert len(create_calls) == 1 @pytest.mark.asyncio async def test_gateway_forwards_wiki_instructions_on_knowledge_base_edit( monkeypatch: pytest.MonkeyPatch, ) -> None: from app.gateway.routers import llmwiki as llmwiki_router forwarded: list[tuple[str, dict[str, object]]] = [] row = { "id": "local-wiki", "weknora_id": "remote-wiki", "owner_user_id": "user-a", "name": "Operations", "description": "", "kb_type": "document", "publication_status": "private", } class Client: async def get_knowledge_base( self, knowledge_base_id: str, ) -> dict[str, object]: return { "id": knowledge_base_id, "name": row["name"], "description": row["description"], "type": "document", "indexing_strategy": {"wiki_enabled": True}, "wiki_config": {}, } async def update_knowledge_base( self, knowledge_base_id: str, changes: dict[str, object], ) -> dict[str, object]: forwarded.append((knowledge_base_id, changes)) return { "id": knowledge_base_id, "name": row["name"], "description": row["description"], "type": "document", "indexing_strategy": {"wiki_enabled": True}, "wiki_config": changes["wiki_config"], } class Store: async def update_mapping( self, *_: object, **changes: object, ) -> dict[str, object]: return {**row, **changes} async def authorized_mapping(*_: object, **__: object) -> tuple[dict[str, object], str, bool]: return row, "user-a", False monkeypatch.setattr(llmwiki_router, "_authorized_mapping", authorized_mapping) monkeypatch.setattr(llmwiki_router, "_client_or_503", lambda: Client()) monkeypatch.setattr(llmwiki_router, "_store", lambda _: Store()) updated = await llmwiki_router.update_knowledge_base( object(), "local-wiki", llmwiki_router.KnowledgeBaseUpdate( wiki_config=llmwiki_router.WikiGenerationInstructions( content_instructions="正式、简洁,保留关键数字", extraction_instructions="优先人物、组织和时间线", ) ), ) assert forwarded == [ ( "remote-wiki", { "wiki_config": { "content_instructions": "正式、简洁,保留关键数字", "extraction_instructions": "优先人物、组织和时间线", } }, ) ] assert updated["wiki_config"]["content_instructions"] == "正式、简洁,保留关键数字" assert updated["wiki_config"]["extraction_instructions"] == "优先人物、组织和时间线" assert updated["wiki_config"]["wiki_config_schema_version"] == 3 assert updated["wiki_config"]["wiki_config_revision"] == 1 assert updated["wiki_config"]["apply_status"] == "active" @pytest.mark.asyncio async def test_gateway_embedding_model_list_allows_empty_configuration( monkeypatch: pytest.MonkeyPatch, ) -> None: from app.gateway.routers import llmwiki as llmwiki_router class Client: async def list_models(self) -> list[dict[str, object]]: return [{"id": "chat-only", "type": "LLM"}] async def actor(_: object) -> tuple[str, bool]: return "user-a", False monkeypatch.setattr(llmwiki_router, "_actor", actor) monkeypatch.setattr(llmwiki_router, "_client_or_503", lambda: Client()) assert await llmwiki_router.list_embedding_models(object()) == { "models": [], "default_model_id": None, } @pytest.mark.asyncio async def test_weknora_client_creates_wiki_with_weknora_owned_models() -> None: model_reads = 0 async def handler(request: httpx.Request) -> httpx.Response: nonlocal model_reads if request.method == "GET" and request.url.path == "/api/v1/models": model_reads += 1 return httpx.Response( 200, json={ "data": [ {"id": "embed-one", "type": "Embedding", "is_default": True}, {"id": "qa-one", "type": "KnowledgeQA", "status": "active"}, ] }, ) if request.url.path == "/api/v1/models/qa-one/debug": assert request.method == "POST" assert request.headers["content-type"].startswith("multipart/form-data;") assert b"Reply with OK." in request.content return httpx.Response(200, json={"data": {"ok": True, "elapsed_ms": 12}}) payload = json.loads(request.content) assert payload["embedding_model_id"] == "embed-one" assert payload["summary_model_id"] == "qa-one" assert payload["indexing_strategy"] == { "vector_enabled": True, "keyword_enabled": True, "wiki_enabled": True, "graph_enabled": False, } assert payload["wiki_config"]["synthesis_model_id"] == "qa-one" assert payload["wiki_config"]["content_instructions"] == "使用正式中文并保留关键数字" assert payload["wiki_config"]["extraction_instructions"] == "重点提取事件、人物与组织" return httpx.Response(201, json={"data": {"id": "wiki-one", "name": "Wiki"}}) async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as http_client: client = WeKnoraClient(base_url="http://weknora.local:8080", http_client=http_client) created = await client.create_knowledge_base( name="Wiki", wiki_enabled=True, wiki_content_instructions=" 使用正式中文并保留关键数字 ", wiki_extraction_instructions=" 重点提取事件、人物与组织 ", ) assert created["id"] == "wiki-one" assert model_reads == 2 @pytest.mark.asyncio async def test_weknora_client_prefers_a_healthy_explicit_knowledge_qa_default() -> None: attempts: list[str] = [] async def handler(request: httpx.Request) -> httpx.Response: if request.method == "GET" and request.url.path == "/api/v1/models": return httpx.Response( 200, json={ "data": [ { "id": "alphabetical-fallback", "name": "Alpha", "type": "KnowledgeQA", "status": "active", "is_default": False, }, { "id": "configured-default", "name": "Zulu", "type": "KnowledgeQA", "status": "active", "is_default": True, }, ] }, ) attempts.append(request.url.path) return httpx.Response(200, json={"data": {"ok": True}}) async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as http_client: client = WeKnoraClient(base_url="http://weknora.local:8080", http_client=http_client) model_id = await client.get_default_knowledge_qa_model_id() assert model_id == "configured-default" assert attempts == ["/api/v1/models/configured-default/debug"] @pytest.mark.asyncio async def test_weknora_client_uses_a_stable_healthy_knowledge_qa_fallback() -> None: attempts: list[str] = [] async def handler(request: httpx.Request) -> httpx.Response: if request.method == "GET" and request.url.path == "/api/v1/models": return httpx.Response( 200, json={ "data": [ { "id": "healthy-zulu", "display_name": "Zulu", "type": "KnowledgeQA", "status": "active", "is_default": False, }, { "id": "inactive-alpha", "display_name": "Alpha", "type": "KnowledgeQA", "status": "inactive", "is_default": False, }, { "id": "disabled-beta", "display_name": "Beta", "type": "KnowledgeQA", "status": "active", "is_default": False, }, ] }, ) attempts.append(request.url.path) ok = request.url.path.endswith("/healthy-zulu/debug") return httpx.Response( 200, json={"data": {"ok": ok, "error": "Model disabled." if not ok else ""}}, ) async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as http_client: client = WeKnoraClient(base_url="http://weknora.local:8080", http_client=http_client) model_id = await client.get_default_knowledge_qa_model_id() assert model_id == "healthy-zulu" assert attempts == [ "/api/v1/models/disabled-beta/debug", "/api/v1/models/healthy-zulu/debug", ] @pytest.mark.asyncio async def test_weknora_client_rejects_wiki_creation_when_no_knowledge_qa_model_is_usable() -> None: knowledge_base_creates = 0 async def handler(request: httpx.Request) -> httpx.Response: nonlocal knowledge_base_creates if request.method == "GET" and request.url.path == "/api/v1/models": return httpx.Response( 200, json={ "data": [ {"id": "embed-one", "type": "Embedding", "is_default": True}, { "id": "disabled-alpha", "display_name": "Alpha disabled", "type": "KnowledgeQA", "status": "active", "is_default": False, }, { "id": "broken-zulu", "display_name": "Zulu broken", "type": "KnowledgeQA", "status": "active", "is_default": False, }, ] }, ) if request.url.path == "/api/v1/models/disabled-alpha/debug": return httpx.Response(200, json={"data": {"ok": False, "error": "Model disabled."}}) if request.url.path == "/api/v1/models/broken-zulu/debug": return httpx.Response(200, json={"data": {"ok": False, "error": "Provider rejected the model."}}) knowledge_base_creates += 1 return httpx.Response(201, json={"data": {"id": "must-not-be-created"}}) 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(WeKnoraError) as raised: await client.create_knowledge_base(name="Wiki", wiki_enabled=True) assert raised.value.status_code == 400 assert "Zulu broken, Alpha disabled" in str(raised.value) assert "working default model" in str(raised.value) assert knowledge_base_creates == 0 @pytest.mark.asyncio async def test_weknora_client_preserves_model_debug_service_failures() -> None: attempts: list[str] = [] async def handler(request: httpx.Request) -> httpx.Response: if request.method == "GET" and request.url.path == "/api/v1/models": return httpx.Response( 200, json={ "data": [ { "id": "alpha", "name": "Alpha", "type": "KnowledgeQA", "status": "active", }, { "id": "zulu", "name": "Zulu", "type": "KnowledgeQA", "status": "active", }, ] }, ) attempts.append(request.url.path) return httpx.Response(503, json={"message": "model debug service unavailable"}) 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(WeKnoraError) as raised: await client.get_default_knowledge_qa_model_id() assert raised.value.status_code == 502 assert str(raised.value) == "model debug service unavailable" assert attempts == ["/api/v1/models/alpha/debug"] @pytest.mark.asyncio async def test_weknora_client_ignores_wiki_instructions_when_wiki_is_disabled() -> None: async def handler(request: httpx.Request) -> httpx.Response: if request.method == "GET": return httpx.Response( 200, json={"data": [{"id": "embed-one", "type": "Embedding", "is_default": True}]}, ) payload = json.loads(request.content) assert payload["embedding_model_id"] == "embed-one" assert "wiki_config" not in payload assert "summary_model_id" not in payload return httpx.Response(201, json={"data": {"id": "docs-one", "name": "Docs"}}) async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as http_client: client = WeKnoraClient(base_url="http://weknora.local:8080", http_client=http_client) created = await client.create_knowledge_base( name="Docs", wiki_enabled=False, wiki_content_instructions="must not be forwarded", wiki_extraction_instructions="must not be forwarded", ) assert created["id"] == "docs-one" @pytest.mark.asyncio async def test_weknora_client_imports_url_manual_content_and_lists_wiki_pages() -> None: async def handler(request: httpx.Request) -> httpx.Response: if request.url.path.endswith("/knowledge/url"): assert json.loads(request.content) == {"url": "https://example.test/guide"} return httpx.Response(201, json={"data": {"id": "url-one"}}) if request.url.path.endswith("/knowledge/manual"): assert json.loads(request.content) == { "title": "Guide", "content": "# Guide", "status": "publish", } return httpx.Response(201, json={"data": {"id": "manual-one"}}) assert request.url.path == "/api/v1/knowledgebase/kb-one/wiki/pages" assert request.url.params["query"] == "guide" return httpx.Response( 200, json={"pages": [{"slug": "guide", "title": "Guide"}], "total": 1, "page": 1, "page_size": 50}, ) async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as http_client: client = WeKnoraClient(base_url="http://weknora.local:8080", http_client=http_client) await client.import_document_url("kb-one", url="https://example.test/guide") await client.create_manual_document("kb-one", title="Guide", content="# Guide") pages = await client.list_wiki_pages("kb-one", query="guide") assert pages["total"] == 1 assert pages["pages"][0]["slug"] == "guide" @pytest.mark.asyncio async def test_weknora_client_uses_native_wiki_regex_search_contract() -> None: async def handler(request: httpx.Request) -> httpx.Response: assert request.method == "GET" assert request.url.path == "/api/v1/knowledgebase/kb-one/wiki/search" assert request.url.params["q"] == "火炬|开幕|闭幕" # WeKnora v0.7.2 caps the endpoint at 50. assert request.url.params["limit"] == "50" return httpx.Response( 200, json={ "pages": [ { "slug": "games/timeline", "title": "赛事时间线", "content": "火炬传递、开幕和闭幕安排。", } ] }, ) async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as http_client: client = WeKnoraClient(base_url="http://weknora.local:8080", http_client=http_client) pages = await client.search_wiki_pages( "kb-one", "火炬|开幕|闭幕", limit=500, ) assert pages == [ { "slug": "games/timeline", "title": "赛事时间线", "content": "火炬传递、开幕和闭幕安排。", } ] @pytest.mark.asyncio async def test_weknora_client_reads_ordered_wiki_index_groups() -> None: async def handler(request: httpx.Request) -> httpx.Response: assert request.url.path == "/api/v1/knowledgebase/kb-one/wiki/index" assert request.url.params["types"] == "entity,concept" assert request.url.params["limit"] == "50" assert request.url.params["cursor"] == "next-page" return httpx.Response( 200, json={ "data": { "intro": "# Wiki Index", "version": 2, "groups": [ { "type": "entity", "total": 1, "items": [{"slug": "entity/one", "title": "实体一"}], } ], } }, ) async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as http_client: client = WeKnoraClient(base_url="http://weknora.local:8080", http_client=http_client) index = await client.get_wiki_index( "kb-one", types=["entity", "concept"], limit=50, cursor="next-page", ) assert index["groups"][0]["items"][0]["slug"] == "entity/one" @pytest.mark.asyncio async def test_weknora_client_creates_updates_and_deletes_wiki_pages() -> None: seen: list[tuple[str, str]] = [] async def handler(request: httpx.Request) -> httpx.Response: seen.append((request.method, request.url.path)) if request.method == "POST": assert request.url.path == "/api/v1/knowledgebase/kb-one/wiki/pages" assert json.loads(request.content) == {"slug": "folder/guide", "title": "Guide", "content": "# Guide"} return httpx.Response(200, json={"success": True, "data": {"slug": "folder/guide", "title": "Guide"}}) assert request.url.path == "/api/v1/knowledgebase/kb-one/wiki/pages/folder/guide" if request.method == "PUT": assert json.loads(request.content) == {"title": "Guide", "content": "# Guide"} return httpx.Response(200, json={"success": True, "data": {"slug": "folder/guide", "title": "Guide"}}) if request.method == "DELETE": return httpx.Response(204) return httpx.Response(405) async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as http_client: client = WeKnoraClient(base_url="http://weknora.local:8080", http_client=http_client) created = await client.create_wiki_page("kb-one", {"slug": "folder/guide", "title": "Guide", "content": "# Guide"}) updated = await client.update_wiki_page("kb-one", "folder/guide", {"title": "Guide", "content": "# Guide"}) await client.delete_wiki_page("kb-one", "folder/guide") assert created["slug"] == "folder/guide" assert updated["slug"] == "folder/guide" assert seen == [ ("POST", "/api/v1/knowledgebase/kb-one/wiki/pages"), ("PUT", "/api/v1/knowledgebase/kb-one/wiki/pages/folder/guide"), ("DELETE", "/api/v1/knowledgebase/kb-one/wiki/pages/folder/guide"), ] @pytest.mark.asyncio async def test_weknora_client_reads_document_chunks_and_wiki_graph() -> None: async def handler(request: httpx.Request) -> httpx.Response: if request.url.path == "/api/v1/chunks/doc-one": assert request.url.params["page"] == "2" assert request.url.params["page_size"] == "10" return httpx.Response( 200, json={ "success": True, "data": [{"id": "chunk-one", "content": "parsed content", "chunk_index": 10}], "total": 31, "page": 2, "page_size": 10, }, ) assert request.url.path == "/api/v1/knowledgebase/kb-one/wiki/graph" assert request.url.params["mode"] == "overview" assert request.url.params["limit"] == "50" return httpx.Response( 200, json={ "success": True, "data": { "nodes": [{"slug": "entity/one", "title": "Entity One", "page_type": "entity"}], "edges": [], "meta": {"mode": "overview", "total": 1, "returned": 1}, }, }, ) async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as http_client: client = WeKnoraClient(base_url="http://weknora.local:8080", http_client=http_client) chunks = await client.list_chunks("doc-one", page=2, page_size=10) graph = await client.get_wiki_graph("kb-one", limit=50) assert chunks["total"] == 31 assert chunks["items"][0]["content"] == "parsed content" assert graph["nodes"][0]["slug"] == "entity/one" @pytest.mark.asyncio async def test_weknora_client_proxies_original_document_preview_without_credentials() -> None: async def handler(request: httpx.Request) -> httpx.Response: assert request.url.path == "/api/v1/knowledge/doc-one/preview" assert "X-API-Key" not in request.headers assert "Authorization" not in request.headers return httpx.Response(200, content=b"preview", headers={"Content-Type": "text/plain; charset=utf-8"}) async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as http_client: client = WeKnoraClient(base_url="http://weknora.local:8080", http_client=http_client) content, media_type = await client.get_document_preview("doc-one") assert content == b"preview" assert media_type == "text/plain" @pytest.mark.asyncio async def test_weknora_client_fetches_protected_wiki_image() -> None: async def handler(request: httpx.Request) -> httpx.Response: assert request.url.path == "/api/v1/knowledge-bases/kb-one/files" assert request.url.params["file_path"] == "resource://wiki/figure.png" return httpx.Response(200, content=b"png-bytes", headers={"Content-Type": "image/png"}) async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as http_client: client = WeKnoraClient(base_url="http://weknora.local:8080", http_client=http_client) content, media_type = await client.get_knowledge_base_file( "kb-one", "resource://wiki/figure.png", ) assert content == b"png-bytes" assert media_type == "image/png" @pytest.mark.asyncio async def test_memory_store_enforces_owner_and_public_visibility() -> None: store = MemoryLlmWikiStore() created = await store.create_mapping( weknora_id="wk-private", owner_user_id="user-a", name="Private notes", description="", kb_type="document", ) assert [row["id"] for row in await store.list_visible("user-a", scope="personal")] == [created["id"]] assert await store.list_visible("user-b", scope="personal") == [] assert await store.get_authorized(created["id"], "user-b", write=False, is_admin=False) is None published = await store.request_publish(created["id"], "user-a", is_admin=False) assert published and published["publication_status"] == "published" assert [row["id"] for row in await store.list_visible("user-b", scope="public")] == [created["id"]] assert await store.get_authorized(created["id"], "user-b", write=False, is_admin=False) is not None assert await store.get_authorized(created["id"], "user-b", write=True, is_admin=False) is None @pytest.mark.asyncio async def test_conversation_deposit_mapping_is_system_owned_and_published(monkeypatch: pytest.MonkeyPatch) -> None: from types import SimpleNamespace from app.gateway import llmwiki_deposit class FakeClient: async def list_knowledge_bases(self, *, refresh: bool = False) -> list[dict[str, object]]: return [] async def create_knowledge_base(self, **_: object) -> dict[str, object]: return { "id": "remote-deposit", "name": "对话沉淀", "description": "global", "type": "document", } store = MemoryLlmWikiStore() app = SimpleNamespace(state=SimpleNamespace(llmwiki_store=store)) monkeypatch.setattr(llmwiki_deposit, "_client", lambda _: FakeClient()) row = await llmwiki_deposit.ensure_conversation_deposit_knowledge_base(app, owner_user_id="user-a") assert row is not None assert row["owner_user_id"] == "system" assert row["publication_status"] == "published" assert row["description"] == llmwiki_deposit.CONVERSATION_DEPOSIT_KB_DESCRIPTION assert row["id"] not in [item["id"] for item in await store.list_visible("user-a", scope="personal")] assert [item["id"] for item in await store.list_visible("user-a", scope="public")] == [row["id"]] markdown = llmwiki_deposit._format_markdown( llmwiki_deposit.ConversationTurnDeposit( thread_id="thread-one", question="question", answer="answer", human_message_id="human-one", assistant_message_id="assistant-one", assistant_id=None, knowledge_base_ids=[], created_by_user_id="user-a", ), "title", ) assert "DeerFlow" not in markdown assert "source: cmzs" in markdown assert "#cmzs" in markdown @pytest.mark.asyncio async def test_existing_conversation_deposit_description_is_refreshed(monkeypatch: pytest.MonkeyPatch) -> None: from types import SimpleNamespace from app.gateway import llmwiki_deposit updates: list[tuple[str, dict[str, object]]] = [] class FakeClient: async def update_knowledge_base(self, knowledge_base_id: str, changes: dict[str, object]) -> dict[str, object]: updates.append((knowledge_base_id, changes)) return {"id": knowledge_base_id, **changes} store = MemoryLlmWikiStore() created = await store.create_mapping( weknora_id="remote-deposit", owner_user_id="system", name=llmwiki_deposit.CONVERSATION_DEPOSIT_KB_NAME, description="old description", kb_type="document", ) await store.request_publish(created["id"], "system", is_admin=True) app = SimpleNamespace(state=SimpleNamespace(llmwiki_store=store)) monkeypatch.setattr(llmwiki_deposit, "_client", lambda _: FakeClient()) row = await llmwiki_deposit.ensure_conversation_deposit_knowledge_base(app, owner_user_id="user-a") assert row is not None assert row["description"] == llmwiki_deposit.CONVERSATION_DEPOSIT_KB_DESCRIPTION assert updates == [ ( "remote-deposit", {"description": llmwiki_deposit.CONVERSATION_DEPOSIT_KB_DESCRIPTION}, ) ] @pytest.mark.asyncio async def test_legacy_user_owned_conversation_deposit_is_adopted(monkeypatch: pytest.MonkeyPatch) -> None: from types import SimpleNamespace from app.gateway import llmwiki_deposit class FakeClient: async def list_knowledge_bases(self, *, refresh: bool = False) -> list[dict[str, object]]: return [ { "id": "remote-deposit", "name": llmwiki_deposit.CONVERSATION_DEPOSIT_KB_NAME, "description": "DeerFlow legacy description", } ] async def update_knowledge_base(self, knowledge_base_id: str, changes: dict[str, object]) -> dict[str, object]: return {"id": knowledge_base_id, **changes} store = MemoryLlmWikiStore() legacy = await store.create_mapping( weknora_id="remote-deposit", owner_user_id="user-a", name=llmwiki_deposit.CONVERSATION_DEPOSIT_KB_NAME, description="DeerFlow legacy description", kb_type="document", ) await store.request_publish(legacy["id"], "user-a", is_admin=False) app = SimpleNamespace(state=SimpleNamespace(llmwiki_store=store)) monkeypatch.setattr(llmwiki_deposit, "_client", lambda _: FakeClient()) row = await llmwiki_deposit.ensure_conversation_deposit_knowledge_base(app, owner_user_id="user-a") assert row is not None assert row["id"] == legacy["id"] assert row["owner_user_id"] == "system" assert row["publication_status"] == "published" assert row["description"] == llmwiki_deposit.CONVERSATION_DEPOSIT_KB_DESCRIPTION assert await store.list_visible("user-a", scope="personal") == [] @pytest.mark.asyncio async def test_listing_conversation_deposit_refreshes_remote_description(monkeypatch: pytest.MonkeyPatch) -> None: from app.gateway import llmwiki_deposit from app.gateway.routers import llmwiki as llmwiki_router updates: list[tuple[str, dict[str, object]]] = [] store = MemoryLlmWikiStore() created = await store.create_mapping( weknora_id="remote-deposit", owner_user_id="system", name=llmwiki_deposit.CONVERSATION_DEPOSIT_KB_NAME, description="DeerFlow legacy description", kb_type="document", ) await store.request_publish(created["id"], "system", is_admin=True) class Client: async def list_knowledge_bases(self, *, refresh: bool = False) -> list[dict[str, object]]: return [ { "id": "remote-deposit", "name": llmwiki_deposit.CONVERSATION_DEPOSIT_KB_NAME, "description": "DeerFlow legacy description", } ] async def create_knowledge_base(self, **_: object) -> dict[str, object]: raise AssertionError("The existing deposit should be reused") async def update_knowledge_base(self, knowledge_base_id: str, changes: dict[str, object]) -> dict[str, object]: updates.append((knowledge_base_id, changes)) return {"id": knowledge_base_id, **changes} async def actor(_: object) -> tuple[str, bool]: return "user-a", False monkeypatch.setattr(llmwiki_router, "_actor", actor) monkeypatch.setattr(llmwiki_router, "_store", lambda _: store) monkeypatch.setattr(llmwiki_router, "_client_or_503", lambda: Client()) payload = await llmwiki_router.list_knowledge_bases( object(), scope="public", refresh=True, ) assert payload["knowledge_bases"][0]["description"] == llmwiki_deposit.CONVERSATION_DEPOSIT_KB_DESCRIPTION assert "DeerFlow" not in json.dumps(payload, ensure_ascii=False) assert updates == [ ( "remote-deposit", {"description": llmwiki_deposit.CONVERSATION_DEPOSIT_KB_DESCRIPTION}, ) ] def test_conversation_deposit_history_hides_legacy_brand() -> None: from app.gateway.routers import llmwiki as llmwiki_router value = { "content": "This page was generated by DeerFlow.", "metadata": {"source": "deer-flow", "tags": ["DEER_FLOW", "keep"]}, } gateway_value = llmwiki_router._sanitize_conversation_deposit_value(value) tool_value = sanitize_tool_deposit_value(value) assert gateway_value == tool_value assert gateway_value == { "content": "This page was generated by cmzs.", "metadata": {"source": "cmzs", "tags": ["cmzs", "keep"]}, } @pytest.mark.asyncio async def test_agent_bound_knowledge_base_list_uses_local_mapping_only(monkeypatch: pytest.MonkeyPatch) -> None: from types import SimpleNamespace from app.gateway.routers import llmwiki as llmwiki_router from deerflow.persistence.agents import MemoryAgentStore from deerflow.persistence.llmwiki import MemoryLlmWikiStore store = MemoryLlmWikiStore() mapping = await store.create_mapping( weknora_id="remote-agent-kb", owner_user_id="user-a", name="本地目录缓存", description="", kb_type="document", ) agent_store = MemoryAgentStore() await agent_store.ensure_builtin({"id": "agent-a", "name": "Agent A", "published": True}) async def actor(_: object) -> tuple[str, bool]: return "user-a", False monkeypatch.setattr(llmwiki_router, "_actor", actor) monkeypatch.setattr(llmwiki_router, "_store", lambda _: store) monkeypatch.setattr(llmwiki_router, "get_agent_store", lambda _: agent_store) monkeypatch.setattr( llmwiki_router, "load_agent_config", lambda _agent_id: SimpleNamespace(llmwiki_knowledge_base_ids=[mapping["id"]]), ) monkeypatch.setattr( llmwiki_router, "_client_or_503", lambda: (_ for _ in ()).throw(AssertionError("normal listing must not call WeKnora")), ) payload = await llmwiki_router.list_agent_knowledge_bases(object(), "agent-a") assert [item["id"] for item in payload["knowledge_bases"]] == [mapping["id"]] assert payload["knowledge_bases"][0]["remote_status"] == "cached" @pytest.mark.asyncio async def test_list_knowledge_bases_personal_scope_filters_to_actor_for_admin(monkeypatch: pytest.MonkeyPatch) -> None: from app.gateway.routers import llmwiki as llmwiki_router calls: list[dict[str, object]] = [] class Store: async def list_visible(self, user_id: str, *, scope: str, is_admin: bool) -> list[dict[str, object]]: calls.append({"user_id": user_id, "scope": scope, "is_admin": is_admin}) return [ { "id": "local-admin", "weknora_id": "remote-admin", "owner_user_id": user_id, "name": "Admin private", "description": "", "kb_type": "document", "publication_status": "private", } ] class Client: async def list_knowledge_bases(self, *, refresh: bool = False) -> list[dict[str, object]]: return [{"id": "remote-admin", "name": "Admin private"}] async def actor(_: object) -> tuple[str, bool]: return "admin-id", True monkeypatch.setattr(llmwiki_router, "_actor", actor) monkeypatch.setattr(llmwiki_router, "_store", lambda _: Store()) monkeypatch.setattr(llmwiki_router, "_client_or_503", lambda: Client()) payload = await llmwiki_router.list_knowledge_bases(object(), scope="personal") assert calls == [{"user_id": "admin-id", "scope": "personal", "is_admin": False}] assert payload["knowledge_bases"][0]["is_owner"] is True def test_mapping_view_distinguishes_owner_from_admin_write_power() -> None: from app.gateway.routers import llmwiki as llmwiki_router view = llmwiki_router._mapping_view( { "id": "local-other", "weknora_id": "remote-other", "owner_user_id": "user-b", "name": "Other private", "description": "", "kb_type": "document", "publication_status": "private", }, {"id": "remote-other", "name": "Other private"}, actor_user_id="admin-id", is_admin=True, ) assert view["is_owner"] is False assert view["can_write"] is True def test_only_system_owned_named_base_is_treated_as_conversation_deposit() -> None: from app.gateway.routers import llmwiki as llmwiki_router personal = { "id": "personal-named-deposit", "owner_user_id": "user-a", "name": "对话沉淀", "publication_status": "private", } system = { "id": "system-deposit", "owner_user_id": "system", "name": "对话沉淀", "publication_status": "published", } assert not llmwiki_router._is_conversation_deposit_mapping(personal) assert llmwiki_router._is_conversation_deposit_mapping(system) system_view = llmwiki_router._mapping_view( system, { "id": "remote-system", "name": "对话沉淀", "description": "DeerFlow legacy description", }, actor_user_id="admin-id", is_admin=True, ) assert system_view["can_write"] is False assert system_view["description"] == "cmzs 全局问答沉淀知识库。所有用户可见,不默认参与问答检索。" assert "DeerFlow" not in system_view["description"] @pytest.mark.asyncio async def test_conversation_deposit_requires_thread_access(monkeypatch: pytest.MonkeyPatch) -> None: from types import SimpleNamespace from app.gateway.routers import llmwiki as llmwiki_router class ThreadStore: async def check_access(self, thread_id: str, user_id: str, *, require_existing: bool) -> bool: assert thread_id == "thread-other" assert user_id == "user-a" assert require_existing is True return False async def actor(_: object) -> tuple[str, bool]: return "user-a", False monkeypatch.setattr(llmwiki_router, "_actor", actor) monkeypatch.setattr(llmwiki_router, "get_thread_store", lambda _: ThreadStore()) monkeypatch.setattr( llmwiki_router, "get_resolved_llmwiki_runtime", lambda _: SimpleNamespace(weknora_enabled=True, provider="weknora"), ) request = SimpleNamespace(app=SimpleNamespace(state=SimpleNamespace(config=object()))) body = llmwiki_router.ConversationDepositCreate( thread_id="thread-other", question="question", answer="answer", assistant_message_id="assistant-one", ) with pytest.raises(HTTPException) as exc: await llmwiki_router.create_conversation_deposit( request, SimpleNamespace(add_task=lambda *_: None), body, ) assert exc.value.status_code == 404 @pytest.mark.asyncio async def test_source_context_prefers_wiki_page_over_raw_chunks(monkeypatch: pytest.MonkeyPatch) -> None: from app.gateway.routers import llmwiki as llmwiki_router class Client: async def get_chunk_context(self, chunk_id: str, *, expected_knowledge_base_id: str, radius: int) -> list[dict[str, object]]: assert chunk_id == "chunk-one" assert expected_knowledge_base_id == "remote-kb" return [ { "id": "chunk-one", "chunk_index": 0, "content": "raw split chunk", "metadata": {"wiki_page_slug": "people/alice"}, } ] async def get_wiki_page(self, knowledge_base_id: str, slug: str) -> dict[str, object]: assert knowledge_base_id == "remote-kb" assert slug == "people/alice" return { "id": "wiki-one", "slug": "people/alice", "title": "Alice", "summary": "clean summary", "content": "clean llmwiki page", "source_refs": ["chunk-one"], } async def authorized_mapping(*_: object, **__: object) -> tuple[dict[str, str], str, bool]: return {"id": "local-kb", "weknora_id": "remote-kb", "name": "Public LLMWiki"}, "user-a", False monkeypatch.setattr(llmwiki_router, "_authorized_mapping", authorized_mapping) monkeypatch.setattr(llmwiki_router, "_client_or_503", lambda: Client()) payload = await llmwiki_router.get_source_context(object(), knowledge_base_id="local-kb", chunk_id="chunk-one") assert payload["data"]["display_mode"] == "wiki" assert payload["data"]["wiki_page"]["content"] == "clean llmwiki page" assert payload["data"]["chunks"][0]["content"] == "raw split chunk" @pytest.mark.asyncio async def test_delete_cleans_owner_mapping_when_remote_is_already_missing(monkeypatch: pytest.MonkeyPatch) -> None: from app.gateway.routers import llmwiki as llmwiki_router deleted: list[str] = [] class MissingClient: async def delete_knowledge_base(self, _: str) -> None: raise WeKnoraError("WeKnora resource was not found", status_code=404) class Store: async def delete_mapping(self, mapping_id: str) -> None: deleted.append(mapping_id) async def authorized_mapping(*_: object, **__: object) -> tuple[dict[str, str], str, bool]: return {"weknora_id": "remote-missing"}, "owner", False monkeypatch.setattr(llmwiki_router, "_authorized_mapping", authorized_mapping) monkeypatch.setattr(llmwiki_router, "_client_or_503", lambda: MissingClient()) monkeypatch.setattr(llmwiki_router, "_store", lambda _: Store()) await llmwiki_router.delete_knowledge_base(object(), "mapping-one") assert deleted == ["mapping-one"] def test_weknora_iframe_cookie_is_signed_and_short_lived() -> None: from app.gateway.routers import llmwiki as llmwiki_router raw = llmwiki_router._sign_embed_payload( { "mapping_id": "local-kb", "weknora_id": "remote-kb", "user_id": "user-a", "is_admin": False, "can_write": True, "exp": 4_102_444_800, } ) ctx = llmwiki_router._verify_embed_cookie(raw) assert ctx.mapping_id == "local-kb" assert ctx.weknora_id == "remote-kb" assert ctx.can_write is True with pytest.raises(HTTPException): llmwiki_router._verify_embed_cookie(raw + "tampered") def test_weknora_iframe_proxy_is_fallback_after_deerflow_api_routes() -> None: assert _first_gateway_match("POST", "/api/v1/auth/login/username") == "/api/v1/auth/login/username" assert _first_gateway_match("POST", "/api/v1/knowledge-search") == "/api/v1/{proxied_path:path}" assert _first_gateway_match("GET", "/api/llmwiki/weknora-embed/api/v1/auth/me") == "/api/llmwiki/weknora-embed/api/v1/{proxied_path:path}" assert _first_gateway_match("GET", "/platform/knowledge-bases/remote-kb") == "/platform/{proxied_path:path}" assert _first_gateway_match("GET", "/deerflow/platform/knowledge-bases/remote-kb") == "/deerflow/platform/{proxied_path:path}" assert _first_gateway_match("GET", "/deerflow/assets/index.js") == "/deerflow/assets/{proxied_path:path}" assert _first_gateway_match("GET", "/deerflow/js/index.js") == "/deerflow/{proxied_path:path}" assert _first_gateway_match("GET", "/deerflow/css/index.css") == "/deerflow/{proxied_path:path}" assert _first_gateway_match("GET", "/deerflow/fonts/index.woff2") == "/deerflow/{proxied_path:path}" assert _first_gateway_match("GET", "/deerflow/api/v1/knowledge-search") == "/deerflow/api/v1/{proxied_path:path}" assert _first_gateway_match("GET", "/js/index.js") == "/{proxied_path:path}" @pytest.mark.parametrize( ("location", "target_base_url", "expected"), [ ("/platform/knowledge-bases/kb?tab=wiki", "http://weknora-web:8080", "/deerflow/platform/knowledge-bases/kb?tab=wiki"), ("/deerflow/platform/knowledge-bases/kb", "http://weknora-web:8080", "/deerflow/platform/knowledge-bases/kb"), ("/deerflow-api/api/v1/auth/me", "http://gateway/deerflow-api", "/deerflow/api/v1/auth/me"), ( "http://gateway/deerflow-api/api/v1/auth/me?next=1#top", "http://gateway/deerflow-api", "/deerflow/api/v1/auth/me?next=1#top", ), ("https://external.test/login", "http://gateway/deerflow-api", "https://external.test/login"), ], ) def test_weknora_redirect_location_is_normalized_for_public_prefix( location: str, target_base_url: str, expected: str, ) -> None: from app.gateway.routers import llmwiki as llmwiki_router assert llmwiki_router._rewrite_weknora_location_header(location, "/deerflow", target_base_url) == expected @pytest.mark.asyncio async def test_prefixed_weknora_proxy_serves_page_static_assets_and_api_without_auth_errors( monkeypatch: pytest.MonkeyPatch, ) -> None: from app.gateway.auth_middleware import AuthMiddleware from app.gateway.csrf_middleware import CSRFMiddleware from app.gateway.routers import llmwiki as llmwiki_router from app.gateway.weknora_embed import WEKNORA_EMBED_COOKIE, sign_weknora_embed_payload class Store: async def get_authorized(self, *_: object, **__: object) -> dict[str, str]: return {"id": "local-kb", "weknora_id": "remote-kb"} async def list_visible(self, *_: object, **__: object) -> list[dict[str, str]]: return [{"id": "local-kb", "weknora_id": "remote-kb"}] runtime = SimpleNamespace( web_base_url="http://weknora-web.test", api_base_url="http://weknora-api.test", ) upstream_requests: list[tuple[str, str, dict[str, str]]] = [] async def fake_session(_: object, *, force_refresh: bool = False) -> dict[str, object]: return {"token": "admin-token", "tenant": {"id": "tenant-1"}} class FakeUpstreamClient: def __init__(self, **_: object) -> None: pass async def __aenter__(self) -> FakeUpstreamClient: return self async def __aexit__(self, *_: object) -> None: return None async def request( self, method: str, url: str, *, headers: dict[str, str], **__: object, ) -> httpx.Response: upstream_requests.append((method, url, headers)) path = httpx.URL(url).path if url.startswith(runtime.api_base_url): return httpx.Response(200, json={"success": True, "data": {"id": "admin"}}) if path == "/platform/knowledge-bases/remote-kb": return httpx.Response( 200, text=(''), headers={"content-type": "text/html; charset=utf-8"}, ) if path == "/assets/main.js": return httpx.Response( 200, text=('const __vite__mapDeps=(i,m=__vite__mapDeps,d=(m.f||(m.f=["assets/lazy.css","assets/lazy.js"])))=>i.map(i=>d[i]);const chunk="/static/chunk.js",matcher=/platform\\//;fetch("/api/v1/auth/me")'), headers={"content-type": "application/javascript"}, ) if path == "/css/app.css": return httpx.Response( 200, text='@font-face{src:url("/fonts/app.woff2")}', headers={"content-type": "text/css"}, ) if path in {"/config.js", "/runtime-config.js", "/static/chunk.js", "/assets/deep.js"}: return httpx.Response(200, text=f'window.loaded="{path}"', headers={"content-type": "application/javascript"}) if path == "/fonts/app.woff2": return httpx.Response(200, content=b"font", headers={"content-type": "font/woff2"}) return httpx.Response(404, text=f"missing upstream path: {path}") monkeypatch.setattr(llmwiki_router, "_runtime_for_iframe_proxy", lambda _: runtime) monkeypatch.setattr(llmwiki_router, "_store", lambda _: Store()) monkeypatch.setattr(llmwiki_router, "_get_weknora_admin_session", fake_session) real_async_client = httpx.AsyncClient monkeypatch.setattr(llmwiki_router.httpx, "AsyncClient", FakeUpstreamClient) test_app = FastAPI() test_app.add_middleware(AuthMiddleware) test_app.add_middleware(CSRFMiddleware) test_app.include_router(llmwiki_router.router) test_app.include_router(llmwiki_router.proxy_router, prefix="/deerflow") test_app.include_router(llmwiki_router.proxy_router) cookie = sign_weknora_embed_payload( { "mapping_id": "local-kb", "weknora_id": "remote-kb", "user_id": "user-a", "is_admin": False, "can_write": True, "exp": 4_102_444_800, } ) transport = httpx.ASGITransport(app=test_app) async with real_async_client(transport=transport, base_url="http://deerflow.test") as client: client.cookies.set(WEKNORA_EMBED_COOKIE, cookie) page = await client.get("/deerflow/platform/knowledge-bases/remote-kb?tab=wiki") assert page.status_code == 200 assert 'src="/deerflow/assets/main.js"' in page.text assert 'href="/deerflow/css/app.css"' in page.text assert 'src="/deerflow/config.js"' in page.text assert 'src="/deerflow/runtime-config.js"' in page.text # The release Nginx strips /deerflow-api before forwarding and tells # the Gateway which public prefix to put back into iframe asset URLs. release_page = await client.get( "/platform/knowledge-bases/remote-kb?tab=wiki", headers={"X-Forwarded-Prefix": "/deerflow-api"}, ) assert release_page.status_code == 200 assert 'src="/deerflow-api/assets/main.js"' in release_page.text assert 'href="/deerflow-api/css/app.css"' in release_page.text assert '"apiPrefix":"/deerflow-api/api/llmwiki/weknora-embed"' in release_page.text script = await client.get("/deerflow/assets/main.js") assert script.status_code == 200 assert '"/deerflow/static/chunk.js"' in script.text assert '"deerflow/assets/lazy.css"' in script.text assert '"deerflow/assets/lazy.js"' in script.text assert '"/deerflow/assets/lazy.css"' not in script.text assert "matcher=/platform\\//" in script.text assert 'fetch("/deerflow/api/v1/auth/me")' in script.text stylesheet = await client.get("/deerflow/css/app.css") assert stylesheet.status_code == 200 assert 'url("/deerflow/fonts/app.woff2")' in stylesheet.text for path in ( "/deerflow/config.js", "/deerflow/runtime-config.js", "/deerflow/platform/knowledge-bases/runtime-config.js", "/deerflow/static/chunk.js", "/deerflow/fonts/app.woff2", "/deerflow/platform/knowledge-bases/remote-kb/assets/deep.js", ): response = await client.get(path) assert response.status_code == 200, f"{path}: {response.status_code} {response.text}" api = await client.get("/deerflow/api/llmwiki/weknora-embed/api/v1/auth/me") assert api.status_code == 200 directory = await client.get("/deerflow/api/llmwiki/weknora-embed/api/v1/knowledge-bases") assert directory.status_code == 200 assert directory.json() == { "success": True, "data": [ { "id": "remote-kb", "name": "", "description": "", "type": "document", "status": "active", "created_at": None, "updated_at": None, } ], "total": 1, } api_request = next(request for request in upstream_requests if request[1].startswith(runtime.api_base_url)) assert api_request[2]["Authorization"] == "Bearer admin-token" assert api_request[2]["X-Tenant-ID"] == "tenant-1" assert not any(request[1].endswith("/api/v1/knowledge-bases") for request in upstream_requests) def test_weknora_iframe_hides_current_outer_sidebar_shell() -> None: from app.gateway.routers import llmwiki as llmwiki_router ctx = llmwiki_router.WeKnoraEmbedContext( mapping_id="local-kb", weknora_id="remote-kb", user_id="user-a", is_admin=False, can_write=False, exp=4_102_444_800, allowed_weknora_ids=("remote-kb", "other-visible-kb"), ) injection = llmwiki_router._weknora_embed_injection( allowed_path="/platform/knowledge-bases/remote-kb", tab="wiki", session={}, ctx=ctx, ) assert "#app > .main > .aside_box" in injection assert '".main > .aside_box"' in injection def test_weknora_iframe_html_rewrites_public_prefix_paths() -> None: from app.gateway.routers import llmwiki as llmwiki_router ctx = llmwiki_router.WeKnoraEmbedContext( mapping_id="local-kb", weknora_id="remote-kb", user_id="user-a", is_admin=False, can_write=False, exp=4_102_444_800, allowed_weknora_ids=("remote-kb", "other-visible-kb"), ) html = b'fetch("/api/v1/auth/me")' injected = llmwiki_router._inject_weknora_embed_html( html, allowed_path="/deerflow/platform/knowledge-bases/remote-kb", tab="wiki", session={}, ctx=ctx, public_prefix="/deerflow", ).decode() assert 'src="/deerflow/assets/index.js"' in injected assert 'fetch("/deerflow/api/v1/auth/me")' in injected assert '"apiPrefix":"/deerflow/api/llmwiki/weknora-embed"' in injected assert '"allowedPath":"/deerflow/platform/knowledge-bases/remote-kb"' in injected assert '"allowedWeKnoraIds":["remote-kb","other-visible-kb"]' in injected assert ".breadcrumb-link.dropdown" not in injected assert "reloadForKnowledgeBaseSwitch" in injected assert "function protectedImageApiUrl" in injected assert "img[data-protected-src]" in injected assert "encodeURIComponent(cfg.weknoraId)" in injected assert "image.src = proxyUrl;" in injected assert "URL.createObjectURL(blob)" not in injected assert 'attributeFilter: ["data-protected-src", "src"]' in injected def test_weknora_iframe_filters_kb_list_to_signed_visible_mappings() -> None: from app.gateway.routers import llmwiki as llmwiki_router ctx = llmwiki_router.WeKnoraEmbedContext( mapping_id="local-kb", weknora_id="remote-kb", user_id="user-a", is_admin=False, can_write=False, exp=4_102_444_800, allowed_weknora_ids=("remote-kb", "other-visible-kb"), ) response = httpx.Response( 200, json={ "success": True, "data": [ {"id": "remote-kb", "name": "Allowed"}, {"id": "other-visible-kb", "name": "Also allowed"}, {"id": "other-kb", "name": "Blocked"}, ], }, ) filtered = llmwiki_router._filter_knowledge_base_list_response(response, ctx) assert filtered is not None payload = json.loads(filtered) assert payload["data"] == [ {"id": "remote-kb", "name": "Allowed"}, {"id": "other-visible-kb", "name": "Also allowed"}, ] @pytest.mark.asyncio async def test_weknora_iframe_switch_rechecks_selected_mapping_permissions( monkeypatch: pytest.MonkeyPatch, ) -> None: from app.gateway.routers import llmwiki as llmwiki_router class Store: async def get_by_weknora_id(self, weknora_id: str) -> dict[str, str] | None: if weknora_id == "other-visible-kb": return {"id": "public-kb", "weknora_id": weknora_id, "owner_user_id": "owner-b"} return None async def get_authorized(self, mapping_id: str, _: str, *, write: bool, **__: object) -> dict[str, str] | None: if mapping_id != "public-kb": return None if write: return None return {"id": "public-kb", "weknora_id": "other-visible-kb", "owner_user_id": "owner-b"} monkeypatch.setattr(llmwiki_router, "_store", lambda _: Store()) original = llmwiki_router.WeKnoraEmbedContext( mapping_id="local-kb", weknora_id="remote-kb", user_id="user-a", is_admin=False, can_write=True, exp=4_102_444_800, ) selected = await llmwiki_router._switch_embed_context( _request("/platform/knowledge-bases/other-visible-kb"), original, "other-visible-kb", ) assert selected.mapping_id == "public-kb" assert selected.weknora_id == "other-visible-kb" assert selected.can_write is False @pytest.mark.asyncio async def test_weknora_platform_proxy_reissues_cookie_for_native_kb_switch( monkeypatch: pytest.MonkeyPatch, ) -> None: from app.gateway.routers import llmwiki as llmwiki_router current = llmwiki_router.WeKnoraEmbedContext( mapping_id="local-kb", weknora_id="remote-kb", user_id="user-a", is_admin=False, can_write=False, exp=4_102_444_800, allowed_weknora_ids=("remote-kb", "other-visible-kb"), ) selected = llmwiki_router.WeKnoraEmbedContext( mapping_id="public-kb", weknora_id="other-visible-kb", user_id="user-a", is_admin=False, can_write=False, exp=4_102_444_800, allowed_weknora_ids=current.allowed_weknora_ids, ) seen: dict[str, object] = {} async def read_context(_: Request) -> object: return current async def switch_context(_: Request, ctx: object, weknora_id: str) -> object: seen["current"] = ctx seen["target"] = weknora_id return selected async def hydrate_visible_bases(_: Request, ctx: object) -> object: return ctx async def proxy_request(*_: object, **kwargs: object) -> Response: seen.update(kwargs) return Response("ok") monkeypatch.setattr(llmwiki_router, "_read_embed_context", read_context) monkeypatch.setattr(llmwiki_router, "_switch_embed_context", switch_context) monkeypatch.setattr(llmwiki_router, "_with_visible_embed_knowledge_bases", hydrate_visible_bases) monkeypatch.setattr( llmwiki_router, "_runtime_for_iframe_proxy", lambda _: SimpleNamespace(web_base_url="http://weknora-web"), ) monkeypatch.setattr(llmwiki_router, "_proxy_weknora_request", proxy_request) response = await llmwiki_router.proxy_weknora_platform_page( _request("/platform/knowledge-bases/other-visible-kb", request_headers={"X-Forwarded-Proto": "https"}), "knowledge-bases/other-visible-kb", ) assert seen["current"] == current assert seen["target"] == "other-visible-kb" assert seen["upstream_path"] == "/platform/knowledge-bases/other-visible-kb" raw_cookie = response.headers["set-cookie"].split("deerflow_weknora_embed=", 1)[1].split(";", 1)[0] switched_context = llmwiki_router._verify_embed_cookie(raw_cookie) assert switched_context.mapping_id == "public-kb" assert switched_context.weknora_id == "other-visible-kb" @pytest.mark.asyncio async def test_conversation_deposit_iframe_sanitizes_legacy_brand(monkeypatch: pytest.MonkeyPatch) -> None: from types import SimpleNamespace from app.gateway.routers import llmwiki as llmwiki_router async def fake_session(_: object, *, force_refresh: bool = False) -> dict[str, object]: return {"token": "token", "tenant": {}} async def authorize(*_: object) -> None: return None class FakeAsyncClient: def __init__(self, **_: object) -> None: pass async def __aenter__(self) -> FakeAsyncClient: return self async def __aexit__(self, *_: object) -> None: return None async def request(self, *_: object, **__: object) -> httpx.Response: return httpx.Response( 200, json={"success": True, "data": {"summary": "Generated by DeerFlow", "source": "deer-flow"}}, ) class FakeRequest: method = "GET" headers = Headers({}) query_params = QueryParams("") async def body(self) -> bytes: return b"" ctx = llmwiki_router.WeKnoraEmbedContext( mapping_id="local-deposit", weknora_id="remote-deposit", user_id="user-a", is_admin=False, can_write=False, exp=4_102_444_800, is_conversation_deposit=True, ) monkeypatch.setattr(llmwiki_router, "_runtime_for_iframe_proxy", lambda _: SimpleNamespace()) monkeypatch.setattr(llmwiki_router, "_get_weknora_admin_session", fake_session) monkeypatch.setattr(llmwiki_router, "_authorize_weknora_api_proxy", authorize) monkeypatch.setattr(llmwiki_router.httpx, "AsyncClient", FakeAsyncClient) response = await llmwiki_router._proxy_weknora_request( FakeRequest(), target_base_url="http://weknora.local", upstream_path="/api/v1/knowledge-bases/remote-deposit", ctx=ctx, is_api=True, ) payload = json.loads(response.body) assert payload["data"] == {"summary": "Generated by cmzs", "source": "cmzs"} @pytest.mark.asyncio @pytest.mark.parametrize("filename", ["wiki-vector.jsonl", "renamed-package.txt"]) async def test_compatible_package_is_not_uploaded_as_an_ordinary_document( monkeypatch: pytest.MonkeyPatch, filename: str, ) -> None: from app.gateway.routers import llmwiki as llmwiki_router forwarded: list[dict[str, object]] = [] syncs: list[dict[str, object]] = [] class Client: async def upload_document(self, knowledge_base_id: str, **kwargs: object) -> dict[str, object]: forwarded.append({"knowledge_base_id": knowledge_base_id, **kwargs}) return {"id": "unexpected-document"} async def authorized_mapping(*_: object, **__: object) -> tuple[dict[str, object], str, bool]: return {"id": "local-kb", "weknora_id": "remote-kb"}, "admin", True app = FastAPI() request = Request( { "type": "http", "method": "POST", "path": "/api/llmwiki/knowledge-bases/local-kb/documents", "headers": [], "app": app, } ) upload = UploadFile( filename=filename, file=BytesIO(_compatible_knowledge_package_bytes(bom=True)), headers=Headers({"content-type": "text/plain"}), ) monkeypatch.setattr(llmwiki_router, "_authorized_mapping", authorized_mapping) monkeypatch.setattr(llmwiki_router, "_client_or_503", lambda: Client()) monkeypatch.setattr(llmwiki_router, "_schedule_local_wiki_sync", lambda _request, row: syncs.append(row)) with pytest.raises(HTTPException) as exc_info: await llmwiki_router.upload_document(request, "local-kb", upload) assert exc_info.value.status_code == 422 assert "本次文件未送入模型" in str(exc_info.value.detail) assert forwarded == [] assert syncs == [] @pytest.mark.asyncio async def test_ordinary_jsonl_is_still_uploaded_as_an_ordinary_document(monkeypatch: pytest.MonkeyPatch) -> None: from app.gateway.routers import llmwiki as llmwiki_router content = '{"question":"如何散热?","answer":"保持风道畅通。"}\n'.encode() forwarded: list[dict[str, object]] = [] syncs: list[dict[str, object]] = [] class Client: async def upload_document(self, knowledge_base_id: str, **kwargs: object) -> dict[str, object]: forwarded.append({"knowledge_base_id": knowledge_base_id, **kwargs}) return {"id": "document-one"} async def authorized_mapping(*_: object, **__: object) -> tuple[dict[str, object], str, bool]: return {"id": "local-kb", "weknora_id": "remote-kb"}, "admin", True app = FastAPI() request = Request( { "type": "http", "method": "POST", "path": "/api/llmwiki/knowledge-bases/local-kb/documents", "headers": [], "app": app, } ) upload = UploadFile( filename="faq.jsonl", file=BytesIO(content), headers=Headers({"content-type": "application/x-ndjson"}), ) monkeypatch.setattr(llmwiki_router, "_authorized_mapping", authorized_mapping) monkeypatch.setattr(llmwiki_router, "_client_or_503", lambda: Client()) monkeypatch.setattr(llmwiki_router, "_schedule_local_wiki_sync", lambda _request, row: syncs.append(row)) result = await llmwiki_router.upload_document(request, "local-kb", upload) assert result == {"id": "document-one"} assert forwarded == [ { "knowledge_base_id": "remote-kb", "filename": "faq.jsonl", "content": content, "content_type": "application/x-ndjson", } ] assert syncs == [{"id": "local-kb", "weknora_id": "remote-kb"}] @pytest.mark.asyncio async def test_weknora_iframe_proxy_rejects_compatible_package_document_upload() -> None: from app.gateway.routers import llmwiki as llmwiki_router boundary = "deerflow-test-boundary" body = ( (f'--{boundary}\r\nContent-Disposition: form-data; name="file"; filename="wiki-vector.jsonl"\r\nContent-Type: application/x-ndjson\r\n\r\n').encode() + _compatible_knowledge_package_bytes(bom=True) + f"\r\n--{boundary}--\r\n".encode() ) class FakeRequest: method = "POST" headers = Headers({"content-type": f"multipart/form-data; boundary={boundary}"}) query_params = QueryParams("") ctx = llmwiki_router.WeKnoraEmbedContext( mapping_id="local-kb", weknora_id="remote-kb", user_id="admin-a", is_admin=True, can_write=True, exp=4_102_444_800, ) with pytest.raises(HTTPException) as exc_info: await llmwiki_router._authorize_weknora_api_proxy( FakeRequest(), ctx, object(), "/api/v1/knowledge-bases/remote-kb/knowledge/file", body, "token", ) assert exc_info.value.status_code == 422 assert "本次文件未送入模型" in str(exc_info.value.detail) @pytest.mark.asyncio async def test_weknora_iframe_proxy_blocks_other_knowledge_bases() -> None: from app.gateway.routers import llmwiki as llmwiki_router class FakeRequest: method = "GET" headers = Headers({}) query_params = QueryParams("") ctx = llmwiki_router.WeKnoraEmbedContext( mapping_id="local-kb", weknora_id="remote-kb", user_id="user-a", is_admin=False, can_write=False, exp=4_102_444_800, ) await llmwiki_router._authorize_weknora_api_proxy( FakeRequest(), ctx, object(), "/api/v1/knowledge-bases/remote-kb", b"", "token", ) await llmwiki_router._authorize_weknora_api_proxy( FakeRequest(), ctx, object(), "/api/v1/me/invitations/pending-count", b"", "token", ) with pytest.raises(HTTPException) as exc_info: await llmwiki_router._authorize_weknora_api_proxy( FakeRequest(), ctx, object(), "/api/v1/knowledge-bases/other-kb", b"", "token", ) assert exc_info.value.status_code == 403 @pytest.mark.asyncio async def test_weknora_iframe_proxy_only_reads_tenant_retrieval_config() -> None: from app.gateway.routers import llmwiki as llmwiki_router class FakeRequest: headers = Headers({}) query_params = QueryParams("") def __init__(self, method: str) -> None: self.method = method ctx = llmwiki_router.WeKnoraEmbedContext( mapping_id="local-kb", weknora_id="remote-kb", user_id="admin-a", is_admin=True, can_write=True, exp=4_102_444_800, ) await llmwiki_router._authorize_weknora_api_proxy( FakeRequest("GET"), ctx, object(), "/api/v1/tenants/kv/retrieval-config", b"", "token", ) for method, path in ( ("GET", "/api/v1/tenants/kv/other-setting"), ("PUT", "/api/v1/tenants/kv/retrieval-config"), ): with pytest.raises(HTTPException) as exc_info: await llmwiki_router._authorize_weknora_api_proxy( FakeRequest(method), ctx, object(), path, b"{}", "token", ) assert exc_info.value.status_code == 403 @pytest.mark.asyncio async def test_incremental_sync_endpoint_queues_owner_authorized_non_force_job( monkeypatch: pytest.MonkeyPatch, ) -> None: from app.gateway.routers import llmwiki_index as index_router mapping = { "id": "local-kb", "weknora_id": "remote-kb", "name": "KB", "wiki_index_enabled": True, } authorization_calls: list[dict[str, object]] = [] sync_calls: list[dict[str, object]] = [] queued_coroutines: list[object] = [] class Store: async def get_authorized( self, mapping_id: str, user_id: str, *, write: bool, is_admin: bool, ) -> dict[str, object] | None: authorization_calls.append( { "mapping_id": mapping_id, "user_id": user_id, "write": write, "is_admin": is_admin, } ) return mapping class Service: embedding = SimpleNamespace(dimensions=3) async def sync_mapping( self, selected_mapping: dict[str, object], *, force: bool, sync_id: str | None = None, ) -> None: sync_calls.append( { "mapping": selected_mapping, "force": force, "sync_id": sync_id, } ) async def runtime_config(_: object) -> object: return SimpleNamespace(enabled=True) async def actor(_: Request) -> tuple[str, bool]: return "owner-a", False def queue(_: FastAPI, coroutine: object) -> None: queued_coroutines.append(coroutine) app = FastAPI() app.state.llmwiki_store = Store() app.state.llmwiki_sync_service = Service() request = _request("/api/llmwiki/knowledge-bases/local-kb/sync") request.scope["app"] = app monkeypatch.setattr(index_router, "rebuild_local_wiki_index_runtime", runtime_config) monkeypatch.setattr(index_router, "_actor", actor) monkeypatch.setattr(index_router, "spawn_local_wiki_index_job", queue) response = await index_router.sync_knowledge_base_incrementally(request, "local-kb") assert response["status"] == "accepted" assert response["mode"] == "incremental" assert response["knowledge_base_id"] == "local-kb" assert response["status_url"].endswith("/local-kb/wiki-index-status") assert authorization_calls == [ { "mapping_id": "local-kb", "user_id": "owner-a", "write": True, "is_admin": False, } ] assert len(queued_coroutines) == 1 await queued_coroutines.pop() assert sync_calls == [ { "mapping": mapping, "force": False, "sync_id": response["job_id"], } ] route = next( route for route in index_router.router.routes if route.path == "/api/llmwiki/knowledge-bases/{mapping_id}/sync" ) assert route.status_code == 202 @pytest.mark.asyncio async def test_incremental_sync_endpoint_rejects_before_encoder_probe( monkeypatch: pytest.MonkeyPatch, ) -> None: from app.gateway.routers import llmwiki_index as index_router class Store: async def get_authorized(self, *_: object, **__: object) -> None: return None class Embedding: dimensions = None async def embed_query(self, _: str) -> list[float]: raise AssertionError("unauthorized requests must not probe the encoder") async def actor(_: Request) -> tuple[str, bool]: return "other-user", False async def unexpected_runtime(_: object) -> object: raise AssertionError("authorization must precede runtime initialization") app = FastAPI() app.state.llmwiki_store = Store() app.state.llmwiki_sync_service = SimpleNamespace(embedding=Embedding()) request = _request("/api/llmwiki/knowledge-bases/private-kb/sync") request.scope["app"] = app monkeypatch.setattr(index_router, "_actor", actor) monkeypatch.setattr( index_router, "rebuild_local_wiki_index_runtime", unexpected_runtime, ) with pytest.raises(HTTPException) as exc_info: await index_router.sync_knowledge_base_incrementally(request, "private-kb") assert exc_info.value.status_code == 404 @pytest.mark.asyncio async def test_proxy_sync_scheduler_uses_the_authorized_local_mapping( monkeypatch: pytest.MonkeyPatch, ) -> None: from app.gateway.routers import llmwiki as llmwiki_router mapping = {"id": "local-kb", "weknora_id": "remote-kb", "name": "KB"} scheduled: list[dict[str, object]] = [] class Store: async def get_authorized( self, mapping_id: str, user_id: str, *, write: bool, is_admin: bool, ) -> dict[str, object] | None: assert (mapping_id, user_id, write, is_admin) == ( "local-kb", "owner-a", False, False, ) return mapping monkeypatch.setattr(llmwiki_router, "_store", lambda _: Store()) monkeypatch.setattr( llmwiki_router, "_schedule_local_wiki_sync", lambda _request_value, value: scheduled.append(value), ) context = llmwiki_router.WeKnoraEmbedContext( mapping_id="local-kb", weknora_id="remote-kb", user_id="owner-a", is_admin=False, can_write=True, exp=4_102_444_800, ) await llmwiki_router._schedule_proxy_local_wiki_sync( _request("/api/v1/knowledge-bases/remote-kb"), context, ) assert scheduled == [mapping] @pytest.mark.asyncio @pytest.mark.parametrize( ("method", "upstream_status", "expected_schedules"), [("POST", 201, 1), ("POST", 422, 0), ("GET", 200, 0)], ) async def test_proxy_only_schedules_after_successful_content_mutation( monkeypatch: pytest.MonkeyPatch, method: str, upstream_status: int, expected_schedules: int, ) -> None: from app.gateway.routers import llmwiki as llmwiki_router scheduled: list[object] = [] class FakeRequest: headers = Headers({}) query_params = QueryParams("") def __init__(self) -> None: self.method = method async def body(self) -> bytes: return b"{}" class FakeAsyncClient: def __init__(self, **_: object) -> None: pass async def __aenter__(self) -> FakeAsyncClient: return self async def __aexit__(self, *_: object) -> None: return None async def request(self, *_: object, **__: object) -> httpx.Response: return httpx.Response(upstream_status, json={"ok": upstream_status < 400}) async def session(*_: object, **__: object) -> dict[str, object]: return {"token": "token", "tenant": {}} async def authorize(*_: object, **__: object) -> None: return None async def schedule(*args: object) -> None: scheduled.append(args) context = llmwiki_router.WeKnoraEmbedContext( mapping_id="local-kb", weknora_id="remote-kb", user_id="owner-a", is_admin=False, can_write=True, exp=4_102_444_800, ) monkeypatch.setattr(llmwiki_router, "_runtime_for_iframe_proxy", lambda _: object()) monkeypatch.setattr(llmwiki_router, "_get_weknora_admin_session", session) monkeypatch.setattr(llmwiki_router, "_authorize_weknora_api_proxy", authorize) monkeypatch.setattr(llmwiki_router, "_schedule_proxy_local_wiki_sync", schedule) monkeypatch.setattr(llmwiki_router.httpx, "AsyncClient", FakeAsyncClient) response = await llmwiki_router._proxy_weknora_request( FakeRequest(), target_base_url="http://weknora.local", upstream_path="/api/v1/knowledge-bases/remote-kb/wiki-pages", ctx=context, is_api=True, ) assert response.status_code == upstream_status assert len(scheduled) == expected_schedules